Incr-Sync/Full-Import 调优

更新时间：2026-04-14 15:40:56

TPS

源端每秒拉取到的消息数量。
延时

当前任务的延时时长，单位为秒。延时计算不考虑安全位点。
ReadQ

在异步传输时，中间框架通过 ReadQ 获取数据并将数据写入目标端。当前进程未被中间框架获取的数据将被缓存在 ReadQ 中。ReadQ 的默认最大值是 4096。如果值比较小，说明源端可能没有数据或者源端拉取存在瓶颈。
WriteConsume

写入一次 {batch.size} 的时间，单位毫秒。WriteConsume 的数值越低，说明目标端性能越好。

查询 Metrics 信息

执行 ./connector_utils.sh metrics 命令，即可查询 Metrics 信息。

./connector_utils.sh metrics

2022-09-22 12:49:48.876
SOURCE: [RPS:0.0, IOPS:0.0M, delay:1489ms]
SINK: [RPS:0.0, TPS:0.0, IOPS:0.0M, delay:2986440ms]
SINK_TIME: [execute_time:0.0ms/record, commit_time:0.0ms/batch]
SINK_SLOW_ROUTES:
SINK_THREAD: 4/4
DISPATCHER: wait record:0, ready batch:0, shardTime:nullms/record
forward_slot0 batchAccumulate: 0, recordAccumulate: 0
queue_slot1 batchAccumulate: 0, recordAccumulate: 0
heap:620M/1945M, noHeap:52M/488M, threadCount:18, cpu:0.054, sysCpu:51.555
ParNew(count:0, cost:0) ConcurrentMarkSweep(count:0, cost:0)

说明如下：

SOURCE RPS/IOPS/DELAY
SINK RPS/TPS(RecordBatch/s)/IOPS/DELAY
SINK_TIME: execute_time 表示每个 record 的执行时间，commit_time 表示 recordBatch 的执行时间。
SINK_SLOW_ROUTES：内部统计下，部分执行较慢的 SINK_ROUTES 信息，SINK_ROUTE 表示可并行写入单元。例如，kafka 的 partition、datahub 的 shard、rocketmq 的队列。
SINK_THREAD：当前活跃 sinkThread / 最大 sinkThread。如果 sinkThread 较少证明 sink 端较为空闲，未达到瓶颈。
DISPATCHER 中间队列情况，wait record: 表示等待被分配的消息数，ready batch 表示等待 sink thread 执行的 record batch 数量。

如果 wait record 数量较多，证明消息数量较大，可能受到 GC 影响。

如果 ready batch 较多，则证明 sink 端存在瓶颈，可以尝试提高 sink 写入速率（例如，增加线程）。
{框架队列名称} batchAccumulate: {积攒 recordBatch 数量}, recordAccumulate: {积攒 record 数量}。
1. 如果第一个队列 batchAccumulate 为空则证明当前瓶颈在 source 端无消息进入。
2. 如果最后一个队列 batchAccumulate 满则证明瓶颈在 RecordDispatcher(冲突矩阵/散列) 中，此时需要看 DISPATCHER 的 metrics。
堆内存使用/堆内存最大，非堆内存使用/非堆内存最大，线程数，进程 cpu && sys cpu。
{时间} {youngGcName}(count:{累计次数}, cost:{累计耗时}) {fullGcName}(count:{累计次数}, cost:{累计耗时})。

诊断 Incr-Sync/Full-Import

查询 Incr-Sync/Full-Import 组件的组件 ID。
1. 登录 OMS 社区版控制台。
2. 在左侧导航栏，单击 数据迁移。
3. 在 迁移项目列表 页面，单击目标数据迁移项目的名称，进入详情页面。
4. 单击页面右上角的 查看组件监控。
5. 在 查看组件监控 对话框，查看 Incr-Sync 增量同步组件 或 Full-Import 全量导入组件 的 组件 ID。
进入 Incr-Sync/Full-Import 组件目录。
1. 登录 OMS 社区版部署服务器。
2. 进入 Docker 容器。
```
docker exec -it ${CONTAINER_NAME} bash
```
3. 执行进入组件目录的命令。
```
cd /home/ds/run/${组件 ID}
```

