Flink CDC里为什么10分钟才600多条 这么慢的吗？

Flink CDC 读取速度较慢可能与以下几个因素有关，以下是详细分析和可能的解决方案：

1. Checkpoint 配置问题

Flink CDC 的读取速度可能受到 Checkpoint 配置的影响。如果 execution.checkpointing.interval 设置过大（例如 10 分钟），会导致作业在全量阶段或增量阶段的读取效率降低，尤其是在全量数据读取完成后切换到增量阶段时需要等待一个完整的 Checkpoint 周期。

解决方案：

• 调整 Checkpoint 时间间隔：将 execution.checkpointing.interval 设置为更短的时间（例如 1 分钟或 30 秒），以减少等待时间。

  execution.checkpointing.interval: 1min
  

• 增加容忍失败的 Checkpoint 次数：通过设置 execution.checkpointing.tolerable-failed-checkpoints 参数，避免因 Checkpoint 失败导致的性能下降。

  execution.checkpointing.tolerable-failed-checkpoints: 100
  

2. Binlog 数据过滤开销

Flink CDC 在读取 MySQL Binlog 时，会拉取整个实例级别的变更日志，即使只关注部分表的数据。这种全量拉取后过滤的方式可能导致带宽和处理资源的浪费。

解决方案：

• 启用 Source 复用：通过复用同一个 MySQL CDC Source，避免多个作业重复拉取 Binlog 数据，从而减少带宽消耗。
• 优化正则表达式匹配：确保 table-name 参数的正则表达式尽可能精确，避免不必要的表被扫描。例如：

  'table-name' = '(t_process_wi_history_\d{1}|t_process_wi_history_\d{2})'
  

3. 全量阶段并发度不足

如果 Flink CDC 正处于全量数据读取阶段，并发度不足可能导致读取速度变慢。特别是在数据量较大的情况下，单线程读取可能会成为瓶颈。

解决方案：

• 增加 Source 并发度：通过提高 MySQL CDC Source 的并发度，加快全量数据的读取速度。例如：

  'scan.incremental.snapshot.chunk.size' = '10000',
  'parallelism' = '4'
  

• 分片优化：确保全量数据读取时的分片大小合理，避免单个分片数据量过大导致 OOM 问题。

4. 网络或数据库性能瓶颈

Flink CDC 的读取速度还可能受到网络 I/O 或数据库性能的限制。如果 current Fetch Event Time Lag 和 current Emit Event Time Lag 指标较大，说明可能存在网络延迟或数据库压力过大的问题。

解决方案：

• 检查网络状况：确保 Flink 集群与 MySQL 数据库之间的网络连接稳定，避免高延迟或丢包。
• 监控数据库性能：查看 MySQL 数据库的 CPU、内存和磁盘 I/O 使用情况，确保数据库有足够的资源支持 Binlog 读取。
• 减少数据库压力：如果存在多个 CDC 作业，可以考虑将表同步到 Kafka 消息队列中，再通过消费 Kafka 数据进行解耦。

5. Postgres CDC 特殊注意事项

如果是 Postgres CDC，还需要注意 Replication Slot 的管理。如果未及时清理无用的 Replication Slot，可能会导致磁盘空间浪费，进而影响性能。

解决方案：

• 手动释放 Replication Slot：

  SELECT pg_drop_replication_slot('rep_slot');
  

• 自动清理 Replication Slot：在作业配置中添加以下参数：

  'debezium.slot.drop.on.stop' = 'true'
  

6. 其他潜在原因

• 数据倾斜：如果某些分片的数据量远大于其他分片，可能导致整体读取速度变慢。
• 状态后端性能：如果使用的是旧版本的状态后端，可能会导致状态访问性能较低。建议升级到 VVR 8.0.1 及以上版本，利用其优化后的状态存储机制。

总结与建议

根据上述分析，您可以从以下几个方面排查和优化 Flink CDC 的读取速度： 1. 调整 Checkpoint 配置，缩短间隔时间并增加容忍失败次数。 2. 启用 Source 复用，优化正则表达式匹配。 3. 提高全量阶段的并发度，优化分片大小。 4. 检查网络和数据库性能，减少数据库压力。 5. 如果是 Postgres CDC，及时管理 Replication Slot。

如果问题仍未解决，建议结合具体的监控指标（如 numRecordsIn、numRecordsOut、current Fetch Event Time Lag 等）进一步定位瓶颈点。