OSS CDC（公测中）-实时计算 Flink版(Flink)-阿里云帮助中心

OSS CDC连接器支持通过阿里云消息服务MNS事件通知或周期性轮询OSS对象元数据两种方式，持续发现对象存储OSS中的数据变化。其中，MNS模式可以消费对象创建、删除等事件；POLLING模式无需配置MNS，通过定期扫描发现新增或发生变化的对象。连接器同时支持全量扫描指定路径下的对象元数据。本文为您介绍如何使用OSS CDC连接器。

背景信息

OSS CDC连接器支持的信息如下。

类别	详情
支持类型	源表
运行模式	仅支持流模式
数据格式	暂不适用（固定Schema）
API种类	SQL和数据摄入YAML
是否支持更新或删除结果表数据	不涉及

特色功能

事件驱动：MNS模式通过MNS队列实时消费OSS事件通知，延迟低，无需轮询OSS接口。
轮询模式：POLLING模式无需创建MNS队列或配置OSS事件通知，通过周期性扫描OSS对象元数据发现新增和发生变化的对象。
全量+增量：支持SNAPSHOT模式全量扫描指定路径下的对象；INITIAL模式先输出已有对象，再使用MNS或POLLING方式持续发现增量变化。
路径过滤：SNAPSHOT、INITIAL和POLLING增量读取支持通过path参数指定多个路径前缀，仅扫描目标目录。
限速控制：支持scan.rate-limit.records-per-second参数限制读取速率，防止下游压力过大。
灵活启动：MNS模式支持EARLIEST_OFFSET、LATEST_OFFSET、COMMITTED_OFFSET、TIMESTAMP、INITIAL和SNAPSHOT；POLLING模式支持LATEST_OFFSET、TIMESTAMP、INITIAL和SNAPSHOT，满足不同业务场景。

前提条件

使用OSS CDC连接器前，需满足以下条件。

配置项	要求
OSS事件通知（MNS模式）	使用MNS增量模式时，需要在OSS Bucket上配置事件通知，将ObjectCreated、ObjectRemove等事件推送到MNS队列。POLLING和SNAPSHOT模式不需要配置OSS事件通知。
MNS队列（MNS模式）	使用MNS增量模式时，需要创建MNS队列，并确保其与Flink作业所在网络互通。POLLING和SNAPSHOT模式不需要MNS队列。说明仅在使用MNS模式且使用增量启动模式时，才需要MNS消息队列。
AccessKey	已创建阿里云AccessKey。MNS模式需要相应的MNS访问权限；SNAPSHOT、INITIAL和POLLING需要相应的OSS访问权限。
OSS V4签名（可选）	若OSS开启了V4签名，需设置`oss-enable-v4-signature`为`true`并配置`oss-region`。

使用MNS模式时，MNS事件通知的配置方式请参见OSS事件通知。

使用限制

OSS CDC连接器仅在实时计算引擎VVR 11.9.0及以上版本中支持。
POLLING模式仅在VVR 11.9.0 及以上版本中支持。
仅支持源表，不支持维表和结果表。
DDL中定义的列必须为固定Schema中支持的列（参见元数据），不支持自定义列。
源表不支持定义Watermark。
MNS模式下，MNS队列中的消息保留时间有限。若作业停止时间过长导致消息过期，可能丢失事件。POLLING模式不依赖MNS消息保留时间。
SNAPSHOT和INITIAL的全量阶段，以及POLLING的持续增量阶段，均依赖OSS ListObjects API。对于对象数量较多的Bucket，扫描可能耗时较长，并产生较多OSS API调用。
POLLING模式通过扫描对象元数据发现新增和变化的对象，不产生对象删除事件。需要同步删除事件时，请使用MNS模式并配置ObjectRemove事件通知。

注意事项

DDL中的key字段必须声明为NOT NULL，该字段作为对象的唯一标识。
列名匹配不区分大小写，但建议使用固定Schema中定义的列名。
使用SNAPSHOT、INITIAL或POLLING模式时，必须配置oss-endpoint和oss-bucket，并提供有效的OSS地域。使用POLLING增量模式时，需要显式配置oss-region。若配置了path，则oss-bucket为必填。
使用TIMESTAMP启动模式时，必须配置正数Epoch毫秒值scan.startup.timestamp-millis。POLLING增量模式下不支持从时间戳启动。
MNS模式建议根据业务需求合理设置mns.batch-size和mns.polling-wait-time；POLLING模式建议合理设置polling.interval和polling.prefix-size，以平衡吞吐量、扫描频率和延迟。

