TOS-CDC SQL Connector 是流式计算 Flink 版提供的源表（Source）连接器，用于持续获取火山引擎对象存储 TOS 中的对象元数据。连接器可先扫描指定路径下的已有对象，再从 Kafka 接收对象创建、删除等事件，将全量快照与增量变化组合为一张持续更新的 Flink SQL 表。

• 全量与增量一体化：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服务，并创建用于存放待采集对象的 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。

使用 CREATE TABLE 语句创建 TOS-CDC 源表。字段可以按任意顺序声明，也可以只声明需要使用的字段，但必须包含 object_key字段，且必须为STRING NOT NULL。

CREATE TABLE <source_table> (
    object_key    STRING NOT NULL,
    object_url    STRING,
    bucket_name   STRING,
    file_name     STRING,
    object_etag   STRING,
    object_size   BIGINT,
    mtime         TIMESTAMP_LTZ(3),
    event_time    TIMESTAMP_LTZ(3),
    record_origin STRING
    -- PRIMARY KEY (object_key) NOT ENFORCED -- 可选
) WITH (
    'connector' = 'tos-cdc',
    'path' = 'tos://<bucket>/<path>',
    'scan.startup.mode' = '<startup_mode>',
    'topic' = '<kafka_topic>',
    'properties.bootstrap.servers' = '<kafka_broker>',
    'properties.group.id' = '<consumer_group>'
);

固定Schema说明 #

TOS-CDC SQL Connector 使用固定的对象元数据 Schema。
DDL 可声明下表中的任意字段子集，但 object_key 必须声明为 STRING NOT NULL。

通用连接参数 #

启动模式 #

以下是针对启动模式参数scan.startup.mode 的说明：

Kafka 参数透传 #

所有以 properties. 开头的参数都会去掉该前缀后透传给底层 Kafka Consumer。例如：

CREATE TABLE object_meta (
    object_key STRING NOT NULL
) WITH (
    'connector' = 'tos-cdc',
    'path' = 'tos://my-bucket/ingest/',
    'scan.startup.mode' = 'initial',
    'topic' = 'object-events',
    
    'properties.bootstrap.servers' = 'kafka.example:9092',
    'properties.group.id' = 'ingest-cg',
    'properties.security.protocol' = 'SASL_SSL',
    'properties.sasl.mechanism' = 'PLAIN',
    'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username="<username>" password="<password>";',
    'properties.max.poll.records' = '500'
);

以下示例使用 initial 模式读取指定 Bucket 中两个路径下的已有对象，并持续消费 Kafka 中的对象事件。示例将变更日志写入 Print Connector 来查看输出结果。

CREATE TABLE object_changes (
    object_key    STRING NOT NULL,
    object_url    STRING,
    bucket_name   STRING,
    file_name     STRING,
    object_etag   STRING,
    object_size   BIGINT,
    mtime         TIMESTAMP_LTZ(3),
    event_time    TIMESTAMP_LTZ(3),
    record_origin STRING,
    PRIMARY KEY (object_key) NOT ENFORCED
) WITH (
    'connector'                    = 'tos-cdc',
    'path'                         = 'tos://my-bucket/images01/, tos://my-bucket/images02/',    
    'properties.bootstrap.servers' = 'kafka.example:9092',
    'properties.group.id'          = 'ingest-cg',
    'topic'                        = 'object-events',
    'scan.startup.mode'            = 'initial'
);

CREATE TABLE object_state (
    object_key    STRING NOT NULL,
    object_url    STRING,
    bucket_name   STRING,
    file_name     STRING,
    object_etag   STRING,
    object_size   BIGINT,
    mtime         TIMESTAMP_LTZ(3),
    event_time    TIMESTAMP_LTZ(3),
    record_origin STRING
) WITH (
    'connector' = 'print'
);

INSERT INTO object_state
SELECT
    object_key,
    object_url,
    bucket_name,
    file_name,
    object_etag,
    object_size,
    mtime,
    event_time,
    record_origin
FROM object_changes;

输出结果示例：

