用 Flink CDC 构建 SQL Server 实时数据管道Pipeline 连接器配置、增量快照与类型映射全指南【免费下载链接】flink-cdcFlink CDC is a streaming data integration tool项目地址: https://gitcode.com/GitHub_Trending/flin/flink-cdc本指南围绕 Flink CDC 仓库中的 SQL Server Pipeline 连接器位于 flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver展开系统讲解如何从 SQL Server 读取快照与增量数据、完成端到端的表数据同步并深入剖析tables正则匹配、schema change 同步、启动位置选择与类型映射背后的源码实现。读完本文你将能够在 YAML Pipeline 中正确配置 SQL Server 数据源将其同步到 Doris、Kafka、StarRocks 等任意受支持的下游并掌握对增量快照、metadata 与数据类型边界进行调优的实战技能。连接器定位与整体工作方式SQL Server CDC Pipeline 连接器是 Flink CDC 数据管道体系YAML Pipeline 模式中的 Source 端实现支持从 SQL Server 数据库读取快照数据和增量数据并提供端到端的表数据同步能力。与传统的 Flink SQL Connector 不同Pipeline 模式下你不再需要手写 DDL只需在 YAML 中声明source、sink与pipeline三段配置Flink CDC 便会自动完成表结构发现、schema 变更传播和数据路由。从源码看该连接器的核心实现分为两层工厂层SqlServerDataSourceFactory.java 以标识符sqlserver被 Flink CDC 的工厂机制识别负责解析 YAML 中source段的所有配置项进行合法性校验并完成tables正则到实际表清单的解析数据源层SqlServerDataSource.java 组装SqlServerEventDeserializer事件反序列化、LsnFactoryLSN 偏移工厂与SqlServerDialectSQL Server 方言最终生成可供 Flink 运行的SqlServerPipelineSource。其中增量读取基于 SQL Server 的 Change Data CaptureCDC机制与事务日志 LSN 定位快照读取则采用分块chunk并发扫描的方式两者通过增量快照incremental snapshot算法衔接保证一致性读。前置条件启用数据库与表级 CDC在创建 Pipeline 之前需要满足以下条件SQL Server Agent 正在运行。这是硬性前置Flink CDC 在创建 Pipeline 时会主动校验SqlServerSchemaUtils.validateSqlServerAgentRunning()会执行SELECT TOP(1) status_desc FROM sys.dm_server_services WHERE servicename LIKE SQL Server Agent (%查询见 SqlServerSchemaUtils.java若 Agent 状态不是Running会直接抛出ValidationException并终止任务创建已为数据库和需要捕获的表开启 Change Data Capture配置的 SQL Server 用户可以连接服务器并读取被捕获的表tables配置中数据库只支持单个固定库名schema 和 table 支持正则表达式匹配多个具体规则见下文「表匹配规则」章节。为数据库和表启用 CDC在 SQL Server 中执行以下脚本开启 CDCUSE MyDB; GO EXEC sys.sp_cdc_enable_db; GO EXEC sys.sp_cdc_enable_table source_schema Ndbo, source_name NMyTable, role_name NULL, supports_net_changes 0; GO其中supports_net_changes 0表示不启用净变更net changes查询支持role_name NULL表示不限制捕获实例的访问角色。检查表是否已启用 CDCUSE MyDB; GO EXEC sys.sp_cdc_help_change_data_capture; GO该存储过程会列出当前数据库中所有已启用捕获的表及其捕获实例信息用于确认 CDC 配置是否生效。快速上手SQL Server 同步到 Doris 的完整 Pipeline从 SQL Server 读取数据同步到 Doris 的 Pipeline 可以定义如下完整示例源自 sqlserver.md 的示例章节下游 Sink 的配置请参考 pipeline-connectors 中对应 Sink 文档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各段配置的作用如下source 段声明数据源为sqlserver。hostname/port/username/password为连接信息tables声明要捕获的表集合schema-change.enabled: true表示把表结构变更事件一并下发使下游 Sink 可以同步执行 DDLsink 段声明下游为 Doris填写 FE 地址与认证信息pipeline 段为管道命名并设置并行度parallelism: 4。SQL Server 的增量变更集中在单一事务日志流上但快照阶段可以多并行度并发分块扫描。从源码角度source.type: sqlserver会由 SqlServerDataSourceFactory.java 的IDENTIFIER sqlserver解析随后createDataSource()会按以下顺序完成初始化SqlServerDataSourceFactory.java读取全部配置项并对整数型选项做下限校验如scan.incremental.snapshot.chunk.size、connection.pool.size必须大于 1从tables中解析出固定数据库名getValidateDatabaseName作为databaseList传给 Debezium 引擎校验 SQL Server Agent 是否运行、列出全库表清单再用Selectors按正则过滤出实际捕获表若找不到任何匹配表直接抛出IllegalArgumentException提示检查tables取值将捕获表清单格式为schema.table写入tableList最终构造SqlServerDataSource。连接器配置项详解下表完整覆盖该连接器支持的配置项与 sqlserver.md 的配置表一一对应默认值可在 SqlServerDataSourceOptions.java 中找到源码定义OptionRequiredDefaultTypeDescriptionhostnamerequired(none)StringSQL Server 数据库服务器的 IP 地址或主机名。portoptional1433IntegerSQL Server 数据库服务器的整数端口号。usernamerequired(none)String连接 SQL Server 数据库服务器时使用的 SQL Server 用户名。passwordrequired(none)String连接 SQL Server 数据库服务器时使用的密码。tablesrequired(none)String需要监控的 SQL Server 表名。database 只支持单个固定库名schema 和 table 支持正则表达式匹配多个不支持数据库级别正则表达式或跨数据库匹配。点号.会被视为 database、schema 和 table 名的分隔符如果需要在正则表达式中使用点号.匹配任意字符必须使用反斜杠转义。示例inventory.dbo.\.*、inventory.dbo.user_table_[0-9]、inventory.dbo.(app|web)_order_\.*tables.excludeoptional(none)String在应用tables后需要排除的 SQL Server 表名。schema 和 table 支持正则表达式匹配多个所有排除规则必须与tables选项使用同一个固定库名。示例inventory.dbo.audit_\.*、inventory.dbo.tmp_[0-9]schema-change.enabledoptionaltrueBoolean是否发送 schema change 事件使下游 sink 可以同步表结构变更。server-time-zoneoptional(none)String数据库服务器中的会话时区。如果未设置则使用ZoneId.systemDefault()确定服务器时区。scan.incremental.snapshot.chunk.key-columnoptional(none)String表快照的切分键。默认使用主键的第一列该列必须是主键列。scan.incremental.snapshot.chunk.sizeoptional8096Integer表快照的 chunk 大小单位为行数。scan.snapshot.fetch.sizeoptional1024Integer读取表快照时每次轮询的最大读取条数。scan.startup.modeoptionalinitialStringSQL Server CDC 消费者可选启动模式。合法值为initial、latest-offset、snapshot和timestamp。scan.startup.timestamp-millisoptional(none)Long当scan.startup.mode为timestamp时使用的启动时间戳。scan.incremental.snapshot.backfill.skipoptionalfalseBoolean是否在快照读取阶段跳过 backfill。跳过 backfill 可能导致部分 change log 事件以 at-least-once 语义被重放。scan.incremental.snapshot.unbounded-chunk-first.enabledoptionaltrueBoolean是否在快照读取阶段优先分配无上界 chunk。这可能有助于降低对最大的无上界 chunk 执行快照时 TaskManager 出现内存溢出OOM的风险。connect.timeoutoptional30sDuration连接器尝试连接 SQL Server 数据库服务器后的最长等待时间。connect.max-retriesoptional3Integer连接器建立 SQL Server 数据库服务器连接的最大重试次数。connection.pool.sizeoptional20Integer连接池大小。metadata.listoptional(none)String从 SourceRecord 中读取并传递给下游的元数据列表使用英文逗号分隔。可用元数据包括database_name、schema_name、table_name、op_ts。scan.incremental.close-idle-reader.enabledoptionalfalseBoolean是否在快照阶段结束时关闭空闲 reader。此特性依赖 FLIP-147。scan.newly-added-table.enabledoptionalfalseBoolean是否扫描新增表。该选项仅在作业从 savepoint 或 checkpoint 启动时有用。debezium.*optional(none)String传递给 Debezium Embedded Engine 的 Debezium 属性。jdbc.properties.*optional(none)String传递自定义 JDBC URL 属性。例如jdbc.properties.encryptfalse。关键参数的源码级解读连接与网络参数。connect.timeout、connect.max-retries、connection.pool.size分别控制 JDBC 连接的超时、重试次数与连接池容量默认 30s / 3 次 / 20 个连接定义于 SqlServerDataSourceOptions.java在高并发快照扫描下建议根据表数量和集群规模适当调大连接池。快照 chunk 参数。scan.incremental.snapshot.chunk.size默认 8096 行决定每个快照分块的行数scan.incremental.snapshot.chunk.key-column用于指定切分键默认取主键第一列且该列必须是主键列源码描述见 SqlServerDataSourceOptions.java。对于无主键或主键分布不均的表合理设置切分键能显著影响快照并行度与均衡性。backfill 行为与兼容性。scan.incremental.snapshot.backfill.skip默认 false控制是否在快照阶段跳过 backfill跳过意味着快照阶段发生的数据变更不会被合并进快照而是留待后续 changelog 阶段消费代价是可能重复消费仅保证 at-least-once出现对已更新值再次更新等重复事件。兼容性说明如果希望保持旧版本 Pipeline 行为可以显式配置scan.incremental.snapshot.backfill.skip: true和scan.incremental.snapshot.unbounded-chunk-first.enabled: false。时区参数。server-time-zone未设置时工厂会打印告警日志并回退到ZoneId.systemDefault()见 SqlServerDataSourceFactory.java这可能引起时间字段的数据不一致生产环境建议显式指定。透传参数。debezium.*前缀的属性会整体注入 Debezium Embedded Enginejdbc.properties.*前缀的属性会被合并进 Debezium 的database.*配置见mergeJdbcPropertiesIntoDebeziumPropertiesSqlServerDataSourceFactory.java典型用法如jdbc.properties.encryptfalse关闭 JDBC 加密。表匹配规则tables 与 tables.exclude 的正则语义tables是该连接器最核心、也最容易踩坑的配置。其语义为database.schema.table三段式其中database 必须是单个固定库名不支持数据库级别的正则或跨数据库匹配schema 和 table 支持正则表达式可以匹配多个表点号.是 database、schema、table 名的分隔符若要在正则中表达匹配任意字符必须写成\.转义多个条目用英文逗号分隔当逗号属于正则的一部分时需要用反斜杠转义。常用示例inventory.dbo.\.* # inventory 库 dbo schema 下的所有表 inventory.dbo.user_table_[0-9] # user_table_0 至 user_table_9 等 inventory.dbo.(app|web)_order_\.* # app_order_* 与 web_order_* 系列表tables.exclude在tables匹配结果的基础上做减法同样遵循同一固定库名 schema/table 正则的约束典型场景是排除审计表、临时表inventory.dbo.audit_\.* # 排除 audit_ 开头的表 inventory.dbo.tmp_[0-9] # 排除 tmp_ 数字结尾的表源码实现印证工厂在createDataSource()中调用getValidateDatabaseName(tables)SqlServerDataSourceFactory.java按,拆分为多个表名后要求每个表名严格为三段式且所有条目的第一段数据库名必须完全一致否则抛错同时校验数据库名长度不超过 SQL Server 标识符上限 128 字符。随后通过SelectorsincludeTables(tables)过滤全库表清单得到捕获表若过滤结果为空则拒绝启动tables.exclude同理构建排除选择器并执行removeAllSqlServerDataSourceFactory.java。最终写入 Debezium 的 tableList 格式为schema.table不带数据库前缀。Schema Change 事件与表结构变更同步schema-change.enabled默认 true控制是否将 SQL Server 的表结构变更作为事件发送给下游。开启后下游 Sink如 Doris、StarRocks、Iceberg 等支持 schema evolution 的组件可以自动同步CREATE TABLE、ALTER TABLE、DROP TABLE。这一能力的实现位于 SqlServerEventDeserializer.java 的deserializeSchemaChangeRecord方法连接器从 Debezium 的历史记录中反序列化TableChanges并针对每种变更类型做不同处理——CREATE将表结构转换为CreateTableEvent并写入本地缓存用于后续 ALTER 的增量对比ALTER从缓存中取出旧 schema通过SchemaMergingUtils.getSchemaDifference计算新旧 schema 差异产出最小化的 schema change 事件集合如AddColumnEvent、AlterColumnTypeEventDROP直接产出DropTableEvent并清除缓存。需要注意SQL Server 是单事务日志流因此SqlServerDataSource.isParallelMetadataSource()返回falseSqlServerDataSource.java即增量阶段不会出现不同分区同时发出 schema change 事件的竞争问题。可用 Metadata配置metadata.list多个 metadata 用逗号分隔后以下 metadata 会随数据记录传递到下游数据类型与描述与 sqlserver.md 的 metadata 表一致KeyDataTypeDescriptiondatabase_nameSTRING NOT NULL包含该行的数据库名称。schema_nameSTRING NOT NULL包含该行的 schema 名称。table_nameSTRING NOT NULL包含该行的表名。op_tsTIMESTAMP_LTZ(3) NOT NULL该变更在数据库中发生的时间。对于快照记录该值始终为 0。从源码看metadata.list的解析由 SqlServerDataSourceFactory.java 的listReadableMetadata完成按逗号拆分、去空格后与SqlServerReadableMetadata枚举的 key 逐一比对遇到无法识别的 key 会抛出IllegalArgumentException提示该 metadata 不存在。实际读取时SqlServerEventDeserializer.getMetadata 会对每条 SourceRecord 调用对应 metadata 的 converterop_ts会取毫秒时间戳作为字符串传递。启动读取位置配置项scan.startup.mode指定 SQL Server CDC 消费者的启动模式有效值包括initial默认先对被监控表执行初始快照然后继续读取最新变更。适合全量 增量一体化同步的首启场景latest-offset从最新 change log offset即当前最新 LSN开始读取跳过历史数据snapshot只读取快照不消费增量 changelog适合一次性存量数据迁移timestamp从scan.startup.timestamp-millis指定的时间戳开始读取。源码中工厂的getStartupOptions()SqlServerDataSourceFactory.java将字符串映射为对应的StartupOptions特别地选择timestamp模式但未配置scan.startup.timestamp-millis时会抛出ValidationException明确提示必须设置该参数传入其他非法值时同样会拒绝并列出全部合法取值。数据类型映射SQL Server 类型到 Flink CDC 类型的映射关系如下表完整继承自 sqlserver.md 的类型映射表SQL Server typeCDC typeNOTEBITBOOLEANTINYINTSMALLINTSQL Server 中的TINYINT为无符号类型取值 0-255因此映射为范围更大的SMALLINT。SMALLINTSMALLINTINTINTBIGINTBIGINTREALFLOATFLOATDOUBLEDECIMAL(p, s) / NUMERIC(p, s)DECIMAL(p, s)当精度大于 Flink 支持的上限 38 时会回退为DECIMAL(38, 0)。MONEYDECIMAL(19, 4)SMALLMONEYDECIMAL(10, 4)CHAR(n) / NCHAR(n)CHAR(n)当长度信息不可用时映射为STRING。VARCHAR(n) / NVARCHAR(n)VARCHAR(n)当长度信息不可用时映射为STRING。TEXT / NTEXTSTRINGBINARY / VARBINARY / IMAGEBYTESTIMESTAMP / ROWVERSIONBYTESSQL Server 中的TIMESTAMP/ROWVERSION是行版本二进制值并非日期时间类型。DATEDATETIME(p)TIME(p)SMALLDATETIMETIMESTAMP(0)DATETIMETIMESTAMP(3)DATETIME2(p)TIMESTAMP(p)未显式指定精度时默认精度为 7。DATETIMEOFFSET(p)TIMESTAMP_LTZ(p)未显式指定精度时默认精度为 7。UNIQUEIDENTIFIER / XML / SQL_VARIANT / HIERARCHYID / GEOMETRY / GEOGRAPHYSTRING映射实现细节源码佐证上述映射并非文档层面的约定而是实打实编码在 SqlServerTypeUtils.java 的fromDbzColumn/convertFromColumn方法中值得注意的实现细节包括SQL Server 专有类型优先按类型名匹配money→DECIMAL(19, 4)、smallmoney→DECIMAL(10, 4)精度与小数位为源码中的常量MONEY_PRECISION19、SMALL_MONEY_PRECISION10、MONEY_SCALE4datetimeoffset/datetime2未显式指定精度时默认精度为 7timestamp/rowversion这类行版本二进制值映射为BYTESTINYINT 放大为 SMALLINTSQL Server 的TINYINT是无符号 0-255直接映射会溢出因此提升为SMALLINTDECIMAL 精度回退当精度超过 Flink 上限DecimalType.MAX_PRECISION即 38时回退为DECIMAL(38, 0)长度信息缺失时回退 STRINGCHAR/NCHAR/VARCHAR/NVARCHAR在取不到长度信息时映射为STRING可空性继承fromDbzColumn会根据列是否允许 NULL 自动追加NOT NULL约束不支持的类型会抛出UnsupportedOperationException明确提示Doesnt support SQL Server type ... yet。此外SqlServerSchemaUtils.toSchema 会将 Debezium 表结构转换为 Flink CDC 的Schema包括主键列、注释以及经归一化处理的默认值表达式去除外层括号、解包N...字符串字面量保证 schema 发现阶段拿到的结构与实际 DDL 一致。限制单数据库约束tables中的所有条目必须属于同一个数据库。数据库只支持单个固定库名schema 和 table 支持正则表达式匹配多个。这一限制来自 SQL Server CDC 捕获实例与 Debezium 连接器databaseList的设计一个连接器实例绑定一个数据库跨库捕获需要启动多个 Pipeline 分别指向不同库。与之配套的tables.exclude同样必须使用与tables相同的库名。测试佐证仓库为 SQL Server Pipeline 连接器提供了较完整的测试覆盖可作为理解行为与验证配置的参考位于 src/testSqlServerPipelineITCase.java端到端 Pipeline 同步全流程验证SqlServerFullTypesITCase.java全类型字段的映射与数据往返验证与上文类型映射表一一对应SqlServerTablePatternMatchingITCase.javatables正则匹配与tables.exclude排除规则的行为验证SqlServerOnlineSchemaMigrationITCase.java在线表结构变更schema evolution场景验证SqlServerMetadataAccessorITCase.java数据库/schema/表元数据枚举与表结构获取验证SqlServerPipelineSavepointRestoreITCase.javasavepoint 恢复与scan.newly-added-table.enabled相关行为验证。如果你需要把 SQL Server 数据同步到其他下游可以参考 pipeline-connectors 目录下对应 Sink 连接器的配置文档将上文sink段替换为目标组件即可source段的配置保持不变。【免费下载链接】flink-cdcFlink CDC is a streaming data integration tool项目地址: https://gitcode.com/GitHub_Trending/flin/flink-cdc创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 用 Flink CDC 构建 SQL Server 实时数据管道：Pipeline 连接器配置、增量快照与类型映射全指南
用 Flink CDC 构建 SQL Server 实时数据管道Pipeline 连接器配置、增量快照与类型映射全指南【免费下载链接】flink-cdcFlink CDC is a streaming data integration tool项目地址: https://gitcode.com/GitHub_Trending/flin/flink-cdc本指南围绕 Flink CDC 仓库中的 SQL Server Pipeline 连接器位于 flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver展开系统讲解如何从 SQL Server 读取快照与增量数据、完成端到端的表数据同步并深入剖析tables正则匹配、schema change 同步、启动位置选择与类型映射背后的源码实现。读完本文你将能够在 YAML Pipeline 中正确配置 SQL Server 数据源将其同步到 Doris、Kafka、StarRocks 等任意受支持的下游并掌握对增量快照、metadata 与数据类型边界进行调优的实战技能。连接器定位与整体工作方式SQL Server CDC Pipeline 连接器是 Flink CDC 数据管道体系YAML Pipeline 模式中的 Source 端实现支持从 SQL Server 数据库读取快照数据和增量数据并提供端到端的表数据同步能力。与传统的 Flink SQL Connector 不同Pipeline 模式下你不再需要手写 DDL只需在 YAML 中声明source、sink与pipeline三段配置Flink CDC 便会自动完成表结构发现、schema 变更传播和数据路由。从源码看该连接器的核心实现分为两层工厂层SqlServerDataSourceFactory.java 以标识符sqlserver被 Flink CDC 的工厂机制识别负责解析 YAML 中source段的所有配置项进行合法性校验并完成tables正则到实际表清单的解析数据源层SqlServerDataSource.java 组装SqlServerEventDeserializer事件反序列化、LsnFactoryLSN 偏移工厂与SqlServerDialectSQL Server 方言最终生成可供 Flink 运行的SqlServerPipelineSource。其中增量读取基于 SQL Server 的 Change Data CaptureCDC机制与事务日志 LSN 定位快照读取则采用分块chunk并发扫描的方式两者通过增量快照incremental snapshot算法衔接保证一致性读。前置条件启用数据库与表级 CDC在创建 Pipeline 之前需要满足以下条件SQL Server Agent 正在运行。这是硬性前置Flink CDC 在创建 Pipeline 时会主动校验SqlServerSchemaUtils.validateSqlServerAgentRunning()会执行SELECT TOP(1) status_desc FROM sys.dm_server_services WHERE servicename LIKE SQL Server Agent (%查询见 SqlServerSchemaUtils.java若 Agent 状态不是Running会直接抛出ValidationException并终止任务创建已为数据库和需要捕获的表开启 Change Data Capture配置的 SQL Server 用户可以连接服务器并读取被捕获的表tables配置中数据库只支持单个固定库名schema 和 table 支持正则表达式匹配多个具体规则见下文「表匹配规则」章节。为数据库和表启用 CDC在 SQL Server 中执行以下脚本开启 CDCUSE MyDB; GO EXEC sys.sp_cdc_enable_db; GO EXEC sys.sp_cdc_enable_table source_schema Ndbo, source_name NMyTable, role_name NULL, supports_net_changes 0; GO其中supports_net_changes 0表示不启用净变更net changes查询支持role_name NULL表示不限制捕获实例的访问角色。检查表是否已启用 CDCUSE MyDB; GO EXEC sys.sp_cdc_help_change_data_capture; GO该存储过程会列出当前数据库中所有已启用捕获的表及其捕获实例信息用于确认 CDC 配置是否生效。快速上手SQL Server 同步到 Doris 的完整 Pipeline从 SQL Server 读取数据同步到 Doris 的 Pipeline 可以定义如下完整示例源自 sqlserver.md 的示例章节下游 Sink 的配置请参考 pipeline-connectors 中对应 Sink 文档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各段配置的作用如下source 段声明数据源为sqlserver。hostname/port/username/password为连接信息tables声明要捕获的表集合schema-change.enabled: true表示把表结构变更事件一并下发使下游 Sink 可以同步执行 DDLsink 段声明下游为 Doris填写 FE 地址与认证信息pipeline 段为管道命名并设置并行度parallelism: 4。SQL Server 的增量变更集中在单一事务日志流上但快照阶段可以多并行度并发分块扫描。从源码角度source.type: sqlserver会由 SqlServerDataSourceFactory.java 的IDENTIFIER sqlserver解析随后createDataSource()会按以下顺序完成初始化SqlServerDataSourceFactory.java读取全部配置项并对整数型选项做下限校验如scan.incremental.snapshot.chunk.size、connection.pool.size必须大于 1从tables中解析出固定数据库名getValidateDatabaseName作为databaseList传给 Debezium 引擎校验 SQL Server Agent 是否运行、列出全库表清单再用Selectors按正则过滤出实际捕获表若找不到任何匹配表直接抛出IllegalArgumentException提示检查tables取值将捕获表清单格式为schema.table写入tableList最终构造SqlServerDataSource。连接器配置项详解下表完整覆盖该连接器支持的配置项与 sqlserver.md 的配置表一一对应默认值可在 SqlServerDataSourceOptions.java 中找到源码定义OptionRequiredDefaultTypeDescriptionhostnamerequired(none)StringSQL Server 数据库服务器的 IP 地址或主机名。portoptional1433IntegerSQL Server 数据库服务器的整数端口号。usernamerequired(none)String连接 SQL Server 数据库服务器时使用的 SQL Server 用户名。passwordrequired(none)String连接 SQL Server 数据库服务器时使用的密码。tablesrequired(none)String需要监控的 SQL Server 表名。database 只支持单个固定库名schema 和 table 支持正则表达式匹配多个不支持数据库级别正则表达式或跨数据库匹配。点号.会被视为 database、schema 和 table 名的分隔符如果需要在正则表达式中使用点号.匹配任意字符必须使用反斜杠转义。示例inventory.dbo.\.*、inventory.dbo.user_table_[0-9]、inventory.dbo.(app|web)_order_\.*tables.excludeoptional(none)String在应用tables后需要排除的 SQL Server 表名。schema 和 table 支持正则表达式匹配多个所有排除规则必须与tables选项使用同一个固定库名。示例inventory.dbo.audit_\.*、inventory.dbo.tmp_[0-9]schema-change.enabledoptionaltrueBoolean是否发送 schema change 事件使下游 sink 可以同步表结构变更。server-time-zoneoptional(none)String数据库服务器中的会话时区。如果未设置则使用ZoneId.systemDefault()确定服务器时区。scan.incremental.snapshot.chunk.key-columnoptional(none)String表快照的切分键。默认使用主键的第一列该列必须是主键列。scan.incremental.snapshot.chunk.sizeoptional8096Integer表快照的 chunk 大小单位为行数。scan.snapshot.fetch.sizeoptional1024Integer读取表快照时每次轮询的最大读取条数。scan.startup.modeoptionalinitialStringSQL Server CDC 消费者可选启动模式。合法值为initial、latest-offset、snapshot和timestamp。scan.startup.timestamp-millisoptional(none)Long当scan.startup.mode为timestamp时使用的启动时间戳。scan.incremental.snapshot.backfill.skipoptionalfalseBoolean是否在快照读取阶段跳过 backfill。跳过 backfill 可能导致部分 change log 事件以 at-least-once 语义被重放。scan.incremental.snapshot.unbounded-chunk-first.enabledoptionaltrueBoolean是否在快照读取阶段优先分配无上界 chunk。这可能有助于降低对最大的无上界 chunk 执行快照时 TaskManager 出现内存溢出OOM的风险。connect.timeoutoptional30sDuration连接器尝试连接 SQL Server 数据库服务器后的最长等待时间。connect.max-retriesoptional3Integer连接器建立 SQL Server 数据库服务器连接的最大重试次数。connection.pool.sizeoptional20Integer连接池大小。metadata.listoptional(none)String从 SourceRecord 中读取并传递给下游的元数据列表使用英文逗号分隔。可用元数据包括database_name、schema_name、table_name、op_ts。scan.incremental.close-idle-reader.enabledoptionalfalseBoolean是否在快照阶段结束时关闭空闲 reader。此特性依赖 FLIP-147。scan.newly-added-table.enabledoptionalfalseBoolean是否扫描新增表。该选项仅在作业从 savepoint 或 checkpoint 启动时有用。debezium.*optional(none)String传递给 Debezium Embedded Engine 的 Debezium 属性。jdbc.properties.*optional(none)String传递自定义 JDBC URL 属性。例如jdbc.properties.encryptfalse。关键参数的源码级解读连接与网络参数。connect.timeout、connect.max-retries、connection.pool.size分别控制 JDBC 连接的超时、重试次数与连接池容量默认 30s / 3 次 / 20 个连接定义于 SqlServerDataSourceOptions.java在高并发快照扫描下建议根据表数量和集群规模适当调大连接池。快照 chunk 参数。scan.incremental.snapshot.chunk.size默认 8096 行决定每个快照分块的行数scan.incremental.snapshot.chunk.key-column用于指定切分键默认取主键第一列且该列必须是主键列源码描述见 SqlServerDataSourceOptions.java。对于无主键或主键分布不均的表合理设置切分键能显著影响快照并行度与均衡性。backfill 行为与兼容性。scan.incremental.snapshot.backfill.skip默认 false控制是否在快照阶段跳过 backfill跳过意味着快照阶段发生的数据变更不会被合并进快照而是留待后续 changelog 阶段消费代价是可能重复消费仅保证 at-least-once出现对已更新值再次更新等重复事件。兼容性说明如果希望保持旧版本 Pipeline 行为可以显式配置scan.incremental.snapshot.backfill.skip: true和scan.incremental.snapshot.unbounded-chunk-first.enabled: false。时区参数。server-time-zone未设置时工厂会打印告警日志并回退到ZoneId.systemDefault()见 SqlServerDataSourceFactory.java这可能引起时间字段的数据不一致生产环境建议显式指定。透传参数。debezium.*前缀的属性会整体注入 Debezium Embedded Enginejdbc.properties.*前缀的属性会被合并进 Debezium 的database.*配置见mergeJdbcPropertiesIntoDebeziumPropertiesSqlServerDataSourceFactory.java典型用法如jdbc.properties.encryptfalse关闭 JDBC 加密。表匹配规则tables 与 tables.exclude 的正则语义tables是该连接器最核心、也最容易踩坑的配置。其语义为database.schema.table三段式其中database 必须是单个固定库名不支持数据库级别的正则或跨数据库匹配schema 和 table 支持正则表达式可以匹配多个表点号.是 database、schema、table 名的分隔符若要在正则中表达匹配任意字符必须写成\.转义多个条目用英文逗号分隔当逗号属于正则的一部分时需要用反斜杠转义。常用示例inventory.dbo.\.* # inventory 库 dbo schema 下的所有表 inventory.dbo.user_table_[0-9] # user_table_0 至 user_table_9 等 inventory.dbo.(app|web)_order_\.* # app_order_* 与 web_order_* 系列表tables.exclude在tables匹配结果的基础上做减法同样遵循同一固定库名 schema/table 正则的约束典型场景是排除审计表、临时表inventory.dbo.audit_\.* # 排除 audit_ 开头的表 inventory.dbo.tmp_[0-9] # 排除 tmp_ 数字结尾的表源码实现印证工厂在createDataSource()中调用getValidateDatabaseName(tables)SqlServerDataSourceFactory.java按,拆分为多个表名后要求每个表名严格为三段式且所有条目的第一段数据库名必须完全一致否则抛错同时校验数据库名长度不超过 SQL Server 标识符上限 128 字符。随后通过SelectorsincludeTables(tables)过滤全库表清单得到捕获表若过滤结果为空则拒绝启动tables.exclude同理构建排除选择器并执行removeAllSqlServerDataSourceFactory.java。最终写入 Debezium 的 tableList 格式为schema.table不带数据库前缀。Schema Change 事件与表结构变更同步schema-change.enabled默认 true控制是否将 SQL Server 的表结构变更作为事件发送给下游。开启后下游 Sink如 Doris、StarRocks、Iceberg 等支持 schema evolution 的组件可以自动同步CREATE TABLE、ALTER TABLE、DROP TABLE。这一能力的实现位于 SqlServerEventDeserializer.java 的deserializeSchemaChangeRecord方法连接器从 Debezium 的历史记录中反序列化TableChanges并针对每种变更类型做不同处理——CREATE将表结构转换为CreateTableEvent并写入本地缓存用于后续 ALTER 的增量对比ALTER从缓存中取出旧 schema通过SchemaMergingUtils.getSchemaDifference计算新旧 schema 差异产出最小化的 schema change 事件集合如AddColumnEvent、AlterColumnTypeEventDROP直接产出DropTableEvent并清除缓存。需要注意SQL Server 是单事务日志流因此SqlServerDataSource.isParallelMetadataSource()返回falseSqlServerDataSource.java即增量阶段不会出现不同分区同时发出 schema change 事件的竞争问题。可用 Metadata配置metadata.list多个 metadata 用逗号分隔后以下 metadata 会随数据记录传递到下游数据类型与描述与 sqlserver.md 的 metadata 表一致KeyDataTypeDescriptiondatabase_nameSTRING NOT NULL包含该行的数据库名称。schema_nameSTRING NOT NULL包含该行的 schema 名称。table_nameSTRING NOT NULL包含该行的表名。op_tsTIMESTAMP_LTZ(3) NOT NULL该变更在数据库中发生的时间。对于快照记录该值始终为 0。从源码看metadata.list的解析由 SqlServerDataSourceFactory.java 的listReadableMetadata完成按逗号拆分、去空格后与SqlServerReadableMetadata枚举的 key 逐一比对遇到无法识别的 key 会抛出IllegalArgumentException提示该 metadata 不存在。实际读取时SqlServerEventDeserializer.getMetadata 会对每条 SourceRecord 调用对应 metadata 的 converterop_ts会取毫秒时间戳作为字符串传递。启动读取位置配置项scan.startup.mode指定 SQL Server CDC 消费者的启动模式有效值包括initial默认先对被监控表执行初始快照然后继续读取最新变更。适合全量 增量一体化同步的首启场景latest-offset从最新 change log offset即当前最新 LSN开始读取跳过历史数据snapshot只读取快照不消费增量 changelog适合一次性存量数据迁移timestamp从scan.startup.timestamp-millis指定的时间戳开始读取。源码中工厂的getStartupOptions()SqlServerDataSourceFactory.java将字符串映射为对应的StartupOptions特别地选择timestamp模式但未配置scan.startup.timestamp-millis时会抛出ValidationException明确提示必须设置该参数传入其他非法值时同样会拒绝并列出全部合法取值。数据类型映射SQL Server 类型到 Flink CDC 类型的映射关系如下表完整继承自 sqlserver.md 的类型映射表SQL Server typeCDC typeNOTEBITBOOLEANTINYINTSMALLINTSQL Server 中的TINYINT为无符号类型取值 0-255因此映射为范围更大的SMALLINT。SMALLINTSMALLINTINTINTBIGINTBIGINTREALFLOATFLOATDOUBLEDECIMAL(p, s) / NUMERIC(p, s)DECIMAL(p, s)当精度大于 Flink 支持的上限 38 时会回退为DECIMAL(38, 0)。MONEYDECIMAL(19, 4)SMALLMONEYDECIMAL(10, 4)CHAR(n) / NCHAR(n)CHAR(n)当长度信息不可用时映射为STRING。VARCHAR(n) / NVARCHAR(n)VARCHAR(n)当长度信息不可用时映射为STRING。TEXT / NTEXTSTRINGBINARY / VARBINARY / IMAGEBYTESTIMESTAMP / ROWVERSIONBYTESSQL Server 中的TIMESTAMP/ROWVERSION是行版本二进制值并非日期时间类型。DATEDATETIME(p)TIME(p)SMALLDATETIMETIMESTAMP(0)DATETIMETIMESTAMP(3)DATETIME2(p)TIMESTAMP(p)未显式指定精度时默认精度为 7。DATETIMEOFFSET(p)TIMESTAMP_LTZ(p)未显式指定精度时默认精度为 7。UNIQUEIDENTIFIER / XML / SQL_VARIANT / HIERARCHYID / GEOMETRY / GEOGRAPHYSTRING映射实现细节源码佐证上述映射并非文档层面的约定而是实打实编码在 SqlServerTypeUtils.java 的fromDbzColumn/convertFromColumn方法中值得注意的实现细节包括SQL Server 专有类型优先按类型名匹配money→DECIMAL(19, 4)、smallmoney→DECIMAL(10, 4)精度与小数位为源码中的常量MONEY_PRECISION19、SMALL_MONEY_PRECISION10、MONEY_SCALE4datetimeoffset/datetime2未显式指定精度时默认精度为 7timestamp/rowversion这类行版本二进制值映射为BYTESTINYINT 放大为 SMALLINTSQL Server 的TINYINT是无符号 0-255直接映射会溢出因此提升为SMALLINTDECIMAL 精度回退当精度超过 Flink 上限DecimalType.MAX_PRECISION即 38时回退为DECIMAL(38, 0)长度信息缺失时回退 STRINGCHAR/NCHAR/VARCHAR/NVARCHAR在取不到长度信息时映射为STRING可空性继承fromDbzColumn会根据列是否允许 NULL 自动追加NOT NULL约束不支持的类型会抛出UnsupportedOperationException明确提示Doesnt support SQL Server type ... yet。此外SqlServerSchemaUtils.toSchema 会将 Debezium 表结构转换为 Flink CDC 的Schema包括主键列、注释以及经归一化处理的默认值表达式去除外层括号、解包N...字符串字面量保证 schema 发现阶段拿到的结构与实际 DDL 一致。限制单数据库约束tables中的所有条目必须属于同一个数据库。数据库只支持单个固定库名schema 和 table 支持正则表达式匹配多个。这一限制来自 SQL Server CDC 捕获实例与 Debezium 连接器databaseList的设计一个连接器实例绑定一个数据库跨库捕获需要启动多个 Pipeline 分别指向不同库。与之配套的tables.exclude同样必须使用与tables相同的库名。测试佐证仓库为 SQL Server Pipeline 连接器提供了较完整的测试覆盖可作为理解行为与验证配置的参考位于 src/testSqlServerPipelineITCase.java端到端 Pipeline 同步全流程验证SqlServerFullTypesITCase.java全类型字段的映射与数据往返验证与上文类型映射表一一对应SqlServerTablePatternMatchingITCase.javatables正则匹配与tables.exclude排除规则的行为验证SqlServerOnlineSchemaMigrationITCase.java在线表结构变更schema evolution场景验证SqlServerMetadataAccessorITCase.java数据库/schema/表元数据枚举与表结构获取验证SqlServerPipelineSavepointRestoreITCase.javasavepoint 恢复与scan.newly-added-table.enabled相关行为验证。如果你需要把 SQL Server 数据同步到其他下游可以参考 pipeline-connectors 目录下对应 Sink 连接器的配置文档将上文sink段替换为目标组件即可source段的配置保持不变。【免费下载链接】flink-cdcFlink CDC is a streaming data integration tool项目地址: https://gitcode.com/GitHub_Trending/flin/flink-cdc创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考