准备工作

配置MNS消息队列

本节仅适用于scan.incremental.mode为MNS的增量作业。使用POLLING模式或仅使用SNAPSHOT模式时，可跳过本节。

登录OSS管理控制台，创建或选择需要监控的Bucket。
登录MNS控制台，在左侧导航栏单击队列模型 > 队列列表，创建队列。
说明
- 建议选择「普通队列」而非「顺序队列」模式。
- 建议将「消息可见性超时时间」参数设置为 Checkpoint 间隔的 2 ~ 3 倍，以免消息重复投递。
在MNS控制台左侧导航栏，单击事件通知 > 对象存储OSS，单击创建规则。
在弹窗中输入规则名称，勾选需要监控的事件动作。建议仅勾选创建、复制、删除三类事件，不要勾选下载文件事件，避免出现事件循环。建议取消勾选和分片上传相关的InitiateMultipartUpload、UploadPart、UploadPartCopy三类事件，仅保留CompleteMultipartUpload事件，以免数据重复。
选择需要监控的OSS Bucket及匹配前缀。例如，仅监控images/虚拟目录下的文件变更，可在匹配规则处选择前缀并输入images/。
创建完成后，回到队列模型 > 队列列表，复制接入点中的公网访问地址，并将其配置为连接器的mns.endpoint。

配置POLLING模式

使用POLLING模式时，无需创建MNS队列或配置OSS事件通知。请确认以下事项：

Flink作业能够访问oss-endpoint。
AccessKey具有列举OSS对象及读取对象元数据的权限。
已确定需要扫描的oss-bucket、oss-region和可选的path。
根据Bucket规模合理配置polling.interval和polling.prefix-size。

配置OSS鉴权

无论使用CDC YAML还是Flink SQL作业，均需在作业的运行参数配置 > 其他配置中增加OSS鉴权参数。详情请参见配置Bucket鉴权信息。

SQL

语法结构

CREATE TEMPORARY TABLE oss_cdc_source (
  `url` STRING,
  `key` STRING NOT NULL,
  region STRING,
  bucket STRING,
  fileName STRING,
  `size` BIGINT,
  modificationTime TIMESTAMP_LTZ(3),
  eTag STRING
) WITH (
  'connector' = 'oss-cdc',
  'scan.incremental.mode' = 'MNS',
  'mns.endpoint' = '<yourMnsEndpoint>',
  'mns.region' = '<yourMnsRegion>',
  'mns.queue-name' = '<yourMnsQueueName>',
  'access-key-id' = '...',
  'access-key-secret' = '...'
);

DDL中声明的列可以是固定Schema支持列的子集，但key列必须声明且为NOT NULL。不支持声明固定Schema以外的列。

WITH参数

以下为OSS CDC连接器在Flink SQL中的WITH参数。connector固定填写oss-cdc。

通用参数

参数	说明	类型	是否必填	默认值	备注
connector	连接器类型。	STRING	是	无	固定填写`oss-cdc`。
scan.incremental.mode	增量发现模式。	STRING	否	MNS	可选值为`MNS`和`POLLING`。 MNS消费事件通知； POLLING周期性扫描OSS对象元数据。
mns.endpoint	MNS数据面服务Endpoint。	STRING	是	无	格式：`http://{account-id}.mns.{region}.aliyuncs.com`。
mns.region	MNS服务所在地域。	STRING	是	无	例如`cn-hangzhou`。
mns.queue-name	MNS队列名称。	STRING	是	无	用于消费OSS事件通知的MNS队列名称。
access-key-id	阿里云AccessKey ID。	STRING	是	无	无
access-key-secret	阿里云AccessKey Secret。	STRING	是	无	无

启动模式参数