在 Incr-Sync/Full-Import 组件目录下执行 ./connector_utils.sh diagnose，即可进行诊断。

./connector_utils.sh diagnose -s 'YYYY-MM-DDTHH:mm:ss' -e 'YYYY-MM-DDTHH:mm:ss'

上述命令中，-s 和 -e 均为可选参数，-s 表示开始分析日志的时间，-e 表示结束分析日志的时间。时间格式为 'YYYY-MM-DDTHH:mm:ss'，例如 '2023-06-01T12:00:00'。

./connector_utils.sh diagnose 的默认值为 10 分钟（-e 默认值为当前时间）。

返回结果示例如下：

[Metrics]
TPS: [last:345,avg:277.28,p99:911.00]
RPS: [last:106,avg:257.08,p99:968.00]
IOPS: [last:2KB,avg:21.33KB]
EXECUTE_TIME: [last:34ms,avg:220.44ms,p99:783.00ms]
SINK_DELAY: [last:19ms,avg:260.31ms,p99:819.00ms]
SOURCE_DELAY: [
source_subtopic2_source_delay: [last:702ms,avg:525.00ms,p99:986.00ms]
source_subtopic1_source_delay: [last:14ms,avg:490.69ms,p99:973.00ms]
]
QUEUE_BATCH_ACCUMULATE: [
frame_queue_slot_1.batchAccumulate: [last:420,avg:496.00,p99:975.00]
frame_queue_slot_2.batchAccumulate: [last:310,avg:470.05,p99:975.00]
]
JVM-MEM: heap:34.28M/3641M, noHeap:19.38M/0M]
THREAD: [count:4, sink:14/16]
CPU: [last:17,avg:27.95,p99:62.00]
[Pref]
sink block: true
youngGc: true
[Suggest]
config[coordinator.shuffleMinBatchSize]:20 to 40
config[coordinator.shuffleMaxBatchSize]:40 to 80
jvm to: -Xmx4096m

说明如下：

Metrics 信息为判断依据。
Pref 信息为根据 Metrics 判断分析的瓶颈点。
Suggest 信息是推进优化的点。例如上述示例中可以更新 Incr-Sync/Full-Import 组件 coordinator 中的 shuffleMinBatchSize、shuffleMaxBatchSize 和 connectorJvmParam 参数。

workerNum

workerNum 已达到上限，metrics 日志中 sink 的 executeTime 和 commitTime 的时间可以接受。
1. 从项目详情页面进入 查看组件监控 对话框。
2. 单击目标组件后的更新。
3. 在 更新配置 对话框，鼠标悬停至 sink > workerNum 参数，单击编辑图标。
  
  如果没有该参数，请将鼠标悬停至 sink 参数旁边的空白处，单击添加图标。
4. 根据机器资源，调大 workerNum 的值。
5. 在文本框中输入修改后的参数后，单击确认图标。
6. 在 更新配置 对话框，单击确定。
workerNum 未达到上限，metrics 日志中每两次之间的 GC 时间很长。
1. 从项目详情页面进入 查看组件监控 对话框。
2. 单击目标组件后的更新。
3. 在 更新配置 对话框，鼠标悬停至 source > splitThreshold 参数，单击编辑图标。
  
  如果没有该参数，请将鼠标悬停至 source 参数旁边的空白处，单击添加图标。
4. splitThreshold 参数的默认值为 128，调小该参数的值。
5. 在文本框中输入修改后的参数后，单击确认图标。
6. 在 更新配置 对话框，单击确定。
workerNum 只有 1 或 2，connector.log 打印 conflictKey 或者 deepSize 关键字。
1. 从项目详情页面进入 查看组件监控 对话框。
2. 单击目标组件后的更新。
3. 在 更新配置 对话框，鼠标悬停至 coordinator 参数旁边的空白处，单击添加图标。
4. 输入需要添加的 Key Name 为 hotKeyMerge，单击对号。
5. 在 更新配置 对话框中，找到新增的 Key Name，其默认值为 NULL。
6. 鼠标悬停至新增参数，单击显示的编辑图标，修改参数值为 true，单击确认图标。
7. 在 更新配置 对话框，单击确定。

