SQL Server CDC (Public Preview)-Realtime Compute for Apache Flink(Flink)-阿里云帮助中心

Background information

The connector supports the following features.

Category	Details
Supported type	Source table
Run mode	Only stream mode is supported.
Data format	Not applicable
Connector-specific monitoring metrics	Connector-specific monitoring metrics currentFetchEventTimeLag: The delay between event generation and when the source operator fetches it. currentEmitEventTimeLag: The delay between event generation and when the source operator emits it. sourceIdleTime: The duration the source has been idle without producing data. Note These metrics are 0 during the full phase. For more information about these metrics, see Monitoring Metrics.
API type	SQL
Supports updates or deletions in the result table	Not applicable

Prerequisites

Before using the SQL Server CDC connector, you must enable the Change Data Capture (CDC) feature for the target database and tables.

To enable CDC for a table, run the following SQL statements:

RDS SQL Server

EXEC sp_rds_cdc_enable_db;       -- Enable CDC for the database
EXEC sys.sp_cdc_enable_table
@source_schema = N'dbo',         -- Specify the schema of the source table
@source_name = N'MyTable',       -- Specify the name of the table to capture
@role_name = N'MyRole',          -- Specify the role MyRole
@supports_net_changes = 1;

Self-managed SQL Server

EXEC sys.sp_cdc_enable_table
@source_schema = N'dbo',        -- Specify the schema of the source table
@source_name   = N'MyTable',    -- Specify the name of the table to capture
@role_name     = N'MyRole',     -- Specify the role MyRole. You can add users who require SELECT 
                                -- permissions on the captured columns to this role.
                                -- Members of the sysadmin or db_owner roles can also
                                -- access the specified change tables. If this parameter is NULL, 
                                -- only members of these roles can access the capture data.
@filegroup_name = N'MyDB_CT',   -- Specify the filegroup where SQL Server will place the change
                                -- tables. This filegroup must already exist.
                                -- We recommend placing change tables in a different filegroup from the source tables.
@supports_net_changes = 0;

To verify access permissions for the CDC tables, run the following SQL statement:

EXEC sys.sp_cdc_help_change_data_capture;

This query returns the configuration for each CDC-enabled table in the database. If a table is not listed, verify that CDC is enabled for it and that the user has the required access permissions.

Limitations

This connector is available only for VVR 11.7 and later.
The source database must have CDC enabled.

Because the connector uses the Change Data Capture feature, it has the following SQL Server version requirements:

For Standard Edition, SQL Server 2016 SP1 or later is required.
For Enterprise Edition, SQL Server 2012 or later is required.

SQL

Syntax

CREATE TABLE sqlserver_cdc_source (
    id INT,
    order_date DATE,
    purchaser INT,
    quantity INT,
    product_id INT,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'sqlserver-cdc',
  'hostname'  = '<hostname>',
  'port'      = '<port>',
  'username'  = '<username>',
  'password'  = '<password>',
  'database-name' = '<database name>',
  'table-name'    = 'dbo.orders'
);

WITH parameters

Parameter

Description

Type

Required

Default

Remarks

connector

The connector type.

STRING

Yes

None

The value is fixed at sqlserver-cdc.

hostname

The IP address or hostname of the SQL Server database.

STRING

Yes

None

None.

username

The username to connect to the SQL Server database.

STRING

Yes

None

None.

password

The password for the specified username.

STRING

Yes

None

None.

database-name

The name of the database to monitor.

STRING

Yes

None

None.

table-name

The names of the tables to capture.

STRING

Yes

None

To match multiple tables, you can use , to separate multiple <schemaName>.<tableName> entries, and regular expressions are also supported. For example:

Value	Description
'table-name'='s1.t1,s2.t2'	Includes tables s1.t1 and s2.t2.
'table-name'='s1\..,s2\..'	Includes all tables under schemas s1 and s2.

scan.startup.mode

The startup mode for data consumption.

STRING

No

initial

Valid values:

initial (default): Performs a full snapshot of existing data, then reads the latest change logs.
latest-offset: Skips the snapshot and reads only changes that occurred after the connector started.

Important

The scan.startup.mode option depends on the Debezium snapshot.mode configuration. Do not use both options at the same time. If both are specified in the table DDL, scan.startup.mode may not take effect.

port

The port number of the SQL Server database service.

INTEGER

No

1433

None.

server-time-zone

The session time zone of the database server, such as "Asia/Shanghai"

STRING

No

None

None.

scan.incremental.snapshot.enabled

Enables or disables the incremental snapshot.

BOOLEAN

No

true

Valid values:

false: Disables the incremental snapshot.
true (default): Enables the incremental snapshot.

Note

When the incremental snapshot framework is disabled (scan.incremental.snapshot.enabled = false), checkpoints cannot be performed during the full phase. If a table contains a large amount of data, a checkpoint timeout causes the job to fail and restart. We recommend that you enable the incremental snapshot framework for large tables.
For more information about the benefits, prerequisites, and limitations of incremental snapshots, see Features, Prerequisites, and Limitations.

chunk-meta.group.size

The group size for chunk metadata. If the size is exceeded, the metadata is split into multiple groups.

INTEGER

No

1000

None.

chunk-key.even-distribution.factor.lower-bound

The lower bound of the distribution factor for the chunk key.

DOUBLE

No

0.05

The distribution factor is used to determine whether table data is uniformly distributed. A uniform computation optimization is used for a uniform distribution, while a non-uniform distribution triggers a split query. The formula is: (MAX(id) - MIN(id) + 1) / rowCount.

chunk-key.even-distribution.factor.upper-bound

The upper bound of the distribution factor for the chunk key.

DOUBLE

No

1000.0

The distribution factor is used to determine whether table data is evenly distributed. If the data is evenly distributed, uniform calculation optimization is used. Otherwise, a split query is triggered. The calculation formula is: (MAX(id) - MIN(id) + 1) / rowCount.

scan.incremental.close-idle-reader.enabled

Specifies whether to close idle readers after the snapshot is complete.

BOOLEAN

No

false

For this configuration to take effect, you must set execution.checkpointing.checkpoints-after-tasks-finish.enabled to true.

scan.incremental.snapshot.chunk.key-column

The column to use as a key for splitting chunks during the snapshot phase.

STRING

No

None

The column used to split the table into chunks for the snapshot read. This column must be part of the primary key. Defaults to the first primary key column.

scan.incremental.snapshot.unbounded-chunk-first.enabled

Prioritizes processing the largest, unbounded chunk first. This reduces the risk of TaskManager OutOfMemory (OOM) errors.

BOOLEAN

No

true

None.

scan.incremental.snapshot.backfill.skip

Specifies whether to skip reading the transaction log during the full phase.

Boolean

No

false

Valid values:

true: skips backfill.
false (default): does not skip backfill.

Backfill applies only during the snapshot query of a single chunk and does not cover the entire full-read phase. When backfill is skipped, each chunk's snapshot query reads the latest table data at that instant; updates that occur on a chunk after it has been read are not merged during the full-read phase and are read from the transaction log after entering the incremental phase. For example, an update to chunk5 that occurs while chunk5 is being snapshotted is reflected directly in chunk5's snapshot; if chunk5 is updated after the reader has advanced to chunk80, the update is applied later from the transaction log during the incremental phase.

Important

When enabled, changes that occur during or after a chunk's scan are still delivered from the transaction log in the incremental phase and may be duplicated. Only at-least-once semantics are guaranteed. Enable this only when the downstream sink supports idempotent writes by primary key.

debezium.*

Debezium property parameters.

STRING

No

None

Use these options for finer-grained control over the Debezium client's behavior. For more information, see Configuration Properties.

Use with caution. Incorrect settings can prevent the connector from capturing data correctly.

Type mapping

The following table lists the data type mappings between SQL Server and Flink.

SQL Server type	Flink type
bit	BOOLEAN
tinyint	SMALLINT
smallint	SMALLINT
int	INT
bigint	BIGINT
float	DOUBLE
real	DOUBLE
numeric	DECIMAL(p,s)
decimal(p,s)
money
smallmoney
date	DATE
time(n)	TIME(n)
datetime2	TIMESTAMP(n)
datetime
smalldatetime
datetimeoffset	TIMESTAMP_LTZ(3)
char(n)	CHAR(n)
varchar(n)	VARCHAR(n)
nvarchar(n)
nchar(n)
text	STRING
ntext
xml

Data ingestion

Syntax

source:
  type: sqlserver
  name: SQLServer Source
  hostname: 127.0.0.1
  port: 1433
  username: sa
  password: Password!
  tables: inventory.dbo.orders

sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://127.0.0.1:9030
  load-url: 127.0.0.1:8030
  username: root
  password: pass

The tables option follows the three-part naming convention database.schema.table and supports regular expressions to match multiple tables. For example, inventory.dbo.\.* captures all tables under the dbo schema of the inventory database.

Parameters