参数	说明	类型	是否必填	默认值	备注
scan.startup.mode	消费启动模式。	STRING	否	MNS模式下为COMMITTED_OFFSET； POLLING模式下为 LATEST_OFFSET。	取值如下： `EARLIEST_OFFSET`：从MNS队列最早可用消息开始消费。 `LATEST_OFFSET`：从最新消息开始消费。 `COMMITTED_OFFSET`（默认）：从上次提交的消费位点开始消费，若无已提交位点则等同于`LATEST_OFFSET`。 `TIMESTAMP`：从指定时间戳开始消费，需配合`scan.startup.timestamp-millis`使用。 `INITIAL`：先全量扫描OSS指定路径，再切换为增量消费MNS事件。 `SNAPSHOT`：仅全量扫描OSS指定路径，不消费增量事件。重要 POLLING模式不支持EARLIEST_OFFSET和COMMITTED_OFFSET。
scan.startup.timestamp-millis	启动消费的时间戳（毫秒）。	LONG	`scan.startup.mode`为`TIMESTAMP`时必填。	无	仅在`scan.startup.mode`为`TIMESTAMP`时生效。
scan.read-changelog-as-append-only.enabled	是否将增量阶段的变更流转换为Append Only模式下发。	BOOLEAN	否	false	开启此参数时，所有事件都将被作为INSERT（+I）下发。

消费调优参数

参数	说明	类型	是否必填	默认值	备注
mns.batch-size	单次批量拉取MNS消息的最大条数。	INTEGER	否	1	取值范围：1~16。
mns.polling-wait-time	MNS长轮询等待时间。	DURATION	否	10s	取值范围：0~30秒。
mns.delete-max-retries	MNS消息删除失败时的最大重试次数。	INTEGER	否	3	无
scan.rate-limit.records-per-second	每秒最大输出记录数。	DOUBLE	否	无	用于限制源表读取速率。不设置则不限速。
scan.parallelism	源表并行度。	INTEGER	否	无	不设置则使用作业默认并行度。
mns.use-vpc-endpoint.enabled	是否使用内网VPC地址调用MNS控制面API。	BOOLEAN	否	false	无
mns.batch-delete.enabled	在确认MNS消息时，是否攒批提交。	BOOLEAN	否	false	开启此参数可以提升消息消费吞吐，减少Checkpoint耗时。
polling.interval	两次OSS对象元数据扫描之间的间隔。	DURATION	否	10s	仅适用于POLLING模式，必须至少为1毫秒。
polling.prefix-size	每个查询前缀从一次缓存扫描中分批输出的最大对象数。	INTEGER	否	100	仅适用于POLLING模式，必须为正整数。

OSS访问参数

参数	说明	类型	是否必填	默认值	备注
oss-endpoint	OSS服务Endpoint。	STRING	否	无	SNAPSHOT、INITIAL和POLLING模式下必填；启用预签名URL时也必填。
oss-region	OSS服务地域。	STRING	否	无	若未设置，则回退至`region`参数值。启用V4签名时必须配置。
oss-bucket	OSS Bucket名称。	STRING	否	无	SNAPSHOT和INITIAL模式下必填。配置`path`时也为必填。
oss-enable-v4-signature	是否启用OSS V4签名。	BOOLEAN	否	false	启用时必须配置`oss-region`（或`region`）。
path	OSS对象路径前缀。	STRING	否	无	多个前缀以英文逗号（`,`）分隔。用于SNAPSHOT和INITIAL模式下过滤扫描范围。
oss-objects.pre-sign.enabled	是否在OSS Source中生成预签名的URL，可用于免身份认证下载。	BOOLEAN	否	false	开启此参数后，`url`字段会填充下载链接。
oss-objects.pre-sign.ttl	生成预签名的有效期。	DURATION	否	1h	仅当开启oss-objects.pre-sign.enabled时有效。

元数据

OSS CDC使用固定Schema，DDL中声明的列必须为以下支持的列。

OSS事件字段	DDL列名	Flink SQL类型	是否必填	说明
对象Key	`key`	STRING	是（NOT NULL）	OSS对象的唯一标识，即对象路径。
对象完整URL	`url`	STRING	否	OSS URL，格式如`oss://my-bucket/some/file.bin`，可直接用于`FETCH_CONTENT`函数获取文件内容。
所在地域	`region`	STRING	否	OSS Bucket所在地域。
Bucket名称	`bucket`	STRING	否	OSS Bucket名称。
文件名	`fileName`	STRING	否	对象文件名。
文件大小	`size`	BIGINT	否	对象大小，单位：字节。
修改时间	`modificationTime`	TIMESTAMP_LTZ(3)	否	对象最后修改时间。
ETag	`eTag`	STRING	否	对象的ETag值。

