TOS-CDC Pipeline Source 用于捕获火山引擎对象存储 TOS 中对象的元数据变更。该连接器可先读取指定 TOS 路径下的存量对象，再持续消费 Kafka 中的对象事件通知，并将对象的创建、覆盖和删除操作转换为 Flink CDC Pipeline 事件。

版本兼容性 #

• 全量与增量一体化：initial 模式先扫描指定 TOS 路径下的已有对象，再从全量扫描开始时间对应的 Kafka 位点持续消费对象事件，用于构建并维护对象元数据状态。
• 多种启动位置：支持 initial、snapshot、earliest-offset、latest-offset、group-offsets、timestamp 和 specific-offsets 七种启动模式，可以按需执行全量加增量同步、仅快照读取、增量续读或历史事件回放。
• 多路径扫描：一个 Source 可以配置同一 TOS Bucket 下的多个 URI。Connector 会对重复路径以及已被父路径覆盖的子路径进行去重，便于统一采集同一 Bucket 内多个目录的对象元数据。
• 全量和增量独立并行：支持分别配置 TOS 全量快照阶段和 Kafka 增量阶段的并行度，便于根据对象数量、Kafka Partition 数量和作业资源调整处理能力。
• 消费位点安全回拨：initial 模式支持通过 rewind.offset 将 Kafka 消费起点回拨到全量扫描开始时间之前，降低全量扫描期间对象事件遗漏的风险；同时支持配置 Kafka 历史消息过期时的处理策略。
• 事件范围校验：默认过滤 Kafka Topic 中不属于当前 Source 范围的对象事件；开启 event.fail-on-out-of-scope 后，可在发现范围外事件时使任务失败，帮助及时发现 Topic 或数据源配置错误。
• 统一对象元数据 Schema：快照记录与增量事件使用相同的固定 Schema，并通过 record_origin 区分 SNAPSHOT 和 EVENT 来源，便于下游统一处理和排查重复数据。
• 支持维护对象当前状态：可以将 object_key 声明为逻辑主键，并配合支持 INSERT、DELETE Changelog 的下游 Sink，维护 Bucket 中对象的当前元数据状态。

使用 TOS-CDC Pipeline Source 前，请完成以下准备工作：

• 已开通火山引擎对象存储 TOS，并创建用于存放待采集对象的 Bucket。
• 已确认待采集对象所在的 TOS 路径。一个 Source 可以配置同一 Bucket 下的一个或多个 URI，不支持同时读取多个 Bucket。
• 已为 Flink 作业配置访问目标 Bucket 的权限，确保 Connector 可以列举指定路径下的已有对象并读取对象元数据。
• 已为目标 TOS Bucket 配置事件通知，并将推送目标设置为消息队列 Kafka 版。事件通知至少应订阅对象创建事件 tos:ObjectCreated:*；如需同步对象删除，还应订阅 tos:ObjectRemoved:*。事件通知规则的 Prefix、Suffix 应覆盖 Source 配置的对象路径。详细操作请参见设置事件通知推送至 Kafka
• 已确保 Flink 资源池与 TOS、Kafka 之间网络互通，并准备 Kafka Broker 地址、Topic 和 Consumer Group ID。

以下示例先读取指定 TOS 路径下的已有对象，再持续消费对象事件，并将结果写入 Paimon：

sources:
- source:
    type: tos-cdc
    name: TOS CDC Source
    table-id: object_store.object_changes
    path: tos://my-bucket-test/model-data/images
    schema.primary-key: object_key
    properties.bootstrap.servers: kafka-example01.kafka.ivolces.com:9092
    properties.group.id: test-group-id
    topic: test-topic
    scan.startup.mode: initial

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: tos://my-bucket-test/warehouse

pipeline:
  name: TOS Object Metadata to Paimon
  parallelism: 4

连接器配置项 #

启动模式配置项 #

配置项 scan.startup.mode 用于指定 TOS CDC Source 的启动模式：

输出字段 #

TOS CDC Source 使用固定的对象元数据 Schema：

读取 TOS 全量和增量对象元数据 #

以下示例先扫描 tos://my-bucket-test/model-data/images 路径下的已有对象，再持续消费 Kafka Topic 中的对象创建、覆盖和删除事件，最后将对象元数据写入 Paimon 表。
运行前请替换 TOS 路径、Kafka Broker 地址、Consumer Group ID、Topic 和 Paimon Warehouse 路径。

