create-connections

创建连接

用途

在 connections/<name>.yaml 下编写一个 Connection YAML，将 Source 中选定的流摄取到组织的数据仓库中。每个连接都是一份完整、独立的文档——revos apply 会读取它并与 API 中的配置进行协调。

难点在于流的选择和同步模式的选择，而非 YAML 的格式。大多数源提供的流远多于项目实际所需；数据库尤为如此。请谨慎选择、与用户确认，然后再编写。

---

前置条件

• 服务器上已存在该 Source（你可以通过 revos sources list 获取其 id）。如果用户想从尚未创建的源摄取数据，请先停止并告知他们运行 revos sources create——源配置（连接器选择器、凭据、OAuth）需在 RevOS UI 中完成。用户确认已从 UI 返回后，继续操作前请重新运行 revos connections list --json 和 revos sources list --json——UI 可能已为新源自动创建连接。如果已存在匹配的连接，请建议用户运行 revos pull 拉取，而不要编写重复配置。
• revos auth status 显示已完成身份验证。如果没有，请让用户运行 revos auth login。

---

用户确认点

此技能涉及两个适合明确确认的决策。不要跳过——流或同步模式选错会浪费数据仓库存储和重新加载时间。

确认点 1：选择流

发现流并询问用户的使用场景后，提出一个子集，并为每个流用一句话说明理由。等待用户确认或修改后再继续。

确认点 2：同步模式和键的选择

为每个选定的流推导出同步模式、游标和主键后，展示表格并在写入文件前确认。大多数用户会快速浏览并批准；有些用户可能会希望将某个流改为 full_refresh_overwrite，或更改游标。

---

工作流

阶段 1：识别 Source

首先运行 revos connections list --json，查看已有连接。如果用户刚从 Web UI 返回，重新获取数据至关重要——UI 可能已自动创建连接。如果发现符合用户所述目标的连接，请先告知用户，再继续操作，以免创建重复连接。

然后运行 revos sources list --json，并根据服务器端的 name 进行匹配（也可以运行 revos sources list 查看便于浏览的表格）。记录 Source 的 id——它需要写入 Connection YAML 的 spec.source.id 中。

如果用户没有指定 Source，请询问他们想用哪一个。没有 id 不要继续。

阶段 2：发现流

revos sources list-streams <id> --json

这会返回一个对象数组，结构如下：

{
  "streamName": "customers",
  "streamnamespace": "public",
  "syncModes": ["full_refresh_overwrite", "incremental_deduped_history", ...],
  "defaultCursorField": ["updated_at"],
  "sourceDefinedCursorField": false,
  "sourceDefinedPrimaryKey": [["id"]],
  "propertyFields": [["id"], ["email"], ["updated_at"], ...]
}

保存完整响应——后续阶段会用到。syncModes 数组限定了该流在项目目标端支持的有效模式；建议的模式必须包含在此列表中。

需要了解该数据结构的两点：

• 数据库源通常会将 defaultCursorField 留空，即使它们声明了增量模式。它们会通过 propertyFields 提供所有列，供你选择。对于数据库源，计划从 propertyFields 中选择；SaaS 源通常会预先填好合适的 defaultCursorField。
• 数据库会设置 streamnamespace（例如 public、dbo），而大多数 SaaS 源不会设置。只要发现响应中包含该字段，YAML 就需要 namespace: 行；对于没有该字段的流，请删除这一行。

阶段 3：询问使用场景

提出流建议之前，先问一个简短的问题，以明确选择依据。例如：

• “你想通过此连接分析什么？例如，每位客户的收入、支持工单趋势、营销漏斗。”
• “这里关注源数据的哪一部分——销售渠道、财务记录，还是产品使用情况？”

保持开放式；用户回答一句话就足够。目标是筛除明显无关的流（审计日志、内部队列、系统表），并优先选择业务实体。只有当用户已在初始请求中说明目标时，才跳过此问题。

阶段 4：提出流子集（检查点 1）