自实时计算引擎 VVR 11.9 之后，您可以在建表DDL中通过METADATA FROM语法声明需要读取的扩展元数据列。Flink SQL不需要配置metadata.list。

说明

这些扩展事件元数据仅在MNS采集模式的增量阶段可用，POLLING模式不提供MNS事件上下文。

OSS事件字段	DDL列名	字段类型
事件类型	`eventName`	STRING
事件源	`eventSource`	STRING
事件产生时间	`eventTime`	TIMESTAMP_LTZ(3)
事件协议版本	`eventVersion`	STRING
Bucket 所在地域	`region`	STRING
Bucket 唯一标识符	`ossBucketArn`	STRING
Bucket 名称。	`ossBucketName`	STRING
创建 Bucket 的用户 ID	`ossBucketOwnerIdentity`	STRING
Object 名称	`ossObjectKey`	STRING
Object 大小	`ossObjectSize`	BIGINT
Object ETag 值，用于校验内容是否变化	`ossObjectETag`	STRING
Object 大小变化量	`ossObjectDeltaSize`	BIGINT
文件开始读取的位置	`ossObjectReadFrom`	BIGINT
文件最后读取的位置	`ossObjectReadTo`	BIGINT
OSS Schema 版本号	`ossOssSchemaVersion`	STRING
事件匹配的规则 ID	`ossRuleId`	STRING
请求源 IP	`requestParametersSourceIPAddress`	STRING
请求对应的 Request ID	`responseElementsRequestId`	STRING
请求发起者 UID	`userIdentityPrincipalId`	STRING

DDL中只需声明业务需要的列，未声明的列不会被读取。列名匹配不区分大小写。

通过METADATA FROM声明的额外元数据列说明参见读取OSS事件通知。

使用示例

示例一：从MNS队列消费OSS事件（增量模式）

-- OSS CDC源表：消费MNS队列中的OSS事件通知
CREATE TEMPORARY TABLE oss_cdc_source (
  `key` STRING NOT NULL,
  bucket STRING,
  fileName STRING,
  `size` BIGINT
) WITH (
  'connector' = 'oss-cdc',
  'scan.incremental.mode' = 'MNS',
  'mns.endpoint' = 'http://<yourAccountId>.mns.cn-hangzhou.aliyuncs.com',
  'mns.region' = 'cn-hangzhou',
  'mns.queue-name' = '<yourQueueName>',
  'access-key-id' = '...',
  'access-key-secret' = '...'
);

-- 打印到控制台
CREATE TEMPORARY TABLE print_sink (
  `key` STRING,
  bucket STRING,
  fileName STRING,
  `size` BIGINT
) WITH (
  'connector' = 'print'
);

INSERT INTO print_sink
SELECT `key`, bucket, fileName, `size`
FROM oss_cdc_source;

示例二：SNAPSHOT模式全量扫描OSS路径

CREATE TEMPORARY TABLE oss_cdc_snapshot (
  `key` STRING NOT NULL,
  bucket STRING,
  fileName STRING,
  `size` BIGINT,
  modificationTime TIMESTAMP_LTZ(3)
) WITH (
  'connector' = 'oss-cdc',
  'access-key-id' = '...',
  'access-key-secret' = '...',
  'scan.startup.mode' = 'SNAPSHOT',
  'oss-endpoint' = 'https://oss-cn-hangzhou.aliyuncs.com',
  'oss-region' = 'cn-hangzhou',
  'oss-bucket' = '<yourBucketName>',
  'path' = 'data/log/,data/csv/'
);

示例三：INITIAL模式先全量再增量

CREATE TEMPORARY TABLE oss_cdc_initial (
  `key` STRING NOT NULL,
  bucket STRING,
  `size` BIGINT,
  eTag STRING
) WITH (
  'connector' = 'oss-cdc',
  'scan.incremental.mode' = 'MNS',
  'mns.endpoint' = 'http://<yourAccountId>.mns.cn-hangzhou.aliyuncs.com',
  'mns.region' = 'cn-hangzhou',
  'mns.queue-name' = '<yourQueueName>',
  'access-key-id' = '...',
  'access-key-secret' = '...',
  'scan.startup.mode' = 'INITIAL',
  'oss-endpoint' = 'https://oss-cn-hangzhou.aliyuncs.com',
  'oss-region' = 'cn-hangzhou',
  'oss-bucket' = '<yourBucketName>',
  'path' = 'data/'
);

