功能概述 #

MongoDB CDC Pipeline Source Connector 用于从 MongoDB Collection 读取初始快照，并在快照完成后持续捕获 Change Streams 增量变更。Connector 会根据 BSON 文档推导 Pipeline Schema，并将数据变更发送给下游 Sink。

版本兼容性 #

特色功能 #

• 支持先读取一致性快照，再无缝切换到 Change Streams 增量读取。
• 支持从最新位点或指定毫秒时间戳开始消费增量事件。
• 支持根据采样文档推导 Pipeline Schema。
• 支持展开嵌套文档、合并数值类型，以及将原始 BSON 类型统一转换为字符串。
• 支持读取 ts_ms 和 op_ts 元数据，并可选择忽略 Delete 或输出完整 Changelog。

前提条件 #

1. 已开通火山引擎流式计算 Flink 版，并准备可用的 Pipeline Sink。
2. MongoDB 版本为 3.6 或以上，使用副本集或分片集群、WiredTiger 存储引擎和 Replica Set Protocol v1；分片集群应通过 mongos 连接。
3. Flink 资源池能够访问 MongoDB 地址和端口。
4. 目标 Collection 中的文档包含非空 _id。
5. MongoDB 用户至少具有读取目标数据库、Collection 和 Change Streams 的权限。同步多个库或集合、执行快照切分时，还需要列库、列集合、集合统计和读取分片元信息等权限。
6. 已在 Flink 中创建 MongoDB 用户名、密码等加密变量，不在 YAML 中保存明文凭证。
7. 使用完整 Changelog 时，MongoDB 版本必须为 6.0 及以上，并已开启数据库级 pre/post image 保留和 Collection 级 changeStreamPreAndPostImages。

使用限制 #

• Connector 仅支持作为 Flink CDC Pipeline Source。
• _id 固定作为 Pipeline 主键，必须非空，且统一转换为 STRING；不能指定其他字段作为主键。
• Change Streams 不支持 MongoDB Standalone 部署。
• 初始 Schema 来自有限数量的采样文档。生产使用前应确保采样数据能够覆盖下游需要的字段和主要类型。
• 默认 Update 不包含完整变更前镜像；完整 Changelog 依赖 MongoDB pre-image。
• 开启 scan.ignore-delete.enabled 后，下游不会收到 Delete，目标数据可能保留源端已删除记录。
• latest-offset 不读取任务启动前的存量数据和历史变更；timestamp 只能读取 MongoDB oplog/Change Stream 历史保留范围内的事件。
• 跳过快照 Backfill 可能使快照阶段发生的变更在切换到增量阶段后重复发送，应由下游幂等处理。
• 展开嵌套字段时可能发生字段名冲突，应显式选择冲突策略。

语法结构 #

source:
  type: mongodb
  name: MongoDB Source
  hosts: <mongodb_host>:27017
  username: ${secret_values.mongodb-username}
  password: ${secret_values.mongodb-password}
  connection.options: replicaSet=<replica_set>&authSource=<auth_database>
  database: <database_name>
  collection: <collection_name>
  scan.startup.mode: initial

sink:
  type: values

pipeline:
  name: <pipeline_name>
  parallelism: 2

参数说明 #

连接与同步范围参数 #

启动与增量读取参数 #

Schema、快照与元数据参数 #

类型映射 #

_id 固定映射为 STRING。ObjectId 转换为十六进制字符串；复合文档形式的 _id 转换为字段名排序后的 JSON 字符串。

类型合并规则 #

• 数值候选类型会向更宽类型合并，例如 INT 与 BIGINT 合并为 BIGINT。
• 任一候选类型为 STRING 时，最终类型为 STRING。
• Array 元素无法合并时退化为 ARRAY<STRING>。
• Map Value 无法合并时退化为 MAP<STRING, STRING>。
• 开启 scan.all.primitives.as-string.enabled 后，数值、布尔、时间和二进制等原始类型统一推导为 STRING。

使用示例 #

读取全量和增量数据 #

以下示例从 <database_name>.<collection_name> 读取存量数据，并持续消费 Change Streams。values Sink 用于验证 Source 产生的数据。