Parameter	Description	Data type	Required	Default	Notes
type	Source type.	STRING	Yes	None	Fixed value: `sqlserver`.
hostname	IP address or hostname of the SQL Server database.	STRING	Yes	None	None.
username	Username for the SQL Server database.	STRING	Yes	None	None.
password	Password for the SQL Server database.	STRING	Yes	None	None.
tables	Names of the tables to capture.	STRING	Yes	None	Use the three-part naming convention `database.schema.table`. The `database` segment must be a fixed name and does not support regular expressions; `schema` and `table` segments support regular expressions. Separate multiple tables with commas. Example: `inventory.dbo.orders,inventory.dbo.users`. Regex examples: `inventory.dbo.user_table_[0-9]+` matches user_table_0, user_table_1, and so on; `inventory.dbo.(app\|web)_order_\.*` matches tables prefixed with app_order_ or web_order_.
port	Port of the SQL Server database service.	INTEGER	No	1433	None.
tables.exclude	Names of tables to exclude.	STRING	No	None	Uses the same syntax as `tables`. Excludes specified tables from the matched set.
scan.startup.mode	Startup mode for data consumption.	STRING	No	initial	Valid values: `initial` (default): on first startup, scans the full historical snapshot, then reads the latest transaction log. `latest-offset`: on first startup, skips the full snapshot and reads directly from the latest transaction log. `snapshot`: reads the full historical snapshot only; does not read the transaction log. `timestamp`: reads the transaction log starting from the specified timestamp; must be used together with `scan.startup.timestamp-millis`.
scan.startup.timestamp-millis	Startup timestamp in milliseconds.	LONG	No	None	Required only when `scan.startup.mode` is set to `timestamp`.
server-time-zone	Session time zone of the database server.	STRING	No	None	For example, `Asia/Shanghai`. Falls back to the system default time zone if not set.
scan.incremental.snapshot.chunk.size	Chunk size (in rows) for the table snapshot.	INTEGER	No	8096	The table is split into multiple chunks during snapshot reading.
scan.snapshot.fetch.size	Maximum number of rows fetched per poll.	INTEGER	No	1024	None.
scan.incremental.snapshot.chunk.key-column	Splitting column used during the snapshot phase.	STRING	No	None	Defaults to the first column of the primary key. The specified column must be one of the primary-key columns.
scan.incremental.snapshot.backfill.skip	Whether to skip transaction-log backfill during the full-read phase.	BOOLEAN	No	true	Valid values: `true` (default): skips backfill. `false`: does not skip backfill. Backfill applies only during the snapshot query of a single chunk and does not cover the entire full-read phase. When backfill is skipped, each chunk's snapshot query reads the latest table data at that instant; updates that occur on a chunk after it has been read are not merged during the full-read phase and are read from the transaction log after entering the incremental phase. For example, an update to chunk5 that occurs while chunk5 is being snapshotted is reflected directly in chunk5's snapshot; if chunk5 is updated after the reader has advanced to chunk80, the update is applied later from the transaction log during the incremental phase. Important When enabled, changes that occur during or after a chunk's scan are still delivered from the transaction log in the incremental phase and may be duplicated. Only at-least-once semantics are guaranteed. Enable this only when the downstream sink supports idempotent writes by primary key.
schema-change.enabled	Whether to emit schema change events.	BOOLEAN	No	true	When set to `false`, upstream DDL changes are not synced to the downstream.
connect.timeout	Connection timeout.	DURATION	No	30s	None.
connect.max-retries	Maximum number of connection retries.	INTEGER	No	3	None.
connection.pool.size	Connection pool size.	INTEGER	No	20	None.
scan.incremental.close-idle-reader.enabled	Whether to close idle readers after the snapshot ends.	BOOLEAN	No	false	Requires `execution.checkpointing.checkpoints-after-tasks-finish.enabled` to be set to `true`.
scan.incremental.snapshot.unbounded-chunk-first.enabled	Whether to distribute the unbounded chunk first during the snapshot phase.	BOOLEAN	No	false	Helps reduce the risk of OOM when a TaskManager processes the largest unbounded chunk.
scan.newly-added-table.enabled	Whether to scan newly added tables.	BOOLEAN	No	false	Takes effect only when restarting from a savepoint or checkpoint.
metadata.list	Metadata columns to propagate to the downstream.	STRING	No	None	Comma-separated. Supported values: `database_name`, `schema_name`, `table_name`, `op_ts`.
chunk-meta.group.size	Group size for chunk metadata.	INTEGER	No	1000	None.
chunk-key.even-distribution.factor.upper-bound	Upper bound of the chunk key distribution factor.	DOUBLE	No	1000.0	The distribution factor determines whether the table data is evenly distributed. Formula: (MAX(id) - MIN(id) + 1) / rowCount.
chunk-key.even-distribution.factor.lower-bound	Lower bound of the chunk key distribution factor.	DOUBLE	No	0.05	Same as above.
debezium.*	Debezium property parameters.	STRING	No	None	Provides fine-grained control over Debezium client behavior. Not recommended unless you have a specific need.
jdbc.properties.*	JDBC connection properties.	STRING	No	None	Additional properties passed to the JDBC driver.

