DocsICP FilingConsole
Documentation
Home Page Realtime Compute for Apache Flink Streaming Compute Flink Developer Reference Supported connectors OSS CDC (Beta)

OSS CDC (Beta)

Updated at:
Copy as MD
Product
My Favorites

The OSS CDC connector continuously detects data changes in Object Storage Service (OSS) based on event notifications of Message Service (MNS) or periodic polling of OSS object metadata. In MNS mode, the connector consumes events such as object creation and deletion. In POLLING mode, the connector discovers new or changed objects by periodic scans without MNS configurations. The connector also supports full scans of object metadata under specified paths. This topic describes how to use the OSS CDC connector.

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: MNS mode consumes OSS event notifications in real time from an MNS queue. This mode has low latency and does not poll OSS API operations.

  • Polling mode: POLLING mode does not require an MNS queue or OSS event notifications. It discovers new and changed objects by periodically scanning OSS object metadata.

  • Full and incremental: SNAPSHOT mode performs a full scan of the objects under specified paths. INITIAL mode first outputs existing objects, and then continuously detects incremental changes by using MNS or POLLING mode.

  • Path filtering: Incremental reads in SNAPSHOT, INITIAL, and POLLING modes support the path parameter, which specifies multiple path prefixes so that only the specified directories are scanned.

  • Rate limiting: Supports the scan.rate-limit.records-per-second parameter to limit the read rate and prevent excessive pressure on downstream systems.

  • Flexible startup: MNS mode supports EARLIEST_OFFSET, LATEST_OFFSET, COMMITTED_OFFSET, TIMESTAMP, INITIAL, and SNAPSHOT. POLLING mode supports LATEST_OFFSET, TIMESTAMP, INITIAL, and SNAPSHOT. You can select a startup mode based on your business scenario.

Prerequisites

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

Configuration

Requirement

OSS event notifications (MNS mode)

When you use MNS incremental mode, you must configure event notifications on the OSS bucket to push events such as ObjectCreated and ObjectRemove to an MNS queue. OSS event notifications are not required in POLLING or SNAPSHOT mode.

MNS queue (MNS mode)

When you use MNS incremental mode, you must create an MNS queue and make sure that the queue is connected to the network of the Flink job. MNS queues are not required in POLLING or SNAPSHOT mode.

Note

An MNS queue is required only when you use MNS mode and an incremental startup mode.

AccessKey

An Alibaba Cloud AccessKey is created. MNS mode requires the corresponding MNS permissions. SNAPSHOT, INITIAL, and POLLING modes require the corresponding OSS permissions.

OSS V4 Signature (Optional)

If V4 signing is enabled for OSS, set oss-enable-v4-signature to true and configure oss-region.

In MNS mode, for more information about how to configure event notifications, see OSS event notifications.

Limits

  • The OSS CDC connector is supported only in Realtime Compute engine VVR 11.9.0 and later.

  • POLLING mode is supported only in VVR 11.9.0 and later.

  • Only source tables are supported. Dimension tables and result tables are not supported.

  • The columns defined in the DDL statement must be the columns supported by the fixed schema. For more information, see Metadata. Custom columns are not supported.

  • Watermark definition is not supported for source tables.

  • In MNS mode, messages are retained in the MNS queue for a limited period of time (the message retention period ranges from 1 minute to 7 days, and the default value is 4 days). If a job is suspended for an extended period and the messages expire, events may be lost. POLLING mode does not depend on the MNS message retention period.

  • The full phase of SNAPSHOT and INITIAL modes and the continuous incremental phase of POLLING mode all rely on the OSS ListObjects API operation. If a bucket contains a large number of objects, the scan may require a long period of time and generate a large number of OSS API calls.

  • POLLING mode discovers new and changed objects by scanning object metadata, and does not generate object deletion events. To synchronize deletion events, use MNS mode and configure ObjectRemove event notifications.

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, INITIAL, or POLLING mode, you must configure oss-endpoint and oss-bucket and specify a valid OSS region. When you use POLLING incremental mode, you must explicitly configure oss-region. If you configure path, oss-bucket is required.

  • When you use the TIMESTAMP startup mode, you must set scan.startup.timestamp-millis to a positive epoch value in milliseconds. Starting from a timestamp is not supported in POLLING incremental mode.

  • In MNS mode, set mns.batch-size and mns.polling-wait-time based on your business requirements. In POLLING mode, set polling.interval and polling.prefix-size to balance the throughput, scan frequency, and latency.

Preparations

Configure the MNS Queue

