全量+增量同步方案——一张10亿行的订单表怎么平滑迁移到数仓？

数据采集手册第 3 题约 10 分钟读完

本文目录

先说结论
业务场景：10亿行订单表迁移Delta Lake
方案一：传统离线——DataX全量+Canal增量
方案二：Flink CDC一体化——推荐做法
方案三：双写灰度——业务侵入但最稳
三方案对比
面试高频追问
实战踩坑：切换点选错导致重复数据
总结

先说结论

核心思路：先记下"切换点"（一个明确的binlog position或时间戳），然后并发跑全量同步，全量完成后从切换点开始消费增量，最后做数据校验。

三种主流方案：① 传统离线：DataX全量+Canal增量；② 一体化：Flink CDC全量增量自动衔接（推荐）；③ 双写灰度：业务代码同时写新旧，数据自然对齐。

一致性保证：全量必须基于一个明确snapshot（比如mysqldump --master-data=2拿到的position），增量从这个position严格往后追，避免重复或丢失。

先把这段讲顺；面试官追问时，再展开下面的原理和细节。

详细讲解

"全量+增量"是面试CDC时最爱考的实操题，因为它把架构、性能、一致性、容错全考到了。之前做过一次10亿行订单表的迁移，从0开始设计这套方案，踩了不少坑。这道题不要只回答"先跑全量再切增量"——面试官想听的是衔接窗口怎么处理、边界数据怎么对齐、怎么验证不丢数据。

业务场景：10亿行订单表迁移Delta Lake

那次迁移的表大概10亿行、约800GB、单行约800字节，每天新增300万行、更新约500万行。要从MySQL同步到Databricks Delta Lake，要求：① 不停机；② 不丢数据；③ 切换前后差异不超过0.001%。当时评估了三套方案，最终选了"DataX全量+Canal增量+Spark Merge"的组合。

时间线设计：
  T-1天 19:00  确定切换点：记录MySQL master当前position
  T-1天 19:00  Canal开始消费，把增量binlog落到Kafka(只存不消费)
  T-1天 19:30  DataX启动全量，按id范围切16个并发，预计8小时
  T0天  03:30  全量完成，spark写入Delta Lake的staging表
  T0天  03:35  启动Spark消费Kafka，从切换点position开始merge增量
  T0天  04:00  增量追平(Kafka消费lag=0)
  T0天  04:30  数据校验：count、sum、按天分布对比
  T0天  05:00  切流，下游报表指向Delta Lake

方案一：传统离线——DataX全量+Canal增量

这是最经典的方案，也是大多数传统数仓团队的做法。关键在于"切换点"的选取。

# Step 1: 在Canal启动前，记录当前MySQL master position（关键!）
mysql> FLUSH TABLES;  -- 不加锁，只是触发一次binlog flush
mysql> SHOW MASTER STATUS;
# 记下: mysql-bin.001234, position=587326441
 
# Step 2: Canal instance.properties设置起始位点为这个值
canal.instance.master.journal.name = mysql-bin.001234
canal.instance.master.position = 587326441
# Canal从这个点开始消费，所有增量都进Kafka
 
# Step 3: DataX跑全量(从从库读，不影响主库)
datax.py --jvm="-Xms4G -Xmx4G" mysql_to_delta.json
# 注意querySql不要带where create_time，因为我们要全表全量
 
# Step 4: 全量完成后，启动Spark消费Kafka做MERGE INTO
spark-submit --class CanalMergeJob canal-merge.jar

关键点：Canal的起始position必须在DataX全量开始之前确定并启动，否则DataX跑的过程中产生的更新会丢失。这是新手最容易踩的坑——"先跑全量再开Canal"，中间几小时的数据全没了。

方案二：Flink CDC一体化——推荐做法

Flink CDC 2.x的杀手级特性就是把上面这套流程内置了。它用Chunk-based无锁snapshot：把表按主键切chunk，每个chunk独立读取并对齐binlog位点，全量完成后无缝切到增量。开发量从几百行Spark+几个调度任务，简化到一段Flink SQL。

-- 一段SQL搞定全量+增量
CREATE TABLE orders_cdc (
  id BIGINT,
  user_id BIGINT,
  amount DECIMAL(10,2),
  status STRING,
  create_time TIMESTAMP(3),
  update_time TIMESTAMP(3),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'mysql-readonly-slave',  -- 走从库
  'database-name' = 'order_db',
  'table-name' = 'orders',
  'scan.startup.mode' = 'initial',  -- 全量+增量
  'scan.incremental.snapshot.chunk.size' = '50000',  -- 每chunk 5万行
  'scan.incremental.snapshot.chunk.key-column' = 'id',
  'server-id' = '5400-5410'  -- 多个server-id允许并发snapshot
);
 
