SQL Server Connector

SQL Server CDC Pipeline 连接器支持从 SQL Server 数据库读取快照数据和增量数据，并提供端到端的表数据同步能力。本文描述如何设置 SQL Server CDC Pipeline 连接器。

前置条件

• SQL Server Agent 正在运行。
• 已为数据库和需要捕获的表开启 Change Data Capture。
• 配置的 SQL Server 用户可以连接服务器并读取被捕获的表。
• 当前连接器的 tables 配置中，数据库只支持单个固定库名，schema 和 table 支持正则表达式匹配多个。

为数据库和表启用 CDC：

USE MyDB;
GO
EXEC sys.sp_cdc_enable_db;
GO

EXEC sys.sp_cdc_enable_table
@source_schema = N'dbo',
@source_name   = N'MyTable',
@role_name     = NULL,
@supports_net_changes = 0;
GO

检查表是否已启用 CDC：

USE MyDB;
GO
EXEC sys.sp_cdc_help_change_data_capture;
GO

示例

从 SQL Server 读取数据同步到 Doris 的 Pipeline 可以定义如下：

source:
   type: sqlserver
   name: SQL Server Source
   hostname: 127.0.0.1
   port: 1433
   username: root
   password: 123456
   # 数据库只支持单个固定库名，schema 和 table 支持正则匹配多个。
   tables: inventory.dbo.\.*
   schema-change.enabled: true

sink:
  type: doris
  name: Doris Sink
  fenodes: 127.0.0.1:8030
  username: root
  password: 123456

pipeline:
   name: SQL Server to Doris Pipeline
   parallelism: 4

连接器配置项

Option	Required	Default	Type	Description
hostname	required	(none)	String	SQL Server 数据库服务器的 IP 地址或主机名。
port	optional	1433	Integer	SQL Server 数据库服务器的整数端口号。
username	required	(none)	String	连接 SQL Server 数据库服务器时使用的 SQL Server 用户名。
password	required	(none)	String	连接 SQL Server 数据库服务器时使用的密码。
tables	required	(none)	String	需要监控的 SQL Server 表名。database 只支持单个固定库名，schema 和 table 支持正则表达式匹配多个；不支持数据库级别正则表达式或跨数据库匹配。点号（.）会被视为 database、schema 和 table 名的分隔符。如果需要在正则表达式中使用点号（.）匹配任意字符，必须使用反斜杠转义。 示例：inventory.dbo.\.*、inventory.dbo.user_table_[0-9]+、inventory.dbo.(app|web)_order_\.*
tables.exclude	optional	(none)	String	在应用 `tables` 后需要排除的 SQL Server 表名。schema 和 table 支持正则表达式匹配多个。所有排除规则必须与 `tables` 选项使用同一个固定库名。 示例：inventory.dbo.audit_\.*、inventory.dbo.tmp_[0-9]+
schema-change.enabled	optional	true	Boolean	是否发送 schema change 事件，使下游 sink 可以同步表结构变更。
server-time-zone	optional	(none)	String	数据库服务器中的会话时区。如果未设置，则使用 ZoneId.systemDefault() 确定服务器时区。
scan.incremental.snapshot.chunk.key-column	optional	(none)	String	表快照的切分键。默认使用主键的第一列，该列必须是主键列。
scan.incremental.snapshot.chunk.size	optional	8096	Integer	表快照的 chunk 大小，单位为行数。
scan.snapshot.fetch.size	optional	1024	Integer	读取表快照时每次轮询的最大读取条数。
scan.startup.mode	optional	initial	String	SQL Server CDC 消费者可选启动模式。合法值为 "initial"、"latest-offset"、"snapshot" 和 "timestamp"。
scan.startup.timestamp-millis	optional	(none)	Long	当 `scan.startup.mode` 为 "timestamp" 时使用的启动时间戳。
scan.incremental.snapshot.backfill.skip	optional	false	Boolean	是否在快照读取阶段跳过 backfill。跳过 backfill 可能导致部分 change log 事件以 at-least-once 语义被重放。
scan.incremental.snapshot.unbounded-chunk-first.enabled	optional	true	Boolean	是否在快照读取阶段优先分配无上界 chunk。这可能有助于降低对最大的无上界 chunk 执行快照时 TaskManager 出现内存溢出（OOM）的风险。
connect.timeout	optional	30s	Duration	连接器尝试连接 SQL Server 数据库服务器后的最长等待时间。
connect.max-retries	optional	3	Integer	连接器建立 SQL Server 数据库服务器连接的最大重试次数。
connection.pool.size	optional	20	Integer	连接池大小。
metadata.list	optional	(none)	String	从 SourceRecord 中读取并传递给下游的元数据列表，使用英文逗号分隔。可用元数据包括：database_name、schema_name、table_name、op_ts。
scan.incremental.close-idle-reader.enabled	optional	false	Boolean	是否在快照阶段结束时关闭空闲 reader。此特性依赖 FLIP-147。
scan.newly-added-table.enabled	optional	false	Boolean	是否扫描新增表。该选项仅在作业从 savepoint 或 checkpoint 启动时有用。
debezium.*	optional	(none)	String	传递给 Debezium Embedded Engine 的 Debezium 属性。
jdbc.properties.*	optional	(none)	String	传递自定义 JDBC URL 属性。例如：jdbc.properties.encrypt=false。