sources:
  - source:
      type: tos-cdc
      name: TOS CDC Source
      table-id: object_store.object_changes
      # 需要采集的 TOS 路径
      path: tos://my-bucket-test/model-data/images
      # Source 输出表的逻辑主键
      schema.primary-key: object_key
      # 接收 TOS 对象事件通知的 Kafka
      properties.bootstrap.servers: kafka-example01.kafka.ivolces.com:9092
      properties.group.id: test-group-id
      topic: test-topic
      # 先读取已有对象，再持续读取增量事件
      scan.startup.mode: initial
      
sink:
  type: paimon
  name: Paimon Sink
  # 使用文件系统方式管理 Paimon Catalog
  catalog.properties.metastore: filesystem
  # Paimon Warehouse 存储路径
  catalog.properties.warehouse: tos://my-bucket-test/warehouse
  
pipeline:
  name: TOS Object Metadata to Paimon
  parallelism: 4

配置说明：

• sources：定义 Pipeline 的数据来源。本示例使用 TOS-CDC Source。
• path：指定需要扫描的 TOS Bucket 和对象路径。
• table-id：定义 Source 在 Pipeline 中产生的逻辑表标识。本示例使用 object_store.object_changes。
• schema.primary-key：声明 Source 输出表的逻辑主键。本示例使用 object_key 标识对象。
• properties.bootstrap.servers：Kafka Broker 地址。
• properties.group.id：Kafka Consumer Group ID，用于标识该作业的 Kafka 消费组。
• topic：接收 TOS 对象事件通知的 Kafka Topic。
• scan.startup.mode: initial：启动时读取指定路径下的已有对象，完成后持续处理 Kafka 中的增量事件。
• sink：定义数据写入目标。本示例将结果写入 Paimon。
• pipeline.parallelism：设置 Pipeline 的默认并行度。

预期结果 #

假设作业启动前，指定路径下已有以下对象：

model-data/images/a.jpg
model-data/images/b.jpg

全量扫描阶段会为两个对象分别输出一条 INSERT 事件。
作业进入增量阶段后，如果依次发生以下操作：

创建 model-data/images/c.jpg
覆盖 model-data/images/a.jpg
删除 model-data/images/b.jpg

Source 会输出：

该 Connector 不会为对象覆盖操作输出 UPDATE_BEFORE 和 UPDATE_AFTER。
如果 Paimon 目标表以 object_key 为主键并按主键更新，最终表中预计包含 a.jpg 和 c.jpg，不再包含已删除的 b.jpg。实际结果还取决于 Paimon 目标表结构及 Sink 的写入语义。

TOS-CDC Pipeline Source 通过“TOS 全量扫描 + Kafka 增量事件消费”持续维护指定路径下的对象元数据。

全量扫描 #

当 scan.startup.mode 设置为 initial 时，Connector 首先列举 path 指定范围内的已有对象，读取每个对象的元数据信息，并将每个对象转换为一条 INSERT 事件。
例如，指定：

path: tos://my-bucket-test/model-data/images

启动时该路径下存在：

model-data/images/a.jpg
model-data/images/b.jpg

Source 会分别输出 a.jpg 和 b.jpg 的对象元数据。

全量与增量衔接 #

全量扫描用于发现作业启动前已经存在的对象，Kafka 事件用于捕获作业运行期间发生的对象变化。
全量扫描完成后，Source 持续消费 Kafka Topic 中的对象事件通知。为降低全量扫描和增量消费交接过程中遗漏对象变化的风险，Kafka 增量读取范围可能与全量扫描范围发生重叠。
例如：

1. 作业开始扫描 images/。
2. 扫描期间，用户创建了 images/c.jpg。
3. 全量扫描可能已经发现 c.jpg。
4. Kafka Topic 中也可能收到 c.jpg 的创建事件。
5. 因此，Source 可能输出两条与 c.jpg 有关的 INSERT 事件。

增量事件处理 #

Connector 持续消费指定 Kafka Topic 中的 TOS 对象事件，并根据 Source 的采集范围筛选事件。
当前事件转换规则如下：

对象覆盖不会转换为 UPDATE_BEFORE 和 UPDATE_AFTER。如果下游需要把同一对象的多次 PUT 合并为当前状态，可以使用 object_key 等对象标识作为主键，通过主键表的 Upsert 语义处理。

状态保存与故障恢复 #

作业运行期间，Flink 通过 Checkpoint 定期保存 Source 的处理状态，例如全量扫描进度和 Kafka 消费位置。
任务发生故障时，会从最近一次成功的 Checkpoint 恢复。Checkpoint 完成后新处理、但尚未被下一次 Checkpoint 确认的事件可能被重新消费，因此 Source 采用至少一次（At-Least-Once）处理语义：

• 优先保证对象事件不被遗漏。
• 故障恢复后可能重新处理部分事件。
• 下游可能收到内容相同的重复数据。