source:
  type: mongodb
  name: MongoDB Source
  hosts: <mongodb_host>:27017
  username: ${secret_values.mongodb-username}
  password: ${secret_values.mongodb-password}
  connection.options: replicaSet=<replica_set>&authSource=<auth_database>
  database: <database_name>
  collection: <collection_name>
  scan.startup.mode: initial
  scan.max.pre.fetch.records: 50
  scan.local-time-zone: Asia/Shanghai

sink:
  type: values

pipeline:
  name: MongoDB CDC Pipeline
  parallelism: 2

预期结果。 Source 先输出 Collection 已有文档，再持续输出新增、更新和删除事件；_id 作为字符串主键发送给下游。

从指定时间戳读取增量数据 #

source:
  type: mongodb
  hosts: <mongodb_host>:27017
  username: ${secret_values.mongodb-username}
  password: ${secret_values.mongodb-password}
  database: <database_name>
  collection: <collection_name>
  scan.startup.mode: timestamp
  scan.startup.timestamp-millis: <epoch_milliseconds>

该时间必须位于 MongoDB 仍可恢复的 Change Stream 历史范围内，否则作业无法从指定位点启动。

输出完整 Changelog #

scan.full-changelog: true 会使用 MongoDB pre-image 和 post-image 生成 Insert、Update Before、Update After 和 Delete。按照 Apache Flink CDC 官方 MongoDB Connector 文档，需要完成以下配置。

1. 使用 MongoDB 6.0 或以上版本。
2. 在数据库级配置 pre/post image 保留时间。以下示例使用 MongoDB 默认保留策略：

db.runCommand({
  setClusterParameter: {
    changeStreamOptions: {
      preAndPostImages: {
        expireAfterSeconds: "off"
      }
    }
  }
})

1. 为需要同步的 Collection 开启 pre/post image：

db.runCommand({
  collMod: "<collection_name>",
  changeStreamPreAndPostImages: {
    enabled: true
  }
})

1. 在 Source 中开启完整 Changelog：

source:
  type: mongodb
  # 其他连接参数
  scan.full-changelog: true

pre/post image 的保留时间应覆盖作业可能的消费延迟和恢复时间。镜像已过期或未启用时，无法生成完整变更前数据。

运行机制 #

initial 模式首先按快照 Chunk 并行读取已有文档，快照完成后切换到 Change Streams。MongoDB CDC Source 基于 Checkpoint 保存快照进度和 Change Stream 读取位置，可提供 Exactly-Once 处理能力。端到端一致性还取决于下游 Sink 的交付语义；对于不支持事务或 Exactly-Once 的下游，建议使用业务主键实现幂等写入。
Connector 对采样文档执行初始 Schema 推导。生产使用前应使用具有代表性的文档验证字段和类型映射，并确保下游 Schema 与推导结果兼容。
默认 Update 主要输出更新后的文档。启用 scan.full-changelog 后，Connector 会尝试读取 Update/Delete 前的文档镜像；MongoDB 未返回 pre-image 时，变更前数据可能为空或仅包含文档键。

验证与排障 #

验证方法 #

1. 启动作业并确认进入运行状态，日志中没有网络、鉴权、Change Streams 或 Schema 推导错误。
2. 等待初始快照完成，核对下游主键数与 MongoDB Collection 文档数。
3. 新增、更新和删除同一 _id 文档，确认下游收到对应变更。
4. 等待成功 Checkpoint 后执行受控重启，确认作业从保存位点恢复且最终数据正确。
5. 使用完整 Changelog 时，更新并删除测试文档，确认下游能够读取变更前镜像。

常见问题 #

注意事项 #

• 不要在需要同步删除的链路中开启 scan.ignore-delete.enabled。
• scan.incremental.snapshot.backfill.skip: true 可能产生重复事件，只适用于下游具备幂等能力的场景。
• 初始 Schema 由采样文档推导，生产使用前应使用代表性数据验证字段覆盖和类型映射。
• MongoDB 密码必须通过 Flink 加密变量引用，不要明文写入 Pipeline YAML。
• timestamp 和 Checkpoint 恢复依赖 MongoDB Change Stream 历史保留，停机时间不得超过可恢复窗口。

相关文档 #

• Apache Flink CDC：MongoDB CDC Connector
• MongoDB 官方文档：Change Streams
• 火山引擎流式计算 Flink 版：使用变量