This section applies only to incremental jobs in which scan.incremental.mode is set to MNS. Skip this section if you use POLLING mode or only SNAPSHOT mode.

  1. Log on to the OSS console and create or select the bucket you want to monitor.

  2. Log on to the MNS console. In the left-side navigation pane, click Queue Model > Queue List to create a queue.

    Note

    • We recommend that you select the Standard Queue mode instead of the Ordered Queue mode.

    • We recommend that you set the Visibility Timeout parameter to 2 to 3 times the Checkpoint interval to avoid duplicate message delivery.

  3. In the left-side navigation pane of the MNS console, click Event Notification > Object Storage Service OSS, and then click Create Rule.

  4. In the dialog box, enter a rule name and select the event actions to be monitored. Select only the create, copy, and delete events. Do not select file download events. Otherwise, event loops may occur. Deselect InitiateMultipartUpload, UploadPart, and UploadPartCopy, which are related to multipart upload, and retain only the CompleteMultipartUpload event. Otherwise, data may be duplicated.

  5. 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/.

  6. After the queue is created, return to Queue Model > Queues. In Endpoints, copy the Public Access Endpoint address and configure it as the value of the connector mns.endpoint.

Configure POLLING mode

POLLING mode does not require an MNS queue or OSS event notifications. Make sure that the following requirements are met:

  • The Flink job can access oss-endpoint.

  • The AccessKey has the permissions to list OSS objects and read the metadata of OSS objects.

  • The oss-bucket, oss-region, and optional path to be scanned are specified.

  • Configure polling.interval and polling.prefix-size based on the number of objects in the bucket.

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',
  'scan.incremental.mode' = 'MNS',
  'mns.endpoint' = '<yourMnsEndpoint>',
  'mns.region' = '<yourMnsRegion>',
  'mns.queue-name' = '<yourMnsQueueName>',
  'access-key-id' = '...',
  'access-key-secret' = '...'
);

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. connector Set 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.

scan.incremental.mode

The incremental discovery mode.

STRING

No

MNS

Valid values: MNS and POLLING.

  • MNS: consumes event notifications.

  • POLLING: periodically scans OSS object metadata.

mns.endpoint

The MNS data service endpoint.

STRING

Yes

N/A

Format: http://{account-id}.mns.{region}.aliyuncs.com.

mns.region

The region where the MNS service is located.

STRING

Yes

N/A

Example: cn-hangzhou.

mns.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

  • In MNS mode, the value is COMMITTED_OFFSET.

  • In POLLING mode, the value is LATEST_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.

Important

POLLING mode does not support EARLIEST_OFFSET or COMMITTED_OFFSET.

scan.startup.timestamp-millis

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

LONG

Required if scan.startup.mode is set to TIMESTAMP.

N/A

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

scan.read-changelog-as-append-only.enabled

Specifies whether to convert the change stream of the incremental phase into Append Only mode.

BOOLEAN

No

false

When this parameter is enabled, all events are delivered as INSERT (+I) messages.

Tuning Parameters

Parameter

Description

Type

Required

Default Value

Remarks

mns.batch-size

The maximum number of MNS messages to pull in a single batch.

INTEGER

No

1

Valid values: 1 to 16.

mns.polling-wait-time

The long polling wait time for MNS.

DURATION

No

10s

Valid values: 0 to 30 seconds.

mns.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.

mns.use-vpc-endpoint.enabled

Specifies whether to use an internal VPC endpoint to call MNS API operations for the control plane.

BOOLEAN

No

false

N/A

mns.batch-delete.enabled

Specifies whether to acknowledge MNS messages in batches.

BOOLEAN

No

false

Enabling this parameter improves message consumption throughput and reduces checkpoint durations.

polling.interval

The interval between two consecutive scans of OSS object metadata.

DURATION

No

10s

Applies only to POLLING mode. The value must be at least 1 ms.

polling.prefix-size

The maximum number of objects that are output in batches from a single cached scan for each query prefix.

INTEGER

No

100

Applies only to POLLING mode. The value must be a positive integer.

mns.eager-delete.enabled

Specifies whether to acknowledge and delete MNS messages in advance.

BOOLEAN

No

false

Enable this parameter only if you created an MNS ordered (FIFO) queue.

An MNS ordered queue makes a subsequent message visible only after the previous message is committed. This parameter acknowledges messages in a timely manner between checkpoints and prevents consumption latency.

OSS access parameters

Parameter

Description

Type

Required

Default Value

Remarks

oss-endpoint

The OSS service endpoint.

STRING

No

N/A

Required in SNAPSHOT, INITIAL, and POLLING modes. This parameter is also required if presigned URLs are enabled.

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.

oss-objects.pre-sign.enabled

Specifies whether to generate pre-signed URLs in the OSS source. Pre-signed URLs allow downloads without identity verification.