1> +I[full-data/images/test09.jpg, tos://zx-source-test/full-data/images/test09.jpg, zx-source-test, test09.jpg, null, 16673, 2026-07-27T11:43:35Z, null, SNAPSHOT]
2> +I[full-data/images/test14.jpg, tos://zx-source-test/full-data/images/test14.jpg, zx-source-test, test14.jpg, "5b31d1c97b76087ba6129148a9fdd3b1", 517, null, 2026-07-27T18:01:45Z, EVENT]
1> +I[full-data/images/test13.jpg, tos://zx-source-test/full-data/images/test13.jpg, zx-source-test, test13.jpg, "5b31d1c97b76087ba6129148a9fdd3b1", 517, null, 2026-07-27T18:01:45Z, EVENT]
1> -D[full-data/images/test13.jpg, tos://zx-source-test/full-data/images/test13.jpg, zx-source-test, test13.jpg, "5b31d1c97b76087ba6129148a9fdd3b1", 517, null, 2026-07-27T18:02:00Z, EVENT]
1> -D[full-data/images/test06.jpg, tos://zx-source-test/full-data/images/test06.jpg, zx-source-test, test06.jpg, "ef4bc467b27e4bc95a7d23f534ae95b8", 427, null, 2026-07-27T18:02:00Z, EVENT]

全量与增量衔接 #

在 initial 模式下，连接器启动时记录时间 T_snapshot_start，然后对 path 指定的对象范围执行全量 LIST。全量阶段结束后，连接器从 Kafka Topic 中 T_snapshot_start - rewind.offset 对应的位置开始消费，避免遗漏全量扫描期间产生的对象事件。
生产环境建议将 rewind.retention-miss-policy 保持为默认值 fail。当回拨目标早于 Kafka 最早保留消息时，作业会失败，避免在没有提示的情况下丢失事件。只有在明确接受历史事件丢失风险时，才设置为 earliest。

变更日志语义 #

全量快照记录和对象 PUT 事件输出 RowKind.INSERT，对象删除事件输出 RowKind.DELETE。连接器不输出 UPDATE_BEFORE 或 UPDATE_AFTER。
若需要将删除事件同步到结果表，下游 Sink 必须支持 Flink DELETE 变更日志，并配置能够唯一标识对象的主键。单 Bucket 场景可使用 object_key；将多个 Bucket 汇总到同一结果表时，建议使用 object_url。

验证方法 #

1. 使用“使用示例”中的 SQL 启动作业。
2. 确认全量扫描阶段能够输出 record_origin = SNAPSHOT 的 +I 记录。
3. 在配置路径下上传对象，确认作业输出 record_origin = EVENT 的 +I 记录。
4. 删除测试对象，确认作业输出对应的 -D 记录。

常见问题 #

• 使用 initial 模式前，应根据 Bucket 对象规模评估全量扫描耗时，并为 Kafka Topic 配置足够的消息保留时间。
• rewind.offset 会扩大重复消费的时间窗口。增大该值前，应确认下游具备符合业务语义的去重能力。
• 一个 TOS-CDC Source 只能读取一个 Bucket。配置多个 URI 时，所有 URI 必须属于同一个 Bucket，并使用英文逗号分隔。重复路径以及已被父路径覆盖的子路径会被去重。
• Source 表是否声明主键不影响连接器输出 DELETE 事件；下游 Sink 是否支持删除及是否配置主键，决定删除事件能否同步到结果表。

• 设置事件通知推送至 Kafka：配置 TOS Bucket 的事件通知规则，将对象创建、覆盖和删除事件推送至 Kafka。它是本 Connector 实现增量采集最重要的前置文档。
• TOS Filesystem Connector：用于按文件系统语义读取或写入TOS对象数据。
• 对象存储 TOS 产品文档：了解 TOS Bucket、对象 Key、访问权限、SDK 和 API 等基础能力。
• Flink SQL 连接器列表：了解火山引擎 Flink SQL Connector 的文档分类和支持范围。
• Flink CDC Pipeline TOS 连接器：了解火山引擎 Flink CDC Pipeline TOS Connector的配置参数和数据读取方式。