兼容性说明：如果希望保持旧版本 Pipeline 行为，可以显式配置 scan.incremental.snapshot.backfill.skip: true 和 scan.incremental.snapshot.unbounded-chunk-first.enabled: false。

可用 Metadata

配置 metadata.list 后，以下 metadata 可以传递到下游。

Key	DataType	Description
database_name	STRING NOT NULL	包含该行的数据库名称。
schema_name	STRING NOT NULL	包含该行的 schema 名称。
table_name	STRING NOT NULL	包含该行的表名。
op_ts	TIMESTAMP_LTZ(3) NOT NULL	该变更在数据库中发生的时间。对于快照记录，该值始终为 0。

启动读取位置

配置项 scan.startup.mode 指定 SQL Server CDC 消费者的启动模式。有效值包括：

• initial：先对被监控表执行初始快照，然后继续读取最新变更。
• latest-offset：从最新 change log offset 开始读取。
• snapshot：只读取快照。
• timestamp：从 scan.startup.timestamp-millis 指定的时间戳开始读取。

数据类型映射

SQL Server type	CDC type	NOTE
BIT	BOOLEAN
TINYINT	SMALLINT	SQL Server 中的 TINYINT 为无符号类型（取值 0-255），因此映射为范围更大的 SMALLINT。
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
REAL	FLOAT
FLOAT	DOUBLE
DECIMAL(p, s) NUMERIC(p, s)	DECIMAL(p, s)	当精度大于 Flink 支持的上限 38 时，会回退为 DECIMAL(38, 0)。
MONEY	DECIMAL(19, 4)
SMALLMONEY	DECIMAL(10, 4)
CHAR(n) NCHAR(n)	CHAR(n)	当长度信息不可用时，映射为 STRING。
VARCHAR(n) NVARCHAR(n)	VARCHAR(n)	当长度信息不可用时，映射为 STRING。
TEXT NTEXT	STRING
BINARY VARBINARY IMAGE	BYTES
TIMESTAMP ROWVERSION	BYTES	SQL Server 中的 TIMESTAMP/ROWVERSION 是行版本二进制值，并非日期时间类型。
DATE	DATE
TIME(p)	TIME(p)
SMALLDATETIME	TIMESTAMP(0)
DATETIME	TIMESTAMP(3)
DATETIME2(p)	TIMESTAMP(p)	未显式指定精度时，默认精度为 7。
DATETIMEOFFSET(p)	TIMESTAMP_LTZ(p)	未显式指定精度时，默认精度为 7。
UNIQUEIDENTIFIER XML SQL_VARIANT HIERARCHYID GEOMETRY GEOGRAPHY	STRING

限制

单数据库

tables 中的所有条目必须属于同一个数据库。数据库只支持单个固定库名，schema 和 table 支持正则表达式匹配多个。