Type mapping

SQL Server to Flink CDC data type mappings:

SQL Server data type	Flink CDC data type
bit	BOOLEAN
tinyint	SMALLINT
smallint	SMALLINT
int	INT
bigint	BIGINT
real	FLOAT
float	DOUBLE
numeric(p,s)	DECIMAL(p,s)
decimal(p,s)	DECIMAL(p,s)
money	DECIMAL(19,4)
smallmoney	DECIMAL(10,4)
date	DATE
time(n)	TIME(n)
datetime2(n)	TIMESTAMP(n)
datetime	TIMESTAMP(3)
smalldatetime	TIMESTAMP(0)
datetimeoffset(n)	TIMESTAMP_LTZ(n)
char(n)	CHAR(n)
nchar(n)	CHAR(n)
varchar(n)	VARCHAR(n)
nvarchar(n)	VARCHAR(n)
text	STRING
ntext	STRING
xml	STRING
uniqueidentifier	STRING
binary(n)	BYTES
varbinary(n)	BYTES
image	BYTES
timestamp / rowversion	BYTES

Metadata

Metadata in Flink CDC data-ingestion jobs falls into two categories:

1. Framework metadata columns: automatically inferred by the Flink CDC framework from each event's attributes. No need to configure metadata.list in the source. See Metadata columns in the Flink CDC documentation.

2. Source-provided metadata columns: read from change events emitted by the source. Configure metadata.list in the source. Supported metadata varies by source. SQL Server supports the following. Use commas to declare multiple.

Key	Data type	Description
database_name	STRING NOT NULL	Name of the database that owns the row.
schema_name	STRING NOT NULL	Name of the schema that owns the row.
table_name	STRING NOT NULL	Name of the table that owns the row.
op_ts	TIMESTAMP_LTZ(3) NOT NULL	Time when the row was changed in the database. Returns 0 (1970-01-01 00:00:00) for rows read from a snapshot during the full-read phase.

Example of declaring SQL Server metadata in Flink CDC:

source:
  type: sqlserver
  name: SQLServer Source
  hostname: 127.0.0.1
  port: 1433
  username: sa
  password: Password!
  tables: inventory.dbo.orders
  -- SQL Server source-provided metadata; use commas for multiple entries
  metadata.list: database_name,schema_name,table_name,op_ts

transform:
  - source: inventory.dbo.orders
    -- * refers to all source columns; table_name and op_ts are source-provided metadata; op_type is a framework metadata column
    projection: "*, table_name AS table_name, op_ts AS op_ts, __data_event_type__ AS op_type"

Note: __data_event_type__ is a framework-provided metadata column used to identify the event type. Flink CDC recognizes three event types: insert, update, and delete. Unlike Flink SQL, which stores an update as two separate records, Flink CDC keeps each update as a single record. This preserves complete update semantics and lets Flink CDC forward original update events to downstream systems.

Additional information

Metadata

Metadata in the following formats can be exposed as read-only (VIRTUAL) columns in a table definition:

Key	Type	Description
database_name	STRING NOT NULL	The name of the database where the row originated.
schema_name	STRING NOT NULL	The name of the schema where the row originated.
table_name	STRING NOT NULL	The name of the table where the row originated.
op_ts	TIMESTAMP_LTZ(3) NOT NULL	The time the change occurred in the database. This value is 0 for records from a table snapshot instead of the transaction log.

The following example creates a table that uses metadata fields:

CREATE TABLE products (
    table_name    STRING METADATA FROM 'table_name' VIRTUAL,
    schema_name   STRING METADATA FROM 'schema_name' VIRTUAL,
    db_name       STRING METADATA FROM 'database_name' VIRTUAL,
    operation_ts  TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    id            INT NOT NULL,
    name          STRING,
    description   STRING,
    weight        DECIMAL(10, 3)
) WITH (
    'connector'     = 'sqlserver-cdc',
    'hostname'      = 'localhost',
    'port'          = '1433',
    'username'      = 'sa',
    'password'      = 'Password!',
    'database-name' = 'inventory',
    'table-name'    = 'dbo.products'
);