示例四：带MNS调优参数的增量消费

CREATE TEMPORARY TABLE oss_cdc_tuned (
  `key` STRING NOT NULL,
  bucket STRING,
  fileName STRING,
  `size` BIGINT
) WITH (
  'connector' = 'oss-cdc',
  'scan.incremental.mode' = 'MNS',
  'mns.endpoint' = '...',
  'mns.region' = 'cn-hangzhou',
  'mns.queue-name' = '...',
  'access-key-id' = '...',
  'access-key-secret' = '...',
  'mns.batch-size' = '8',
  'mns.polling-wait-time' = '5s',
  'scan.parallelism' = '4'
);

示例五：使用FETCH_CONTENT函数获取文件内容

CREATE TEMPORARY TABLE oss_cdc_source (
  `key` STRING NOT NULL,
  `url` STRING,
  bucket STRING,
  `size` BIGINT,
  eTag STRING
) WITH (
  'connector' = 'oss-cdc',
  'scan.incremental.mode' = 'MNS',
  'mns.endpoint' = '...',
  'mns.region' = 'cn-hangzhou',
  'mns.queue-name' = '...',
  'access-key-id' = '...',
  'access-key-secret' = '...',
  'scan.startup.mode' = 'INITIAL',
  'oss-endpoint' = 'oss-cn-hangzhou.aliyuncs.com',
  'oss-region' = 'cn-hangzhou',
  'oss-bucket' = '<yourBucketName>',
  'path' = 'images/'
);

CREATE TEMPORARY TABLE print_sink (
  `key` STRING,
  bucket STRING,
  `size` BIGINT,
  eTag STRING,
  `value` BYTES
) WITH (
  'connector' = 'print'
);

INSERT INTO print_sink
SELECT `key`, bucket, `size`, eTag, FETCH_CONTENT(url) AS `value`
FROM oss_cdc_source;

示例六：使用POLLING模式监听OSS变更示例

CREATE TEMPORARY TABLE oss_cdc_polling (
  `key` STRING NOT NULL,
  bucket STRING,
  fileName STRING,
  `size` BIGINT,
  modificationTime TIMESTAMP_LTZ(3),
  eTag STRING,
  PRIMARY KEY (`key`) NOT ENFORCED
) WITH (
  'connector' = 'oss-cdc',
  'scan.incremental.mode' = 'POLLING',
  'scan.startup.mode' = 'INITIAL',
  'access-key-id' = '...',
  'access-key-secret' = '...',
  'oss-endpoint' = 'https://oss-cn-hangzhou.aliyuncs.com',
  'oss-region' = 'cn-hangzhou',
  'oss-bucket' = '<yourBucketName>',
  'path' = 'data/',
  'polling.interval' = '10s',
  'polling.prefix-size' = '100'
);

POLLING模式无需配置MNS Endpoint、地域或队列。默认情况下，全量阶段输出INSERT，后续发现的对象变化输出UPDATE_AFTER；POLLING模式不产生对象删除事件。

使用元数据列

在Flink SQL中，通过METADATA FROM语法将OSS元数据声明为表字段。无需在WITH参数中配置metadata.list。

元数据字段名与元数据键名相同时，可以省略FROM。如果希望使用不同的DDL列名，可以通过METADATA FROM指定原始元数据键名：

以下示例读取OSS事件类型、事件时间、对象Key、对象大小及MNS请求ID：

CREATE TEMPORARY TABLE oss_cdc_metadata_source (
  `key` STRING NOT NULL,
  bucket STRING,
  fileName STRING,
  `size` BIGINT,
  event_type STRING METADATA FROM 'eventName' VIRTUAL,
  event_time TIMESTAMP_LTZ(3) METADATA FROM 'eventTime' VIRTUAL,
  object_key STRING METADATA FROM 'ossObjectKey' VIRTUAL,
  object_size BIGINT METADATA FROM 'ossObjectSize' VIRTUAL,
  request_id STRING METADATA FROM 'responseElementsRequestId' VIRTUAL
) WITH (
  'connector' = 'oss-cdc',
  'scan.incremental.mode' = 'MNS',
  'scan.startup.mode' = 'LATEST_OFFSET',
  'mns.endpoint' = 'http://<yourAccountId>.mns.cn-hangzhou.aliyuncs.com',
  'mns.region' = 'cn-hangzhou',
  'mns.queue-name' = '<yourQueueName>',
  'access-key-id' = '...',
  'access-key-secret' = '...'
);

