在快速发展的数据驱动时代，实时数据处理已经成为企业决策和运营的关键因素。特别是在处理来自各种数据源的信息时，如何确保数据的及时、准确和高效同步变得尤为重要。本文着重介绍了如何利用 SqlServer CDC 源连接器在 SeaTunnel 框架下实现 SQL Server 到其他数据系统的实时数据同步，这对于希望提升数据处理能力和实时数据分析的企业来说，具有重要的实践意义。

SQL Server CDC

SqlServer CDC 源连接器

支持 SQL Server 版本

• 服务器：2019（或更高版本，仅供参考）

支持引擎

SeaTunnel Zeta
Flink

主要特性

• [ ] 批处理
• [x] 流处理
• [x] 精确一次
• [ ] 列投影
• [x] 并行处理
• [x] 支持用户自定义分片

描述

SqlServer CDC 连接器允许从 SqlServer 数据库读取快照数据和增量数据。本文档描述了如何设置 SqlServer CDC 连接器以在 SqlServer 数据库上运行 SQL 查询。

支持的数据源信息

数据源	支持的版本	驱动	URL	Maven
SqlServer	服务器：2019（或更高版本，仅供参考）	com.microsoft.sqlserver.jdbc.SQLServerDriver	jdbc:sqlserver://localhost:1433;databaseName=column_type_test	下载

安装 Jdbc 驱动

请下载并将 SqlServer 驱动放在 ${SEATUNNEL_HOME}/lib/ 目录下。例如：cp mssql-jdbc-xxx.jar ${SEATUNNEL_HOME}/lib/

数据类型映射

SQL Server 数据类型	SeaTunnel 数据类型
CHAR VARCHAR NCHAR NVARCHAR STRUCT CLOB LONGVARCHAR LONGNVARCHAR	STRING
BLOB	BYTES
INTEGER	INT
SMALLINT TINYINT	SMALLINT
BIGINT	BIGINT
FLOAT REAL	FLOAT
DOUBLE	DOUBLE
NUMERIC DECIMAL(column.length(), column.scale().orElse(0))	DECIMAL(column.length(), column.scale().orElse(0))
TIMESTAMP	TIMESTAMP
DATE	DATE
TIME	TIME
BOOLEAN BIT	BOOLEAN

源选项