按以下顺序应用这些规则：

1. 排除技术/系统流，除非用户的目标明确需要它们。名称匹配以下任一模式：以 pg_、information_schema、_airbyte、temp_、tmp_、audit_、system_、migration、schema_migrations 开头；以 _history、_log、_audit、_archive 结尾。谨慎处理——不确定时就保留，并标明这一点。

2. 对于数据库（postgres / mysql / mssql / mongodb 风格的数据源，可根据大量流以及 public、dbo、sales 等命名空间识别）：预计排除 30–80% 的流。项目用例很少需要每一张业务表。优先选择名称与用户目标相符的流（分析收入时选择 orders、customers、products；支持场景则选择 tickets、contacts、messages）。

3. 对于 SaaS 数据源（流列表较短且扁平、没有命名空间，名称类似 companies、deals、tickets、engagements）：默认包含所有核心业务实体。仅排除明显的冗余项（*_metadata、*_history、*_changelog）。

4. 如实说明不确定之处。 如果仅凭名称无法判断流的用途，就直说——不要编造理由。

用便于用户浏览的表格呈现建议：

Proposed streams (12 of 47):

  customers         core entity — revenue analysis needs this
  orders            transactions table; cursor: updated_at
  order_items       order line items; needed to break revenue by SKU
  products          dimension table for orders/order_items
  ...

Dropped (35): pg_stat_*, _airbyte_internal_*, audit_*, schema_migrations,
              user_sessions (not in scope for revenue analysis), ...

询问：“看起来合适吗？有什么要添加或删除的吗？”等待用户回复。他们可能会说“添加 users”或“删除 products，我们不需要它”——据此调整。

阶段 5：确定每个流的同步模式、游标和主键

对于每个选定的流，选择取决于源支持的内容（syncModes）以及其提供的字段。你将使用以下四种同步模式：

同步模式	要求	适用场景
incremental_deduped_history	游标 和 主键	最佳默认选择。源提供游标（例如 updated_at）和主键。仅拉取新增或变更的行；按主键合并重复项。
incremental_append	仅游标	源有游标但没有主键（事件流、仅追加日志）。行会不断累积；不去重。
full_refresh_overwrite_deduped	仅主键	源有主键但没有可用游标。每次同步都会替换目标表，并按主键去重。适用于小型或中型表。
full_refresh_overwrite	无	小型维度表（<10k 行），或其他方法都行不通时。每次同步都会覆盖。

每个流的决策算法：

1. 选择源支持的最强模式，优先级如下：incremental_deduped_history → incremental_append → full_refresh_overwrite_deduped → full_refresh_overwrite。跳过不在流的 syncModes 数组中的模式。
2. 对于需要游标的模式：如果 defaultCursorField 非空，则使用它。否则，在 propertyFields 中查找第一个可用的时间戳类字段（updated_at、modified_at、last_modified、_updated、timestamp）。如果没有合适的字段，则改用全量刷新模式。
3. 对于需要主键的模式：如果 sourceDefinedPrimaryKey 非空，则使用它。否则，在 propertyFields 中查找类似 id 的字段（id、<entity>_id、uuid、key）。如果没有合适的字段，则改用 incremental_append 或 full_refresh_overwrite。

展示时使用简洁表示法（检查点 2）：

Stream             Sync mode                       Cursor        Primary key
customers          incremental_deduped_history     updated_at    id
orders             incremental_deduped_history     updated_at    id
order_items        incremental_append              created_at    —
products           full_refresh_overwrite          —             —

询问：“同步模式看起来合适吗？需要调整吗？” 常见调整：仅当大型表实际很小或静态时，才将其改为 full_refresh_overwrite；切换游标字段。

第 6 阶段：编写 YAML

需要选择两个名称，规则各不相同：