Flink SQL完整入湖示例

以下示例展示通过Flink SQL将OSS变更数据写入Paimon Blob存储的完整流程。

步骤一：创建Paimon目标表

BLOB列在建表时需通过blob-field和blob-as-descriptor选项指定。详情请参见DLF Paimon Blob存储。

CREATE TABLE my_catalog.my_db.image_table (
    key STRING,
    size BIGINT,
    eTag STRING,
    image BYTES,
    embedding ARRAY<FLOAT>
) WITH (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'blob-field' = 'image',
    'blob-as-descriptor' = 'true'
);

步骤二：创建OSS CDC源表

CREATE TEMPORARY TABLE oss_cdc_source (
  `key` STRING NOT NULL,
  `url` STRING,
  bucket STRING,
  `size` BIGINT,
  eTag STRING
) WITH (
  'connector' = 'oss-cdc',
  'scan.incremental.mode' = 'MNS',
  'mns.endpoint' = '...',
  'mns.region' = 'cn-hangzhou',
  'mns.queue-name' = '...',
  'access-key-id' = '...',
  'access-key-secret' = '...',
  'scan.startup.mode' = 'INITIAL',
  'oss-region' = 'cn-hangzhou',
  'oss-endpoint' = 'oss-cn-hangzhou.aliyuncs.com',
  'oss-bucket' = '<yourBucketName>',
  'path' = 'images/'
);

步骤三：写入Paimon表

通过dlf_catalog.sys.path_to_descriptor内置过程将URL转换为Blob描述符写入。

INSERT INTO dlf_catalog.my_db.image_table
SELECT
  `key`, `size`, eTag, dlf_catalog.sys.path_to_descriptor(url) AS image
FROM
  oss_cdc_source;

数据摄入

语法结构

source:
  type: oss-cdc
  name: OSS CDC Source
  scan.incremental.mode: MNS
  mns.endpoint: http://<yourAccountId>.mns.cn-hangzhou.aliyuncs.com
  mns.region: cn-hangzhou
  mns.queue-name: <yourQueueName>
  access-key-id: ...
  access-key-secret: ...

transform:
  - source-table: <yourBucketName>
    projection: <columns>
    table-options: <tableOptions>

route:
  - source-table: <yourBucketName>
    sink-table: <sinkDatabase>.<sinkTable>

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: <dlfEndpoint>
  catalog.properties.warehouse: <warehouseName>
  catalog.properties.token.provider: dlf

配置项

以下为OSS CDC连接器在数据摄入YAML中的配置项。type固定填写oss-cdc。

通用参数

参数	说明	类型	是否必填	默认值	备注
type	连接器类型。	STRING	是	无	固定填写`oss-cdc`。
scan.incremental.mode	增量发现模式。	STRING	否	MNS	可选值为`MNS`和`POLLING`。
mns.endpoint	MNS数据面服务Endpoint。	STRING	是	无	格式：`http://{account-id}.mns.{region}.aliyuncs.com`。
mns.region	MNS服务所在地域。	STRING	是	无	例如`cn-hangzhou`。
mns.queue-name	MNS队列名称。	STRING	是	无	用于消费OSS事件通知的MNS队列名称。
access-key-id	阿里云AccessKey ID。	STRING	是	无	无
access-key-secret	阿里云AccessKey Secret。	STRING	是	无	无
oss-key.to.table-id.matching.pattern	根据完整OSS对象路径生成源表TableId的正则表达式。	STRING	否	无	匹配`oss://<bucket>/<key>`。支持1～3个捕获组，分别生成table、namespace.table或namespace.schema.table；不匹配的对象会被过滤。

启动模式参数

