# 如何解决CDC同步作业表主键分布不均导致CDC卡顿、内存溢出问题
# 问题背景
场景：使用SQLServer-CDC进行数据实时抽取到StarRocks
SeaTunnel版本：DWS720
SQLServer版本：SQLServer2019
StarRocks版本：3.2.9
源表数据量：167W
# 问题现象
使用DWS数据同步CDC组件从业务库抽数到数仓（StarRocks），已经成功使用SQLServer-CDC做了大量的实时同步。 但最近在抽一个SQLServer表的时候遇到了异常情况，作业启动之后，日志显示读写数量一直为0，且长时间不停止，运行一段时间后发现内存溢出,作业异常停止。
作业模型如下：
{
"env": {
"shade.identifier": "base64",
"job.mode": "STREAMING",
"parallelism": 1,
"job.name": "sqlserver-cdc2doris_1776852127901",
"checkpoint.interval": 5000
},
"source": [
{
"username": "c2E=",
"password": "******",
"plugin_name": "SqlServer-CDC",
"result_table_name": "CDCSource1776412849012",
"node_code": "CDCSource1776412849012",
"connection_code": "sqlserver2019",
"database-names": [
"test_li_dws"
],
"table-names": [
"test_li_dws.dbo.tbl_PlannedOrderDetail"
],
"table-names-config": [
{
"table": "test_li_dws.dbo.tbl_PlannedOrderDetail",
"primaryKeys": [
"PODID"
],
"snapshotSplitColumn": "PODID"
}
],
"base-url": "jdbc:sqlserver://192.168.16.181:1433;databaseName=test_li_dws;trustServerCertificate=true",
"startup.mode": "initial",
"incremental.parallelism": 1,
"exactly_once": true,
"format": "default",
"schema-changes.enabled": false
}
],
"transform": [],
"sink": [
{
"username": "ZHdz",
"password": "******",
"database": "demo_sink",
"plugin_name": "StarRocks",
"source_table_name": "CDCSource1776412849012",
"source_node_code": "CDCSource1776412849012",
"node_code": "StarRocksSink1776412883181",
"connection_code": "StarRocks146",
"nodeUrls": [
"192.168.0.146:8030"
],
"base-url": "jdbc:mysql://192.168.0.146:9030?nullCatalogMeansCurrent=true&allowLoadLocalInfile=true&useSSL=false",
"batch_max_rows": 1024,
"batch_max_bytes": 5242880,
"enable_upsert_delete": false,
"schema_save_mode": "CREATE_SCHEMA_WHEN_NOT_EXIST",
"data_save_mode": "DROP_DATA",
"field_ide": "ORIGINAL",
"tableMapping": [
{
"source_table_name": "tbl_PlannedOrderDetail",
"table_name": "tbl_PlannedOrderDetail",
"save_mode_create_template": "CREATE TABLE `${database}`.`${table}` \n ( \n `PODID`,\n ${rowtype_fields} \n) ENGINE=OLAP\n PRIMARY KEY (`PODID`)\n COMMENT '${comment}'\n DISTRIBUTED BY HASH (`PODID`)"
}
]
}
]
}
# 关键日志
INFO [AbstractJdbcSourceChunkSplitter] - Splitting table test_li_dws.dbo.tbl_PlannedOrderDetail into chunks, split column: PODID, min: C317D088-40EC-4173-B938-000002777154, max: 0FC8A607-9805-4978-9256-FFFFFE3BD93F, chunk size: 8096, distribution factor upper: 100.0, distribution factor lower: 0.05, sample sharding threshold: 1000
INFO [AbstractJdbcSourceChunkSplitter] - The distribution factor of table test_li_dws.dbo.tbl_PlannedOrderDetail is 5.52297726757771E12 according to the min split key C317D088-40EC-4173-B938-000002777154, max split key 0FC8A607-9805-4978-9256-FFFFFE3BD93F and approximate row count 1670000
INFO [AbstractJdbcSourceChunkSplitter] - Use unevenly-sized chunks for table test_li_dws.dbo.tbl_PlannedOrderDetail, the chunk size is 8096
# 根本原因分析
核心问题：UUID 主键 + 分片算法决策路径不匹配，导致分片阶段卡死或内存溢出。
通过阅读SeaTunnel源码，追踪整个分片算法的执行路径如下：
# 1. 分片算法整体流程
generateSplits()
├── getSplitColumn() // 选择分片列
│ ├── 检查用户配置的 snapshotSplitColumn
│ ├── 检查主键第一列
│ └── 检查唯一键第一列
│ （优先选择数值类型：TINYINT > SMALLINT > INT > BIGINT > DECIMAL > STRING）
│
├── splitTableIntoChunks() // 执行分片
│ ├── queryMinMax() // 查询 MIN/MAX 值
│ ├── isEvenlySplitColumn() // 判断是否为均匀分布列
│ │ （STRING 类型返回 true，UUID 被视为 STRING）
│ ├── calculateDistributionFactor() // 计算分布因子
│ │ （STRING 类型返回 Long.MAX_VALUE → 因子极大）
│ │
│ ├── [均匀分布] → splitEvenlySizedChunks()
│ │ └── ObjectUtils.plus() // 步长递增（UUID不会走此路径）
│ │
│ ├── [不均匀分布] → 检查 shardCount vs sampleShardingThreshold
│ │ ├── shardCount > threshold → sampleDataFromColumn()
│ │ │ └── skipReadAndSortSampleData() // 全表扫描采样
│ │ │
│ │ └── shardCount <= threshold → splitUnevenlySizedChunks()
│ │ └── queryNextChunkMax() // 迭代查询分片边界
│ │
│ └── [非均匀列] → splitUnevenlySizedChunks()
│
└── 返回 List<SnapshotSplit>
# 2. UUID 类型映射链
SQLServer UNIQUEIDENTIFIER → SeaTunnel STRING 类型 → 被视为"可均匀分片"
源码位置：
connector-jdbc/.../sqlserver/SqlServerTypeConverter.javacase SQLSERVER_UNIQUEIDENTIFIER: builder.sourceType(SQLSERVER_UNIQUEIDENTIFIER); builder.dataType(BasicType.STRING_TYPE);源码位置：
connector-cdc-base/.../splitter/JdbcSourceChunkSplitter.java:166-179default boolean isEvenlySplitColumn(Column splitColumn) { switch (fromDbzColumn(splitColumn).getSqlType()) { case TINYINT: case SMALLINT: case INT: case BIGINT: case DECIMAL: case STRING: // ← UUID 映射为 STRING，返回 true return true; default: return false; } }
# 3. 分布因子计算失效
源码位置：connector-cdc-base/.../utils/ObjectUtils.java:72-73
} else if (minuend instanceof String) {
return BigDecimal.valueOf(Long.MAX_VALUE); // 9223372036854775807
}
UUID 作为 String 类型传入 minus() 方法，直接返回 Long.MAX_VALUE，而非真实的差值。
分布因子计算公式（AbstractJdbcSourceChunkSplitter.java:359-363）：
distributionFactor = (MAX - MIN + 1) / rowCount
= (Long.MAX_VALUE + 1) / 1670000
≈ 5.52 × 10¹²
远超 distributionFactorUpper（默认 100.0），数据被判定为分布不均匀。
# 4. 分片路径决策
计算分片数量：
shardCount = approximateRowCnt / snapshot.split.size
源码中的决策逻辑（AbstractJdbcSourceChunkSplitter.java:168-197）：
if (sampleShardingThreshold < shardCount) {
→ 采样分片路径 (全表扫描)
} else {
→ 非均匀分片路径 (迭代查询)
}
# 5. 当前配置下的决策路径
默认配置参数：
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| snapshot.split.size | Integer | 8096 | 表快照的分割大小（行数），读取表快照时，捕获的表会被分割为多个分割。 |
| snapshot.fetch.size | Integer | 1024 | 读取表快照时每次轮询的最大获取大小。 |
| connection.pool.size | Integer | 20 | 连接池大小。 |
| chunk-key.even-distribution.factor.upper-bound | Double | 100 | 分块键分布因子的上界。此因子用于确定表数据是否均匀分布。如果计算的分布因子小于或等于此上界（即，(MAX(id) - MIN(id) + 1) / 行数），表分块将被优化以实现均匀分布。否则，如果分布因子较大，如果估计的分片数超过 sample-sharding.threshold 指定的值，表将被视为不均匀分布并使用基于采样的分片策略。默认值为 100.0。 |
| chunk-key.even-distribution.factor.lower-bound | Double | 0.05 | 分块键分布因子的下界。此因子用于确定表数据是否均匀分布。如果计算的分布因子大于或等于此下界（即，(MAX(id) - MIN(id) + 1) / 行数），表分块将被优化以实现均匀分布。否则，如果分布因子较小，如果估计的分片数超过 sample-sharding.threshold 指定的值，表将被视为不均匀分布并使用基于采样的分片策略。默认值为 0.05。 |
| sample-sharding.threshold | int | 1000 | 此配置指定了触发采样分片策略的估计分片数阈值。当分布因子超出 chunk-key.even-distribution.factor.upper-bound 和 chunk-key.even-distribution.factor.lower-bound 指定的范围，并且估计的分片数（计算为近似行数 / 分块大小）超过此阈值时，将使用采样分片策略。这可以帮助更有效地处理大型数据集。默认值为 1000 分片。 |
| inverse-sampling.rate | int | 1000 | 采样分片策略中使用的采样率的倒数。例如，如果此值设置为 1000，则意味着在采样过程中应用 1/1000 的采样率。此选项提供了控制采样粒度的灵活性，从而影响最终的分片数量。对于非常大的数据集，首选较低的采样率时，此选项特别有用。默认值为 1000。 |
计算过程：
shardCount = 1670000 / 8096 = 206
sampleShardingThreshold = 1000
206 < 1000 → 非均匀分片路径
# 为什么读写数量一直为 0？
作业卡在 generateSplits() 的 split 计算阶段或快照数据读取阶段，导致读写计数保持为 0。
根据不同的参数配置，存在两种卡顿场景：
# 场景一：非均匀分片路径（默认配置）
当使用默认配置（snapshot.split.size=8096，sample-sharding.threshold=1000）时，shardCount=206 < 1000，走非均匀分片路径。
Split 计算阶段执行约 206 次迭代查询（SqlServerUtils.java:227-258）：
SELECT MAX([PODID]) FROM (
SELECT TOP (8096) [PODID] FROM [table] WHERE [PODID] >= ? ORDER BY [PODID] ASC
) AS T
此路径 Split 计算本身可以完成，但后续的快照数据读取阶段存在以下问题：
- UUID 聚簇索引碎片化：SQL Server 中 UUID 作为聚簇索引键（主键默认聚簇），非连续 UUID 导致索引碎片化严重。每次范围查询
[PODID] >= ? AND [PODID] <= ?需要大量随机 I/O。 - 参数类型隐式转换：
ps.setObject(1, includedLowerBound)传入 Java String 类型，SQL Server 对uniqueidentifier列做隐式转换，可能导致索引扫描效率降低。 - 单线程串行读取：
parallelism=1时所有 206 个分片串行读取，每个分片最多读 8096 行，总计 167 万行的 UUID 范围扫描非常耗时。
Split 计算完成后，数据读取阶段由于上述原因速度极慢，长时间看不到读写数量增长。
# 场景二：采样分片路径（低 threshold 配置）
当 sample-sharding.threshold 设置过小（如 100），导致 shardCount > threshold 时，触发采样分片路径。
SqlServerChunkSplitter 的 sampleDataFromColumn() 方法委托给 SqlServerUtils.skipReadAndSortSampleData()（SqlServerUtils.java:170-221）：
public static Object[] skipReadAndSortSampleData(
JdbcConnection jdbc, TableId tableId, String columnName, int inverseSamplingRate) {
final String sampleQuery = String.format("SELECT %s FROM %s", quote(columnName), quote(tableId));
Statement stmt = jdbc.connection()
.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
stmt.setFetchSize(1024);
ResultSet rs = stmt.executeQuery(sampleQuery);
List<Object> results = new ArrayList<>();
while (rs.next()) {
count++;
if (count % inverseSamplingRate == 0) {
results.add(rs.getObject(1)); // 累积采样 UUID
}
}
Object[] resultsArray = results.toArray();
Arrays.sort(resultsArray); // 排序额外内存开销
return resultsArray;
}
执行流程：
- 执行
SELECT [PODID] FROM [test_li_dws].[dbo].[tbl_PlannedOrderDetail]—— 全表扫描 167 万行 - JDBC 驱动在
TYPE_FORWARD_ONLY + fetchSize=1024模式下流式传输所有行 - 每 1000 行保留一个采样值（
inverseSamplingRate=1000），约保留 1670 个 UUID - 对采样结果排序
Arrays.sort(resultsArray) - Split 计算迟迟无法完成 → 读写计数保持为 0
# 为什么会内存溢出？
内存溢出由以下因素叠加导致：
因素 1 - UUID 索引碎片化导致全表扫描： UUID 作为聚簇索引键，插入时数据页频繁分裂，索引碎片化导致查询性能急剧下降。在碎片化严重的情况下，原本应走索引的范围查询退化为近似全表扫描，大量数据页被加载到内存。
因素 2 - 采样路径全表扫描（场景二）：
当触发采样路径时，skipReadAndSortSampleData() 全表扫描 167 万行：
List<Object> results = new ArrayList<>(); // 累积约 1670 个 UUID 字符串
while (rs.next()) { ... } // 迭代 167 万次
Object[] resultsArray = results.toArray();
Arrays.sort(resultsArray); // 排序额外内存开销
JDBC 驱动在 TYPE_FORWARD_ONLY 模式下内部缓冲数据，叠加应用层对象创建，GC 压力持续增大。
因素 3 - Checkpoint 超时连锁反应：
默认 checkpoint.timeout 为 30s,checkpoint.interval为5s ，如果 split 阶段或快照读取阶段超时未完成，checkpoint 失败触发作业重启。重启后重新执行 split 计算，反复全表扫描，最终导致 OOM。
因素 4 - 快照数据读取内存累积：
Split 完成后进入快照数据读取阶段，每个分片的数据通过 SnapshotSplitReadTask 加载。如果 Sink（如 StarRocks）写入速度慢于 Source 读取速度，数据在内存中累积。167 万行 UUID + 业务列数据，在 parallelism=1 串行模式下内存占用持续增长。
# 优化方案
# 方案一：调整分片参数 + 增加 JVM 内存（推荐，无需改表结构）
核心思路：
- 确保走非均匀分片路径（
shardCount < sample-sharding.threshold），避免全表扫描采样 - 适当减小
snapshot.split.size，缩小单次查询范围，降低内存峰值 - 增大
snapshot.fetch.size，减少网络往返 - 增加 JVM 内存，应对 UUID 索引碎片化导致的内存压力
{
"source": [
{
"plugin_name": "SqlServer-CDC",
"snapshot.split.size": 5000,
"sample-sharding.threshold": 10000,
"inverse-sampling.rate": 2000,
"snapshot.fetch.size": 2048,
"exactly_once": true
}
]
}
参数计算验证：
shardCount = 1670000 / 5000 = 334
sampleShardingThreshold = 10000
334 < 10000 → 走非均匀分片路径 ✓
关键参数说明：
| 参数 | 推荐值 | 默认值 | 作用 |
|---|---|---|---|
| snapshot.split.size | 5000 | 8096 | 每个分片的行数，UUID 场景建议适当调小，缩小单次查询范围 |
| sample-sharding.threshold | 10000 | 1000 | 调大以确保 shardCount < threshold，避免触发采样全表扫描 |
| inverse-sampling.rate | 2000 | 1000 | 采样率的倒数（仅在触发采样路径时生效），增大以减少内存占用 |
| snapshot.fetch.size | 2048 | 1024 | JDBC 每次轮询的获取大小，增大可减少网络往返次数 |
| exactly_once | true | false | 确保分片时必须有主键 |
原理：非均匀分片使用 queryNextChunkMax() 迭代查询，每次只查一个分片边界：
SELECT MAX([PODID]) FROM (
SELECT TOP (5000) [PODID] FROM [table] WHERE [PODID] >= ? ORDER BY [PODID] ASC
) AS T
- UUID 有聚簇索引（主键），
ORDER BY + TOP可走索引，单次查询效率高 - 334 次迭代，每次查询一个分片边界，不会 OOM
snapshot.split.size=5000相比默认的 8096，减小了每次数据读取的内存峰值- 每 10 次查询 sleep 100ms（
AbstractJdbcSourceChunkSplitter.java:481），总耗时可控
# 方案二：设置快照切分列（如果表除了主键还有其他数值型唯一键）
如果表中有自增列、时间列或其他数值类型的唯一索引列，可指定替代分片列：
{
"table-names-config": [
{
"table": "test_li_dws.dbo.tbl_PlannedOrderDetail",
"primaryKeys": [
"PODID"
],
"snapshotSplitColumn": "ID"
}
]
}
注意事项：
snapshotSplitColumn必须是唯一键（主键或唯一索引）中的列- 支持的类型：TINYINT / SMALLINT / INT / BIGINT / DECIMAL / STRING
- 优先选择数值类型（优先级从高到低：TINYINT > SMALLINT > INT > BIGINT > DECIMAL > STRING）
- 数值类型分片列可走均匀分片路径，性能最优
# 方案三：增加 JVM 内存 + checkpoint 超时
当无法避免采样路径或 split 阶段耗时较长时，增大资源上限：
修改 $SEATUNNEL_HOME/config/jvm_options，增加 JVM 内存：
# JVM Heap
-Xms16g
-Xmx16g
增加 checkpoint 超时时间，给 split 阶段留足时间：
{
"env": {
"checkpoint.timeout": "600000",
"checkpoint.interval": 10000
}
}
# 推荐组合配置
针对 167 万行 UUID 主键表的最优配置（在原始作业模型基础上增加优化参数）：
{
"env": {
"shade.identifier": "base64",
"job.mode": "STREAMING",
"parallelism": 2,
"checkpoint.interval": 10000,
"checkpoint.timeout": "600000",
"job.name": "sqlserver-cdc-uuid-optimized"
},
"source": [
{
"plugin_name": "SqlServer-CDC",
"username": "***",
"password": "***",
"base-url": "jdbc:sqlserver://host:1433;databaseName=db;trustServerCertificate=true",
"database-names": [
"your_db"
],
"table-names": [
"your_db.dbo.your_table"
],
"table-names-config": [
{
"table": "your_db.dbo.your_table",
"primaryKeys": [
"PODID"
]
}
],
"startup.mode": "initial",
"incremental.parallelism": 1,
"snapshot.split.size": 5000,
"sample-sharding.threshold": 10000,
"inverse-sampling.rate": 2000,
"snapshot.fetch.size": 2048,
"exactly_once": true,
"format": "default",
"schema-changes.enabled": false
}
],
"sink": [
{
"plugin_name": "StarRocks",
"username": "***",
"password": "***",
"database": "demo_sink",
"nodeUrls": [
"192.168.0.146:8030"
],
"base-url": "jdbc:mysql://192.168.0.146:9030?nullCatalogMeansCurrent=true&allowLoadLocalInfile=true&useSSL=false",
"batch_max_rows": 2048,
"batch_max_bytes": 10485760,
"enable_upsert_delete": false,
"schema_save_mode": "CREATE_SCHEMA_WHEN_NOT_EXIST",
"data_save_mode": "DROP_DATA",
"field_ide": "ORIGINAL"
}
]
}
CDC可选参数配置：
checkpoint参数配置
配置要点总结：
| 配置项 | 原始值 | 优化值 | 说明 |
|---|---|---|---|
parallelism | 1 | 2 | 适当增大并行度，加快分片读取速度 |
checkpoint.timeout | 未设置 | 600000 | 给 split + 快照读取阶段留足时间，避免超时重启 |
snapshot.split.size | 8096（默认） | 5000 | 减小分片大小，降低单次查询内存峰值 |
sample-sharding.threshold | 1000（默认） | 10000 | 确保 shardCount < threshold，避免触发采样全表扫描 |
inverse-sampling.rate | 1000（默认） | 2000 | 增大采样间隔（仅在触发采样时生效） |
snapshot.fetch.size | 1024（默认） | 2048 | 增大 JDBC fetch size，减少网络往返 |
batch_max_rows | 1024 | 2048 | Sink 批量提交行数，适当增大提升写入吞吐 |
batch_max_bytes | 5242880 | 10485760 | Sink 批量提交大小，从 5MB 增大到 10MB |
# 参数调优速查表
以下为不同数据量级下 UUID 主键表的推荐参数配置：
| 数据量 | snapshot.split.size | sample-sharding.threshold | 说明 |
|---|---|---|---|
| < 50 万 | 8096 | 1000 | 默认配置即可，shardCount < threshold |
| 50 万 ~ 200 万 | 5000 | 10000 | shardCount < threshold，走非均匀分片 |
| 200 万 ~ 500 万 | 4000 | 20000 | 增大 threshold 避免采样 |
| 500 万 ~ 1000 万 | 3000 | 50000 | 增大 threshold + 减小 split size |
| > 1000 万 | 2000 | 100000 | 配合增大 JVM 内存 (16G+) |
计算公式：
shardCount = 总行数 / snapshot.split.size
要求：shardCount < sample-sharding.threshold
即：sample-sharding.threshold > 总行数 / snapshot.split.size
# 源码分析参考
以下为本次分析涉及的关键源码文件路径：
| 文件 | 路径 | 关键作用 |
|---|---|---|
| AbstractJdbcSourceChunkSplitter | connector-cdc-base/.../splitter/AbstractJdbcSourceChunkSplitter.java | 分片算法核心：均匀/非均匀/采样三种策略 |
| JdbcSourceChunkSplitter | connector-cdc-base/.../splitter/JdbcSourceChunkSplitter.java | 接口定义，包含 isEvenlySplitColumn() 判断逻辑 |
| ObjectUtils | connector-cdc-base/.../utils/ObjectUtils.java | 数值运算工具，minus() 对 String 返回 Long.MAX_VALUE |
| SqlServerChunkSplitter | connector-cdc-sqlserver/.../enumerator/SqlServerChunkSplitter.java | SQLServer 分片实现，委托 SqlServerUtils |
| SqlServerUtils | connector-cdc-sqlserver/.../utils/SqlServerUtils.java | SQLServer SQL 构建与执行，包含采样和分片查询方法 |
| SqlServerTypeConverter | connector-jdbc/.../sqlserver/SqlServerTypeConverter.java | 类型转换：UNIQUEIDENTIFIER → STRING |
| JdbcSourceOptions | connector-cdc-base/.../option/JdbcSourceOptions.java | 分片相关配置项定义 |