GC 时间过长

说明

此处 GC 时间过长，指每秒 Young GC 超过 300ms，每秒都存在 Full GC。

查看 GC

在 task 目录下执行以下命令，查看每秒的 GC 的详情：

/opt/alibaba/java/bin/jstat -gcutil `cat task.pid` 1s

增大 JVM 内存：调整 coordinator > connectorJvmParam 为 -Xms12g -Xmx16g。

此处仅为示例，您需要参考当前机器的内存调整。如果之前参数存在，您可以去掉 -Xmn。
同步 && 全量：减小 coordinator > bridgeQueueSize 的值。默认值为 256，可以减小到 32。
同步 Kafka：设置 sink > lingerMs 参数的值为 1。
设置 coordinator > throttleMemoryBound 参数的值为 bytes 字节数来限制内存。建议将该参数设置为最大内存的 1/4。

例如，最大堆内存是 16G，则该值为 16 * 1024 * 1024 * 1/4 = 4294967296。
如果 conf/coordinator.json 或者 conf_new/coordinator.json 中 dispatcherClassName 配置项是 ShuffleRecordDispatcher，则可以更改 coordinator 的如下配置项：
- maxRecordCapacity = 1000 控制 dispatcher 队列总数。默认是根据 shuffleMinBatchSize *(shuffleBucketSize* 1.5) = 3840。
- 设置 shuffleBucketSize 参数的值为 32。减少攒批桶数量，默认为 128。
- 设置 shuffleFlushIntervalMs 参数的值为 10，加速推送到 Sink 端。
增加 sink > workerNum 参数，默认值为 16，最大可以调整到 64。

攒批相关参数

注意

攒批相关参数仅适用于非数据库目标端的增量同步链路。

参考 GC 情况：

在 GC 不严重时，在目的端能承受的范围能增加攒批能力。
在 GC 验证时，减少攒批。
- maxRecordCapacity：攒批队列最大数量，默认 16000。
- shuffleMinBatchSize：攒批最小批次，默认 20。
- shuffleMaxBatchSize：攒批最大批次，默认 64。
- shuffleFlushIntervalMs：刷新时间，默认 100ms。
- shuffleBucketMaxInFlightBatchSize：每个并发允许最多正在投递数量，增量默认 1，全量不限制。
- shuffleBucketSize：允许最大攒批并发。

在 shuffleMaxBatchSize || shuffleFlushIntervalMs 达成的条件下即会投递（写入线程还具备写入能力）。

延迟所需的信息说明

当前项目延迟的截图，包括项目延迟、Store 延迟和 Incr-Sync 延迟。

latency-zh

如上图所示，在一个数据迁移或数据同步项目中，存在多种延迟概念。

序号	延迟类型	描述
1	项目延迟	数据迁移或数据同步项目的延迟时间以 Incr-Sync 延迟时间为准。如果存在多个 Incr-Sync，以最长的 Incr-Sync 延迟时间为准。数据迁移或数据同步项目的延迟时间计算和组件的延迟时间计算本身属于不同的调度存储，可能存在差异。例如，项目数量过多、调度时间过长等可能导致项目延迟时间大于组件延迟时间，但仅是展示延迟时间的差别。
2	Store 延迟	“当前时间 - Store 抓取变更记录的时间”得到的时间差即 Store 组件的延迟时间。该时间计算存在轮询，通常为 10~30 秒。
3	Incr-Sync 延迟	“当前时间 - 当前写入至目标端记录的最小变更时间”得到的时间差即 Incr-Sync 组件的延迟时间。该时间计算存在轮询，轮询时间请参考括号内多少秒前更新的时间。

Incr-Sync 组件的 Metrics 日志记录，详情请参见本文《查询 Metrics 信息》模块的内容。
建议您将 Incr-Sync 组件的 logs 和 conf 目录打包提供。