参数	说明	类型	是否必填	默认值	备注
scan.startup.mode	消费启动模式。	STRING	否	MNS模式为COMMITTED_OFFSET； POLLING模式为LATEST_OFFSET。	取值如下： `EARLIEST_OFFSET`：从MNS队列最早可用消息开始消费。 `LATEST_OFFSET`：从最新消息开始消费。 `COMMITTED_OFFSET`（默认）：从上次提交的消费位点开始消费，若无已提交位点则等同于`LATEST_OFFSET`。 `TIMESTAMP`：从指定时间戳开始消费，需配合`scan.startup.timestamp-millis`使用。 `INITIAL`：先全量扫描OSS指定路径，再切换为增量消费MNS事件。 `SNAPSHOT`：仅全量扫描OSS指定路径，不消费增量事件。重要 POLLING模式不支持EARLIEST_OFFSET及COMMITTED_OFFSET。
scan.startup.timestamp-millis	启动消费的时间戳（毫秒）。	LONG	否	无	仅在`scan.startup.mode`为`TIMESTAMP`时生效。
scan.read-changelog-as-append-only.enabled	是否将增量阶段的变更流转换为Append Only模式下发。	BOOLEAN	否	false	开启此参数时，所有事件都将被作为INSERT（+I）下发。

消费调优参数

参数	说明	类型	是否必填	默认值	备注
mns.batch-size	单次批量拉取MNS消息的最大条数。	INTEGER	否	1	取值范围：1~16。
mns.polling-wait-time	MNS长轮询等待时间。	DURATION	否	10s	取值范围：0~30秒。
mns.delete-max-retries	MNS消息删除失败时的最大重试次数。	INTEGER	否	3	无
scan.rate-limit.records-per-second	每秒最大输出记录数。	DOUBLE	否	无	用于限制源表读取速率。不设置则不限速。
scan.parallelism	源表并行度。	INTEGER	否	无	不设置则使用作业默认并行度。
mns.use-vpc-endpoint.enabled	是否使用内网VPC地址调用MNS API。	BOOLEAN	否	false	无
mns.batch-delete.enabled	在确认MNS消息时，是否攒批提交。	BOOLEAN	否	false	开启此参数可以提升消息消费吞吐，减少Checkpoint耗时。
polling.interval	两次OSS对象元数据扫描之间的间隔。	DURATION	否	10s	仅适用于POLLING模式，必须至少为1毫秒。
polling.prefix-size	每个查询前缀从一次缓存扫描中分批输出的最大对象数。	INTEGER	否	100	仅适用于POLLING模式，必须为正整数。

OSS访问参数

参数	说明	类型	是否必填	默认值	备注
oss-endpoint	OSS服务Endpoint。	STRING	否	无	SNAPSHOT和INITIAL模式下必填。
oss-region	OSS服务地域。	STRING	否	无	若未设置，则回退至`region`参数值。启用V4签名时必须配置。
oss-bucket	OSS Bucket名称。	STRING	否	无	SNAPSHOT和INITIAL模式下必填。配置`path`时也为必填。
oss-enable-v4-signature	是否启用OSS V4签名。	BOOLEAN	否	false	启用时必须配置`oss-region`（或`region`）。
path	OSS对象路径前缀。	STRING	否	无	多个前缀以英文逗号（`,`）分隔。用于SNAPSHOT和INITIAL模式下过滤扫描范围。
metadata.list	需要暴露给下游的元数据列，多个元数据列使用逗号分隔。	STRING	否	无	可选的元数据列包括： eventName eventSource eventTime eventVersion ossBucketName ossBucketArn ossBucketOwnerIdentity ossObjectKey ossObjectSize ossObjectETag ossObjectDeltaSize ossObjectReadFrom ossObjectReadTo ossOssSchemaVersion ossRuleId requestParametersSourceIPAddress responseElementsRequestId userIdentityPrincipalId
oss-objects.pre-sign.enabled	是否在OSS Source中生成预签名的URL，可用于免身份认证下载。	BOOLEAN	否	false	开启此参数后，`url`字段会填充下载链接。
oss-objects.pre-sign.ttl	生成预签名的有效期。	DURATION	否	1h	仅当开启oss-objects.pre-sign.enabled时有效。

元数据

在 Flink CDC 数据摄入作业中，元数据分为两类：

Flink CDC 框架支持的元数据列：由 Flink CDC框架根据每条事件的属性自动推断而来，不需要在source模块配置 metadata.list，参考 Flink CDC 的元数据列（Metadata Column）。
数据源支持的元数据列：从数据源端变更事件中读取，需要在source 模块配置 metadata.list，不同数据源支持的元数据不同，OSS CDC 支持的元数据如下，声明多个元数据时用逗号分隔。