• metadata.name — 本地 IaC slug。适合用作文件名：小写字母、数字和连字符。它决定 YAML 文件路径（connections/<metadata.name>.yaml），以及其他资源如何引用此连接。根据来源和范围命名：hubspot-sales、postgres-prod-revenue、stripe-billing。
• spec.name — 在 RevOS UI 中显示的易读标签。简短，使用句首字母大写的格式，可以包含空格，不要使用下划线。描述此连接同步的内容，让团队成员扫一眼 UI 就能了解，无需打开详情。可用的格式示例："HubSpot sales pipeline"、"Postgres prod → revenue tables"、"Stripe billing & subscriptions"。避免在这里重复使用 slug——hubspot-sales 作为 spec.name 就不如 HubSpot sales pipeline。

如果目标路径下已存在文件，请询问是要覆盖还是选择其他名称。

模板：

apiVersion: revos/v1
kind: Connection
metadata:
  name: hubspot-sales
spec:
  name: HubSpot sales pipeline
  source: { id: <id from `revos sources list`> }
  schedule: { units: 24, timeUnit: hours }
  status: active
  streams:
    - name: customers
      namespace: public # omit if the stream has no namespace
      syncMode: incremental_deduped_history
      cursorField: [updated_at]
      primaryKey: [[id]]

备注：

• source.id 是 Source 的服务端 ID（例如 src_abc123）。Sources 不属于 IaC——使用 revos sources list 获取 ID。
• primaryKey 是一个列表的列表——每个内层列表都是一个主键列路径，支持嵌套键。大多数情况下使用 [[id]]。
• cursorField 是扁平的路径段列表。顶层字段为 [updated_at]；嵌套字段则为 [meta, modified_at]。
• 对于 full_refresh_overwrite，省略 cursorField 和 primaryKey。对于 full_refresh_overwrite_deduped，只省略 cursorField。CLI 会在应用时拒绝缺少必填字段的配置，因此请保持配置简洁。
• 不要设置 metadata.id 或 spec.prefix——首次创建时，revos apply 会填充这两项。

流映射器（streams[].mappers）是在数据行写入 BigQuery 前应用的服务端转换——例如，对 PII 进行哈希处理、重命名列、删除字段、筛选数据行、加密值。默认情况下，进行干净的同步，不使用映射器。出现以下任一情况时，加载 references/mappers.md：

• 用户提到遮蔽、哈希处理、PII、重命名或删除列、筛选数据行或加密。
• 所选流的 propertyFields 包含明显敏感的列（email、phone、ssn、ip_address、成对的姓名字段、支付字段）。此时加载参考文档，然后主动提出映射器，并说明字段名称和理由，再询问用户确认后添加。不要悄悄插入映射器——遮蔽错误字段会破坏分析结果。

第 7 阶段：在本地验证

revos diff

CLI 会通过 zod 解析 YAML，并报告将发生的变更。请检查：

• 解析错误——架构拒绝了配置（缺少必填字段、同步模式无效）。修复后重新运行。
• 差异报告——新连接应显示为单个 create 条目。

如果 revos diff 没有问题，请返回：「连接 YAML 已写入 connections/<slug>.yaml 并通过验证。运行 revos apply 即可在服务端创建连接。」

如果验证失败，请原样提供错误信息，并在宣布成功前修复。

---

常见问题

• 选择了来源不支持的同步模式。 始终将所选模式与第 2 阶段中流的 syncModes 数组取交集。API 会在服务端验证；最好提前在本地发现问题。
• 在需要服务端 ID 的地方使用来源名称。 spec.source.id 必须是服务端 ID（例如 src_abc123），而不是显示名称或 slug。始终从 revos sources list 获取。
• 在 full_refresh_* 模式下设置 cursorField。 CLI 验证允许存在额外字段，但这只会造成干扰。省略不适用于所选模式的字段。
• 提出所有流。 数据库尤其如此。如果用户在看到建议后说「全部都要」，当然可以——但先提供经过筛选的列表。
• 虚构字段。 只能使用发现响应中出现的 cursorField 和 primaryKey 值（defaultCursorField、sourceDefinedPrimaryKey 或 propertyFields）。