-- 直接写入Iceberg/Delta
INSERT INTO iceberg_catalog.dwd.orders SELECT * FROM orders_cdc;

适用场景：有Flink基础设施；表数量多需要统一管理；要求最少代码量。

不适用：表特别大（10亿+）且没有合适的主键切片字段；运维不熟Flink。

方案三：双写灰度——业务侵入但最稳

EMR迁Databricks那波核心订单表用的是这个方案——业务代码同时写MySQL和新的Delta Lake，下游报表先双跑双比对，确认一致后切流。这个方案侵入业务但完全不依赖CDC，可控性最高。

// 业务代码侧的双写（用注解+AOP）
@Service
public class OrderService {
    @DualWrite(target = "delta_lake", async = true, fallback = SILENT)
    public void createOrder(Order order) {
        // 1. 主写：MySQL（事务）
        orderMapper.insert(order);
        // 2. AOP切面：异步写入Kafka -> Delta Lake
        // 失败只记日志不阻塞主链路（async + fallback=SILENT）
    }
}

适用场景：核心交易表，对一致性要求极高；有能力改业务代码；可以接受额外开发成本。

三方案对比

维度	DataX+Canal	Flink CDC	业务双写
开发量	高(多个组件)	低(一段SQL)	中(改业务代码)
一致性保证	需手动对齐position	自动衔接	天然一致
业务侵入	无	无	高
切换风险	中(衔接窗口)	低	低(双跑)
回滚难度	易	易	需下线代码
适用场景	存量改造	新建实时数仓	核心交易表

面试高频追问

追问一：怎么验证全量+增量的数据一致性？

做了三层校验：① 行数对比：MySQL count vs Delta Lake count（差异允许在增量未追平的窗口内）；② 抽样对比：随机抽1万行主键，逐字段对比；③ 业务指标对比：按天聚合GMV、订单数，差异不能超过0.001%。三层都过才允许切流。

追问二：全量过程中如果失败重启了怎么办？

DataX方案下，失败重启会从头跑全量，但已经写入的staging数据要先truncate（避免重复）。Flink CDC方案就优雅多了：它的Chunk算法每个chunk独立，失败的chunk会被Flink Checkpoint机制自动重试，不会从头来。这也是Flink CDC比DataX优秀的地方。

追问三：迁移期间业务方还在改源库结构怎么办？

这是真实业务里最头疼的问题。一般做法是：① 提前公告冻结期（迁移前一周不允许DDL）；② 如果实在要改，必须走变更评审，迁移团队要review；③ Flink CDC从2.4开始支持Schema Evolution，但只支持新增列，删除/改类型仍然要重启作业。

实战踩坑：切换点选错导致重复数据

第一次跑迁移时，手抖让Canal从"DataX全量结束之后"的position开始消费，结果上线后发现Delta Lake里有大量重复订单。原因：DataX全量跑了8小时，这8小时里有上百万次update，但Canal没消费到，导致一些订单的状态停留在了8小时前的版本。后来线上又来了新update，Spark merge的时候因为目标表里是"老版本"，触发了UPDATE，但其实数据状态已经不一致了。

-- 错误的Spark Merge逻辑（没有版本控制）
MERGE INTO target_orders t
USING kafka_orders s ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.status = s.status
-- 问题：如果kafka里来了一条8小时前的旧update，会覆盖掉更新的状态
 
-- 正确做法：用binlog的op_ts或source_ts做版本对比
MERGE INTO target_orders t
USING (
  SELECT id, status, op_ts,
         ROW_NUMBER() OVER (PARTITION BY id ORDER BY op_ts DESC) rn
  FROM kafka_orders
) s ON t.id = s.id AND s.rn = 1
WHEN MATCHED AND s.op_ts > t.op_ts THEN UPDATE SET t.status = s.status, t.op_ts = s.op_ts
WHEN NOT MATCHED THEN INSERT *;

解决方案：调整切换点为"DataX启动之前"，让Canal从全量开始之前就消费，全量结束后所有增量binlog都能追上。同时Merge逻辑加op_ts版本对比，保证幂等。从此再没出过类似问题。

总结

全量+增量的核心是"切换点"的选取——必须在全量启动之前固定position，让增量包含全量期间的所有变更。Flink CDC把这套流程自动化了，是新建项目首选。Merge逻辑必须基于版本号做幂等，否则乱序binlog会污染数据。面试时按"切换点→全量→增量追平→校验→切流"五步讲，逻辑严密分高。

下一篇聊一个让人头大的话题：业务库改了字段类型，下游CDC作业全挂了，Schema演进怎么处理？

你们公司做过最难的迁移是什么？踩过哪些坑？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长

刷题进度保存在当前浏览器