OSS事件字段	DDL列名	字段类型
对象Key	`key`	STRING
对象完整URL	`url`	STRING
所在地域	`region`	STRING
Bucket名称	`bucket`	STRING
文件名	`fileName`	STRING
文件大小	`size`	BIGINT
修改时间	`modificationTime`	TIMESTAMP_LTZ(3)
ETag	`eTag`	STRING
事件类型	`eventName`	STRING
事件源	`eventSource`	STRING
事件产生时间	`eventTime`	TIMESTAMP_LTZ(3)
事件协议版本	`eventVersion`	STRING
Bucket 所在地域	`region`	STRING
Bucket 唯一标识符	`ossBucketArn`	STRING
Bucket 名称。	`ossBucketName`	STRING
创建 Bucket 的用户 ID	`ossBucketOwnerIdentity`	STRING
Object 名称	`ossObjectKey`	STRING
Object 大小	`ossObjectSize`	BIGINT
Object ETag 值，用于校验内容是否变化	`ossObjectETag`	STRING
Object 大小变化量	`ossObjectDeltaSize`	BIGINT
文件开始读取的位置	`ossObjectReadFrom`	BIGINT
文件最后读取的位置	`ossObjectReadTo`	BIGINT
OSS Schema 版本号	`ossOssSchemaVersion`	STRING
事件匹配的规则 ID	`ossRuleId`	STRING
请求源 IP	`requestParametersSourceIPAddress`	STRING
请求对应的 Request ID	`responseElementsRequestId`	STRING
请求发起者 UID	`userIdentityPrincipalId`	STRING

使用示例

文件元数据入湖

以下示例仅将OSS元数据的引用写入Paimon表，不会拷贝OSS文件的实际内容。

source:
  type: oss-cdc
  name: OSS CDC Source
  scan.incremental.mode: MNS
  mns.endpoint: http://<yourAccountId>.mns.cn-hangzhou.aliyuncs.com
  mns.region: cn-hangzhou
  mns.queue-name: <yourQueueName>
  access-key-id: ...
  access-key-secret: ...

transform:
  - source-table: <yourBucketName>
    projection: key, url, fileName, eTag
    table-options: blob-field=url;blob-descriptor-field=url;blob-as-descriptor=true;row-tracking.enabled=true;data-evolution.enabled=true

route:
  - source-table: <yourBucketName>
    sink-table: paimon_db.paimon_table

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: http://cn-shenzhen-vpc.dlf.aliyuncs.com
  catalog.properties.warehouse: dlf_paimon
  catalog.properties.token.provider: dlf

文件元数据和内容同时入湖

以下示例通过FETCH_CONTENT函数获取文件内容，将OSS文件内容和元数据一并写入Paimon表。（替换对应的OSS Bucket和AK/SK）

source:
  type: oss-cdc
  name: OSS CDC Source
  scan.incremental.mode: MNS
  mns.endpoint: http://<yourAccountId>.mns.cn-hangzhou.aliyuncs.com
  mns.region: cn-hangzhou
  mns.queue-name: <yourQueueName>
  access-key-id: ...
  access-key-secret: ...

transform:
  - source-table: <yourBucketName>
    projection: key, url, fileName, eTag, FETCH_CONTENT(`url`) as blob_field
    table-options: blob-field=url,blob_field;blob-descriptor-field=url;blob-as-descriptor=true;row-tracking.enabled=true;data-evolution.enabled=true

route:
  - source-table: <yourBucketName>
    sink-table: paimon_db.paimon_table

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: http://cn-shenzhen-vpc.dlf.aliyuncs.com
  catalog.properties.warehouse: dlf_paimon
  catalog.properties.token.provider: dlf

通过POLLING模式发现OSS对象变化

source:
  type: oss-cdc
  name: OSS CDC Polling Source
  scan.incremental.mode: POLLING
  scan.startup.mode: INITIAL
  access-key-id: ...
  access-key-secret: ...
  oss-endpoint: https://oss-cn-hangzhou.aliyuncs.com
  oss-region: cn-hangzhou
  oss-bucket: <yourBucketName>
  path: data/
  polling.interval: 10s
  polling.prefix-size: 100

transform:
  - source-table: <yourBucketName>
    projection: key, url, fileName, eTag

route:
  - source-table: <yourBucketName>
    sink-table: paimon_db.paimon_table

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: <dlfEndpoint>
  catalog.properties.warehouse: <warehouseName>
  catalog.properties.token.provider: dlf