名称	类型	必需	默认值	描述
username	字符串	是	–	连接数据库服务器时使用的用户名。
password	字符串	是	–	连接数据库服务器时使用的密码。
database-names	列表	是	–	需要监控的数据库名。
table-names	列表	是	–	表名为模式名和表名的组合（databaseName.schemaName.tableName）。
base-url	字符串	是	–	必须包含数据库的URL，如 “jdbc:sqlserver://localhost:1433;databaseName=test”。
startup.mode	枚举	否	INITIAL	SqlServer CDC 消费者的可选启动模式，有效枚举为 “initial”、”earliest”、”latest” 和 “specific”。
startup.timestamp	长整型	否	–	从指定的纪元时间戳（以毫秒为单位）开始。 注意，当使用 “startup.mode” 选项为 ‘timestamp’ 时，此选项是必需的。
startup.specific-offset.file	字符串	否	–	从指定的 binlog 文件名开始。 注意，当 “startup.mode” 选项使用 'specific' 时，此选项是必需的。
startup.specific-offset.pos	长整型	否	–	从指定的 binlog 文件位置开始。 注意，当 “startup.mode” 选项使用 'specific' 时，此选项是必需的。
stop.mode	枚举	否	NEVER	SqlServer CDC 消费者的可选停止模式，有效枚举为 “never”。
stop.timestamp	长整型	否	–	从指定的纪元时间戳（以毫秒为单位）停止。 注意，当 “stop.mode” 选项使用 'timestamp' 时，此选项是必需的。
stop.specific-offset.file	字符串	否	–	从指定的 binlog 文件名停止。 注意，当 “stop.mode” 选项使用 'specific' 时，此选项是必需的。
stop.specific-offset.pos	长整型	否	–	从指定的 binlog 文件位置停止。 注意，当 “stop.mode” 选项使用 'specific' 时，此选项是必需的。
incremental.parallelism	整型	否	1	增量阶段中并行读取器的数量。
snapshot.split.size	整型	否	8096	表快照的分割大小（行数），快照期间的表会被分割成多个分片进行读取。
snapshot.fetch.size	整型	否	1024	读取表快照时每次轮询的最大提取量。
server-time-zone	字符串	否	UTC	数据库服务器中的会话时区。
connect.timeout	时长	否	30s	连接器尝试连接到数据库服务器后等待超时的最大时间。
connect.max-retries	整型	否	3	连接器尝试建立数据库服务器连接的最大重试次数。
connection.pool.size	整型	否	20	连接池大小。
chunk-key.even-distribution.factor.upper-bound	双精度浮点型	否	100	分块键分布因子的上界。此因子用于判断表数据是否均匀分布。如果计算出的分布因子小于或等于此上界值（即 (MAX(id) – MIN(id) + 1) / 行数），则表分块将被优化为均匀分布。否则，如果分布因子更大，则表将被认为是不均匀分布的，并且如果估计的分片数超过 sample-sharding.threshold 指定的值，将使用基于抽样的分片策略。默认值为 100.0。
chunk-key.even-distribution.factor.lower-bound	双精度浮点型	否	0.05	分块键分布因子的下界。此因子用于判断表数据是否均匀分布。如果计算出的分布因子大于或等于此下界值（即 (MAX(id) – MIN(id) + 1) / 行数），则表分块将被优化为均匀分布。否则，如果分布因子更小，则表将被认为是不均匀分布的，并且如果估计的分片数超过 sample-sharding.threshold 指定的值，将使用基于抽样的分片策略。默认值为 0.05。
sample-sharding.threshold	整型	否	1000	触发抽样分片策略的估计分片数阈值。当分布因子超出 chunk-key.even-distribution.factor.upper-bound 和 chunk-key.even-distribution.factor.lower-bound 指定的范围，并且估计的分片数（计算为近似行数 / 分块大小）超过此阈值时，将使用抽样分片策略。这可以帮助更有效地处理大型数据集。默认值为1000分片。
inverse-sampling.rate	整型	否	1000	抽样分片策略中使用的抽样率的倒数。例如，如果这个值设置为1000，意味着抽样过程中应用了1/1000的抽样率。这个选项提供了在控制抽样粒度的灵活性，从而影响最终的分片数量。特别是在处理非常大的数据集时，更低的抽样率是首选。默认值为1000。
exactly_once	布尔型	否	true	启用精确一次语义。
debezium.*	配置	否	–	将Debezium的属性传递给用于从SqlServer服务器捕获数据变化的Debezium嵌入式引擎。 查看Debezium的SqlServer连接器属性获取更多信息
format	枚举	否	DEFAULT	SqlServer CDC 的可选输出格式，有效枚举为 “DEFAULT”、”COMPATIBLE_DEBEZIUM_JSON”。
common-options		否	–	源插件的通用参数，请参考源通用选项获取详细信息。

任务示例

初始读取简单示例

这是一个流模式CDC初始化读取的示例，成功读取表数据后将进行增量读取。以下SQL DDL仅供参考。

env {
  # 在此处设置引擎配置
  execution.parallelism = 1
  job.mode = "STREAMING"
  execution.checkpoint.interval = 5000
}

source {
  # 仅用于测试和演示功能的示例源插件
  SqlServer-CDC {
    result_table_name = "customers"
    username = "sa"
    password = "Y.sa123456"
    startup.mode="initial"
    database-names = ["column_type_test"]
    table-names = ["column_type_test.dbo.full_types"]
    base-url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
  }
}

transform {
}

sink {
  console {
    source_table_name = "customers"
  }
  

增量读取简单示例

这是一个增量阅读示例，用于阅读变更数据并打印。

env {
  # 在此处设置引擎配置
  execution.parallelism = 1
  job.mode = "STREAMING"
  execution.checkpoint.interval = 5000
}

source {
  # 仅用于测试和演示功能的示例源插件
  SqlServer-CDC {
    # 设置精确一次读取
    exactly_once=true 
    result_table_name = "customers"
    username = "sa"
    password = "Y.sa123456"
    startup.mode="latest"
    database-names = ["column_type_test"]
    table-names = ["column_type_test.dbo.full_types"]
    base-url = "jdbc:sqlserver://localhost:1433;databaseName=column_type_test"
  }
}

transform {
}

sink {
  console {
    source_table_name = "customers"
  }
}

随着数据处理需求的不断增长和实时数据同步的重要性日益凸显，SqlServer CDC 源连接器在 SeaTunnel 生态系统中扮演着至关重要的角色。

通过本文的深入解析，我们希望您能够更好地理解并利用这一强大工具，从而实现数据流的高效、稳定和精准同步。

无论您是数据工程师、系统架构师还是业务分析师，掌握如何在 SeaTunnel 中部署和优化 SQL Server CDC 连接器，都将为您的数据处理能力带来显著提升。

本文由 白鲸开源科技 提供发布支持！