BOOLEAN

No

false

After this parameter is enabled, url field is populated with download links.

oss-objects.pre-sign.ttl

The validity period of generated pre-signed URLs.

DURATION

No

1h

This parameter takes effect only when oss-objects.pre-sign.enabled is enabled.

Metadata

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.

In Realtime Compute engine VVR 11.9 and later, you can declare the extended metadata columns to be read by using the METADATA FROM syntax in the table DDL statement. You do not need to configure metadata.list for Flink SQL.

Note

The extended event metadata is available only in the incremental phase of MNS mode. POLLING mode does not provide MNS event contexts.

OSS Event Field

DDL Column

Data Type

Event Type

eventName

STRING

Event Source

eventSource

STRING

Event Time

eventTime

TIMESTAMP_LTZ(3)

Event Protocol Version

eventVersion

STRING

Region of the Bucket

region

STRING

Unique Identifier of the Bucket

ossBucketArn

STRING

Bucket Name

ossBucketName

STRING

User ID of the Bucket Creator

ossBucketOwnerIdentity

STRING

Object Name

ossObjectKey

STRING

Object Size

ossObjectSize

BIGINT

Object ETag value, used to verify whether the content has changed

ossObjectETag

STRING

Object Size Delta

ossObjectDeltaSize

BIGINT

File Read Start Position

ossObjectReadFrom

BIGINT

File Read End Position

ossObjectReadTo

BIGINT

OSS Schema Version

ossOssSchemaVersion

STRING

Matched Event Rule ID

ossRuleId

STRING

Source IP Address of the Request

requestParametersSourceIPAddress

STRING

Request ID

responseElementsRequestId

STRING

UID of the Request Initiator

userIdentityPrincipalId

STRING

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.

For more information about the extra metadata columns declared by using METADATA FROM, see Read OSS event notifications.

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',
  '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' = '...'
);

-- 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',
  '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/'
);

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',
  '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/'
);

Example 4: incremental consumption with MNS tuning parameters

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'
);

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',
  '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;

Example 6: use POLLING mode to listen for OSS changes

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 mode does not require the MNS endpoint, region, or queue. By default, INSERT records are output in the full phase and UPDATE_AFTER records are output for object changes detected afterwards. POLLING mode does not generate object deletion events.

Use metadata columns

In Flink SQL, declare OSS metadata as table fields by using the METADATA FROM syntax. You do not need to configure metadata.list in the WITH parameters.

If the metadata field name is the same as the metadata key name, you can omit FROM. To use a different DDL column name, specify the original metadata key name by using METADATA FROM:

The following example reads the OSS event type, event time, object key, object size, and MNS request 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' = '...'
);

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 ARRAY<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',
  '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/'
);

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

Syntax

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

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

Configuration Parameters

The following sections describe the configuration parameters for the OSS CDC connector in CDC YAML data ingestion. type Set the value to oss-cdc.

General Parameters

Parameter

Description

Type

Required

Default Value

Remarks

type

Connector type.

STRING

Yes

N/A

Set the value to oss-cdc.

scan.incremental.mode

The incremental discovery mode.

STRING

No

MNS

Valid values: MNS and POLLING.

mns.endpoint

The MNS data service endpoint.

STRING

Yes

N/A

Format: http://{account-id}.mns.{region}.aliyuncs.com.

mns.region

The region where the MNS service is located.

STRING

Yes

N/A

Example: cn-hangzhou.

mns.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

oss-key.to.table-id.matching.pattern

The regular expression that is used to generate the TableId of the source table from the full path of an OSS object.

STRING

No

None

Matches oss://<bucket>/<key>. The regular expression supports one to three capture groups, which generate table, namespace.table, or namespace.schema.table. Objects that do not match the regular expression are filtered out.

Startup Mode Parameters

Parameter

Description

Type

Required

Default Value

Remarks

scan.startup.mode

The startup mode for consumption.

STRING

No

  • In MNS mode, the value is COMMITTED_OFFSET.

  • In POLLING mode, the value is LATEST_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 the scan.startup.timestamp-millis parameter.

  • 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.

Important

POLLING mode does not support EARLIEST_OFFSET or COMMITTED_OFFSET.

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.

scan.read-changelog-as-append-only.enabled

Specifies whether to convert the change stream of the incremental phase into Append Only mode.

BOOLEAN

No

false

When this parameter is enabled, all events are delivered as INSERT (+I) messages.

Tuning Parameters

Parameter

Description

Type

Required

Default Value

Remarks

mns.batch-size

The maximum number of MNS messages to pull in a single batch.

INTEGER

No

1

Valid values: 1 to 16.

mns.polling-wait-time

