OSS CDC (Beta)

Background Information

The following table describes the information supported by the OSS CDC connector.

Category	Details
Supported Type	Source table
Execution Mode	Streaming mode only
Data Format	Not applicable (fixed schema)
API	SQL and CDC YAML
Update or Delete Result Table Data	Not applicable

Key Features

Event-driven: Consumes OSS event notifications from MNS queues in real time with low latency and no need to poll OSS APIs.
Full + incremental: Supports SNAPSHOT mode for full scans of objects under specified paths, and INITIAL mode for a full scan followed by incremental consumption, achieving unified full and incremental processing.
Path filtering: SNAPSHOT and INITIAL modes support specifying multiple path prefixes through the pathparameter to scan only target directories.
Rate limiting: Supports the scan.rate-limit.records-per-secondparameter to limit the read rate and prevent excessive pressure on downstream systems.
Flexible startup: Supports six startup modes (EARLIEST_OFFSET, LATEST_OFFSET, COMMITTED_OFFSET, TIMESTAMP, INITIAL, and SNAPSHOT) to meet various business scenarios.

Prerequisites

Before you use the OSS CDC connector, make sure the following conditions are met.

Configuration	Requirement
OSS Event Notifications	Event notifications are configured on the OSS bucket to push events such as ObjectCreated and ObjectRemove to an MNS queue.
MNS Queue (Optional)	An MNS queue is created and accessible from the VPC where the Flink job runs. Note An MNS queue is required only when you use an incremental startup mode.
AccessKey	An Alibaba Cloud AccessKey pair is created with access permissions for both MNS and OSS.
OSS V4 Signature (Optional)	If V4 signing is enabled for OSS, set `oss-enable-v4-signature`to `true` and configure `oss-region`.

For information about how to configure MNS event notifications, see OSS Event Notifications.

Limits

Supported only in Realtime Compute Engine VVR 11.8.0.preview.2 and later.
Only source tables are supported. Dimension tables and result tables are not supported.
Columns defined in the DDL must be from the fixed schema (see Data Type Mapping). Custom columns are not supported.
Watermark definition is not supported for source tables.
Messages in the MNS queue have a limited retention period. If a job is stopped for too long and messages expire, events may be lost.
SNAPSHOT and INITIAL modes depend on the OSS ListObjects API. Full scans of large-scale buckets may take a long time.

Usage Notes

The key field in the DDL must be declared as NOT NULL. This field serves as the unique identifier of the object.
Column name matching is case-insensitive. We recommend that you use the column names defined in the fixed schema.
When you use SNAPSHOT or INITIAL mode, you must configure the oss-endpoint and oss-bucket parameters. If you configure the path parameter, oss-bucketis also required.
When you use the TIMESTAMP startup mode, you must configure the scan.startup.timestamp-millis parameter.
We recommend that you set the batch-size and polling-wait-time parameters based on your business requirements to balance throughput and latency.

Preparations

Configure the MNS Queue

To synchronize incremental file changes from OSS through MNS, complete the following configuration. If you only need a one-time full data synchronization (SNAPSHOT mode), skip this step.

Log on to the OSS console and create or select the bucket you want to monitor.
Log on to the MNS console. In the left-side navigation pane, click Queue Model > Queue List to create a queue. Enter the queue name and keep the default values for other parameters.
In the left-side navigation pane of the MNS console, click Event Notification > Object Storage Service OSS, and then click Create Rule.
In the dialog box, enter a rule name and select the event actions to monitor. We recommend that you select only the Create, Copy, and Delete event types. Do not select the Download File event to avoid event loops.
Select the OSS bucket and matching prefix to monitor. For example, to monitor only file changes in the images/ virtual directory, select Prefix under Matching Rule and enter images/.
After creation, go back to Queue Model > Queue List and copy the Public Access address from the Endpoint column for later use.

Configure OSS Authentication

Regardless of whether you use CDC YAML or Flink SQL jobs, you must add OSS authentication parameters in the Runtime Parameters > Other Configuration section of the job. For more information, see Configure Bucket Authentication.

