如何分批读取CDC表数据且避免重复读取？

核心思路是通过持久化的进度追踪机制，记录每次读取的终点位置，确保每次分批读取都从上次的结束点开始，同时结合幂等性保障避免重复写入。以下是几种落地方案：

方案1：利用成熟CDC工具的内置偏移量管理 #

如果使用Debezium、Canal这类CDC工具，它们自带偏移量（Offset）持久化能力，天然支持断点续传：

• 配置批量处理参数：比如Debezium设置batch.max.rows=1000控制每批处理的事件数量
• 进度自动追踪：工具会将消费到的CDC日志位点（如binlog文件名+位置）持久化到存储（Kafka偏移量主题、本地文件或自定义数据库表）
• 关键规则：必须在成功将批次数据写入新表后，再提交偏移量，避免处理失败导致重复消费
• 示例Debezium配置片段：

  connector.class=io.debezium.connector.mysql.MySqlConnector
  offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore
  offset.storage.file.filename=/data/cdc_offsets.dat
  batch.max.rows=1000
  snapshot.mode=initial  # 先全量迁移历史数据，再增量消费CDC事件
  

方案2：自定义基于时间戳/递增序列号的分批读取 #

如果CDC表自带update_time（更新时间戳）或cdc_seq（全局递增序列号）这类标识字段，可自行实现进度追踪：

1. 创建进度追踪表：单独存储每张CDC表的读取进度

   CREATE TABLE cdc_migration_progress (
     table_name VARCHAR(64) PRIMARY KEY,
     last_max_seq BIGINT NOT NULL DEFAULT 0,
     last_update_time TIMESTAMP NOT NULL DEFAULT '1970-01-01 00:00:00'
   );
   -- 初始化两张表的进度
   INSERT INTO cdc_migration_progress (table_name) VALUES ('cdc_table_a'), ('cdc_table_b');
   

2. 分批读取与写入：
   • 从进度表获取上次读取的终点值
   • 按顺序读取批次数据，写入新表
   • 事务包裹读取、写入、更新进度的操作，确保原子性

   -- 读取cdc_table_a的下一批数据
   SELECT * FROM cdc_table_a 
   WHERE cdc_seq > (SELECT last_max_seq FROM cdc_migration_progress WHERE table_name='cdc_table_a')
   ORDER BY cdc_seq LIMIT 1000;
   
   -- 成功写入新表后，更新进度（假设当前批次最大seq为2000）
   UPDATE cdc_migration_progress 
   SET last_max_seq=2000 
   WHERE table_name='cdc_table_a';
   

3. 注意事项：优先用全局递增的cdc_seq而非时间戳，避免同一时间点多条数据更新导致漏读；写入新表时添加业务主键唯一约束，保证幂等性。

方案3：基于数据库日志位点的直接追踪 #

如果直接读取数据库底层CDC日志（如MySQL Binlog、PostgreSQL WAL）：

• 记录每次读取的日志位点：比如MySQL的binlog_file和binlog_pos，PostgreSQL的WAL LSN
• 每次启动时从记录的位点开始解析日志，提取变更事件分批写入新表
• 每处理完一批，就更新持久化的位点记录（可存在数据库或配置文件中）
• 要求：数据库需开启Binlog（MySQL设为ROW格式）或WAL（PostgreSQL开启逻辑复制），可借助mysqlbinlog等工具解析日志，或通过代码调用数据库日志API实现。

通用关键规则 #

• 幂等性保障：新表添加业务主键唯一约束，或写入前校验数据是否已存在，避免重复读取导致的重复写入
• 故障恢复：进度追踪存储必须持久化，进程重启后能直接恢复到上次的读取位置
• 性能调优：根据数据库负载调整批次大小（建议500-2000条），避免过大导致内存溢出或锁表
• 关联表一致性：若两张CDC表存在业务关联，需保证关联数据的迁移顺序（比如先迁移主表，再迁移从表）

内容的提问来源于stack exchange，提问作者ZedZip