The long polling wait time for MNS.

DURATION

No

10s

Valid values: 0 to 30 seconds.

mns.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.

mns.use-vpc-endpoint.enabled

Specifies whether to call the MNS API by using an internal VPC endpoint.

BOOLEAN

No

false

N/A

mns.batch-delete.enabled

Specifies whether to acknowledge MNS messages in batches.

BOOLEAN

No

false

Enabling this parameter improves message consumption throughput and reduces checkpoint durations.

polling.interval

The interval between two consecutive scans of OSS object metadata.

DURATION

No

10s

Applies only to POLLING mode. The value must be at least 1 ms.

polling.prefix-size

The maximum number of objects that are output in batches from a single cached scan for each query prefix.

INTEGER

No

100

Applies only to POLLING mode. The value must be a positive integer.

mns.eager-delete.enabled

Specifies whether to acknowledge and delete MNS messages in advance.

BOOLEAN

No

false

Enable this parameter only if you created an MNS ordered (FIFO) queue.

An MNS ordered queue makes a subsequent message visible only after the previous message is committed. This parameter acknowledges messages in a timely manner between checkpoints and prevents consumption latency.

OSS access parameters

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.

metadata.list

The metadata columns to expose to downstream operators. Separate multiple metadata columns with commas (,).

STRING

No

N/A

Available metadata columns:

  • eventName

  • eventSource

  • eventTime

  • eventVersion

  • ossBucketName

  • ossBucketArn

  • ossBucketOwnerIdentity

  • ossObjectKey

  • ossObjectSize

  • ossObjectETag

  • ossObjectDeltaSize

  • ossObjectReadFrom

  • ossObjectReadTo

  • ossOssSchemaVersion

  • ossRuleId

  • requestParametersSourceIPAddress

  • responseElementsRequestId

  • userIdentityPrincipalId

oss-objects.pre-sign.enabled

Specifies whether to generate pre-signed URLs in the OSS source. Pre-signed URLs allow downloads without identity verification.

BOOLEAN

No

false

After this parameter is enabled, url field is populated with download links.

oss-objects.pre-sign.ttl

The validity period of generated pre-signed URLs.

DURATION

No

1h

This parameter takes effect only when oss-objects.pre-sign.enabled is enabled.

Metadata

Metadata columns in Flink CDC data ingestion jobs fall into two categories:

  1. Metadata columns supported by the Flink CDC framework: The Flink CDC framework automatically infers these columns from the attributes of each event. You do not need to configure metadata.list in the source module. For more information, see Metadata columns in Flink CDC.

  2. Metadata columns supported by the data source: These columns are read from change events on the data source side. You must configure metadata.list in the source module. Supported metadata varies by data source. The following table lists the metadata supported by OSS CDC. Separate multiple metadata columns with commas.

OSS Event Field

DDL Column

Data Type

Object Key

key

STRING

Full Object URL

url

STRING

Region

region

STRING

Bucket Name

bucket

STRING

File Name

fileName

STRING

File Size

size

BIGINT

Modification Time

modificationTime

TIMESTAMP_LTZ(3)

ETag

eTag

STRING

Event Type

eventName

STRING

Event Source

eventSource

STRING

Event Time

eventTime

TIMESTAMP_LTZ(3)

Event Protocol Version

eventVersion

STRING

Region of the Bucket

region

STRING

Unique Identifier of the Bucket

ossBucketArn

STRING

Bucket Name

ossBucketName

STRING

User ID of the Bucket Creator

ossBucketOwnerIdentity

STRING

Object Name

ossObjectKey

STRING

Object Size

ossObjectSize

BIGINT

Object ETag value, used to verify whether the content has changed

ossObjectETag

STRING

Object Size Delta

ossObjectDeltaSize

BIGINT

File Read Start Position

ossObjectReadFrom

BIGINT

File Read End Position

ossObjectReadTo

BIGINT

OSS Schema Version

ossOssSchemaVersion

STRING

Matched Event Rule ID

ossRuleId

STRING

Source IP Address of the Request

requestParametersSourceIPAddress

STRING

Request ID

responseElementsRequestId

STRING

UID of the Request Initiator

userIdentityPrincipalId

STRING

Examples

File Metadata Lakehouse Ingestion

The following example writes only OSS metadata references to the Paimon table without copying the actual content of OSS files.

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

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

File Metadata and Content Lakehouse Ingestion

The following example uses the FETCH_CONTENT function to retrieve file content and writes both file content and metadata to the Paimon table. Replace the OSS bucket name and AccessKey pair with your own values.

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

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

Discover OSS object changes in POLLING mode

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
Previous：JDBCNext：DingTalk Wiki CDC (Beta)