SQL

Syntax

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',
  'endpoint' = '<yourMnsEndpoint>',
  'region' = '<yourMnsRegion>',
  'queue-name' = '<yourMnsQueueName>',
  'access-key-id' = '<yourAccessKeyId>',
  'access-key-secret' = '<yourAccessKeySecret>'
);

The columns declared in the DDL can be a subset of the fixed schema columns. However, the key column must be declared and set to NOT NULL. Columns outside the fixed schema are not supported.

WITH Parameters

The following sections describe the WITH parameters for the OSS CDC connector in Flink SQL. connectorSet the value to oss-cdc.

General Parameters

Parameter	Description	Type	Required	Default Value	Remarks
connector	Connector type.	STRING	Yes	N/A	Set the value to `oss-cdc`.
endpoint	The MNS data service endpoint.	STRING	Yes	N/A	Format: `http://{account-id}.mns.{region}.aliyuncs.com`.
region	The region where the MNS service is located.	STRING	Yes	N/A	Example: `cn-hangzhou`.
queue-name	The MNS queue name.	STRING	Yes	N/A	The name of the MNS queue used to consume OSS event notifications.
access-key-id	The Alibaba Cloud AccessKey ID.	STRING	Yes	N/A	N/A
access-key-secret	The Alibaba Cloud AccessKey Secret.	STRING	Yes	N/A	N/A

Startup Mode Parameters

Parameter

Description

Type

Required

Default Value

Remarks

scan.startup.mode

The startup mode for consumption.

STRING

No

COMMITTED_OFFSET

Valid values:

EARLIEST_OFFSET: Starts consumption from the earliest available message in the MNS queue.
LATEST_OFFSET: Starts consumption from the latest message.
COMMITTED_OFFSET (default): Starts consumption from the last committed offset. If no committed offset exists, this mode behaves the same as LATEST_OFFSET.
TIMESTAMP: Starts consumption from a specified timestamp. Must be used with scan.startup.timestamp-millis.
INITIAL: Performs a full scan of the specified OSS paths first, then switches to incremental consumption of MNS events.
SNAPSHOT: Performs a full scan of the specified OSS paths only without consuming incremental events.

scan.startup.timestamp-millis

The timestamp (in milliseconds) from which to start consumption.

LONG

No

N/A

Takes effect only when scan.startup.mode is set to TIMESTAMP.

Tuning Parameters

Parameter	Description	Type	Required	Default Value	Remarks
batch-size	The maximum number of MNS messages to pull in a single batch.	INTEGER	No	1	Valid values: 1 to 16.
polling-wait-time	The long polling wait time for MNS.	DURATION	No	10s	Valid values: 0 to 30 seconds.
delete-max-retries	The maximum number of retries when MNS message deletion fails.	INTEGER	No	3	N/A
scan.rate-limit.records-per-second	The maximum number of output records per second.	DOUBLE	No	N/A	Limits the read rate of the source table. No rate limit is applied if this parameter is not set.
scan.parallelism	The parallelism of the source table.	INTEGER	No	N/A	If not set, the default parallelism of the job is used.

OSS Parameters (SNAPSHOT/INITIAL Mode)

Parameter	Description	Type	Required	Default Value	Remarks
oss-endpoint	The OSS service endpoint.	STRING	No	N/A	Required in SNAPSHOT and INITIAL modes.
oss-region	The OSS service region.	STRING	No	N/A	If not set, falls back to the `region` parameter value. Required when V4 signing is enabled.
oss-bucket	The OSS bucket name.	STRING	No	N/A	Required in SNAPSHOT and INITIAL modes. Also required when `path` is configured.
oss-enable-v4-signature	Specifies whether to enable OSS V4 signing.	BOOLEAN	No	false	When enabled, you must configure `oss-region` (or `region`).
path	The OSS object path prefix.	STRING	No	N/A	Separate multiple prefixes with commas (`,`). Used in SNAPSHOT and INITIAL modes to filter the scan range.