端到端一致性不仅取决于 Source，还取决于 Sink 是否支持事务提交、主键更新或幂等写入。

数据写入下游 #

Source 输出的对象变更事件经过 Pipeline 传递给 Sink。本示例使用 Paimon Sink：

• INSERT 用于新增对象记录，或者更新主键相同的对象状态。
• DELETE 用于删除主键对应的对象记录。
• 重复的 INSERT 是否形成重复行，取决于目标表是否具有主键以及 Paimon Sink 的具体写入语义。

如果下游保存的是对象当前状态，建议采用主键表；由于当前输出 Schema 不包含稳定的事件标识，无法可靠区分事件重放与同一对象的多次真实变更，因此不适合用于需要精确事件级去重或审计的完整变更历史场景。

验证方法 #

首次使用 TOS-CDC Pipeline Source 时，建议使用测试 Bucket、测试路径和独立的 Kafka Consumer Group 完成全量扫描、增量采集和故障恢复验证，再应用于生产环境。
建议按照以下步骤进行验证：

1. 检查 path 中的 Bucket 和对象路径是否存在，并确认 Flink 作业具有列举对象和读取对象元数据的权限。
2. 检查 Flink 资源池与 TOS、Kafka 和 Paimon Warehouse 之间是否网络互通。
3. 检查 Kafka Topic 是否存在，并确认 TOS 对象事件通知已发送至该 Topic。
4. 提交 Pipeline，确认作业进入运行状态。
5. 检查全量扫描结果，确认 path 范围内的已有对象已写入下游。
6. 在指定 TOS 路径下依次创建、覆盖和删除测试对象。
7. 检查下游是否收到对应的 INSERT 和 DELETE 事件。
8. 等待至少一次 Checkpoint 成功后执行受控重启，确认作业可以恢复全量扫描进度或 Kafka 消费位置。
9. 在 Kafka 消息仍然有效且事件通知正常投递的前提下，恢复期间可能出现重复事件，但 Source 会按照已保存的消费位置继续处理。

验证对象事件 #

假设 Source 配置如下：

path: tos://my-bucket-test/model-data/images

可以在该路径下执行以下测试：

对象覆盖事件能否覆盖下游已有记录，取决于 Sink 是否支持主键更新以及目标表是否正确配置主键。

常见问题 #

1. 一个 TOS-CDC Source 只能读取一个 Bucket。配置多个 URI 时，所有 URI 必须属于同一个 Bucket，并使用英文逗号分隔。重复路径以及已被父路径覆盖的子路径会被去重；URI 中的空格、中文和其他非法 URI 字符必须进行编码，避免路径解析失败。
2. 应确保 Kafka Topic 中的对象事件与 Source 配置的 Bucket 和路径范围一致。Topic 混入其他范围的事件时，事件可能被过滤，或者在开启严格范围校验后导致任务失败。
3. scan.startup.mode: initial 会读取已有对象并继续消费增量事件。全量扫描与增量消费可能存在重叠，下游不能假设每个对象事件只会收到一次。
4. 调整 rewind.offset 前应评估重复数据范围。生产作业不建议在缺少去重方案的情况下随意向前回拨消费位置。
5. 如果下游用于保存对象当前状态，建议使用能够唯一标识对象的主键，并选择支持 Upsert 和 DELETE 的 Sink。如果一个 Topic 或一张目标表可能包含多个 Bucket 的对象，不能仅使用 object_key 作为全局主键，应考虑使用 Bucket 和对象 Key 的组合作为主键。

设置事件通知推送至 Kafka：配置 TOS Bucket 的事件通知规则，将对象创建、覆盖和删除事件推送至 Kafka。它是本 Connector 实现增量采集最重要的前置文档。
对象存储 TOS 产品文档：了解 TOS Bucket、对象 Key、访问权限、SDK 和 API 等基础能力。
Flink CDC Pipeline 连接器列表：了解火山引擎 Flink CDC Pipeline Connector 的文档分类和支持范围。
Kafka Pipeline Connector：了解 Kafka Connector 的配置参数和数据读取、写入方式。
Paimon Pipeline Connector：了解示例中 Paimon Sink 的 Catalog、Warehouse、主键和写入配置。
Apache Flink CDC：Data Pipeline：了解 Flink CDC Pipeline 中 Source、Sink、Transform、Route 和 Pipeline 的关系。
Apache Flink：Checkpoint：了解 Flink 如何保存作业状态和数据流读取位置，以及故障恢复机制。
Apache Flink：Fault Tolerance：了解 At-Most-Once、At-Least-Once 和 Exactly-Once 等处理语义。