Data Type Mapping

The OSS CDC connector uses a fixed schema. Columns declared in the DDL must be from the following supported columns.

OSS Event Field	DDL Column	Flink SQL Type	Required	Description
Object Key	`key`	STRING	Yes (NOT NULL)	The unique identifier of the OSS object, which is the object path.
Full Object URL	`url`	STRING	No	The OSS URL in a format such as `oss://my-bucket/some/file.bin`. You can use this URL directly with the `FETCH_CONTENT` function to retrieve file content.
Region	`region`	STRING	No	The region where the OSS bucket is located.
Bucket Name	`bucket`	STRING	No	The OSS bucket name.
File Name	`fileName`	STRING	No	The object file name.
File Size	`size`	BIGINT	No	The object size in bytes.
Modification Time	`modificationTime`	TIMESTAMP_LTZ(3)	No	The last modification time of the object.
ETag	`eTag`	STRING	No	The ETag value of the object.

You only need to declare the columns required by your business in the DDL. Columns not declared will not be read. Column name matching is case-insensitive.

Examples

Example 1: Consume OSS events from the MNS queue (incremental mode)

-- OSS CDC source table: consumes OSS event notifications from the MNS queue
CREATE TEMPORARY TABLE oss_cdc_source (
  `key` STRING NOT NULL,
  bucket STRING,
  fileName STRING,
  `size` BIGINT
) WITH (
  'connector' = 'oss-cdc',
  'endpoint' = 'http://<yourAccountId>.mns.cn-hangzhou.aliyuncs.com',
  'region' = 'cn-hangzhou',
  'queue-name' = '<yourQueueName>',
  'access-key-id' = '<yourAccessKeyId>',
  'access-key-secret' = '<yourAccessKeySecret>'
);

-- Print to console
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;

Example 2: Full scan of OSS paths in SNAPSHOT mode

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

Example 3: Full scan then incremental in INITIAL mode

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

Example 4: Incremental consumption with tuning parameters

CREATE TEMPORARY TABLE oss_cdc_tuned (
  `key` STRING NOT NULL,
  bucket STRING,
  fileName STRING,
  `size` BIGINT
) WITH (
  'connector' = 'oss-cdc',
  'endpoint' = 'http://<yourAccountId>.mns.cn-hangzhou.aliyuncs.com',
  'region' = 'cn-hangzhou',
  'queue-name' = '<yourQueueName>',
  'access-key-id' = '<yourAccessKeyId>',
  'access-key-secret' = '<yourAccessKeySecret>',
  'batch-size' = '8',
  'polling-wait-time' = '5s',
  'scan.parallelism' = '4'
);

Example 5: Retrieve file content using the FETCH_CONTENT function

CREATE TEMPORARY TABLE oss_cdc_source (
  `key` STRING NOT NULL,
  `url` STRING,
  bucket STRING,
  `size` BIGINT,
  eTag STRING
) WITH (
  'connector' = 'oss-cdc',
  'endpoint' = 'http://<yourAccountId>.mns.cn-hangzhou.aliyuncs.com',
  'region' = 'cn-hangzhou',
  'queue-name' = '<yourQueueName>',
  'access-key-id' = '<yourAccessKeyId>',
  'access-key-secret' = '<yourAccessKeySecret>',
  'scan.startup.mode' = 'INITIAL',
  'oss-endpoint' = 'oss-cn-hangzhou.aliyuncs.com',
  '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;

Complete Flink SQL Lakehouse Ingestion Example

The following example demonstrates the complete workflow of writing OSS change data to Paimon Blob Storage using Flink SQL.

Step 1: Create a Paimon target table

BLOB columns must be specified during table creation through the blob-field and blob-as-descriptor options. For more information, see DLF Paimon Blob Storage.

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

Step 2: Create an OSS CDC source table

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

Step 3: Write to the Paimon table

Use the built-in procedure dlf_catalog.sys.path_to_descriptor to convert URLs to Blob descriptors for writing.

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;

CDC YAML Data Ingestion