CDC 与增量同步 · 数据开发学习站

本章目标：能说清 CDC 的原理，设计一条“全量初始化 + 增量同步 + 幂等落库”的链路，并处理删除、重复和 schema 变更。前置知识：数据采集与同步、建表与增删改预计用时：110 分钟

每天全量抽一遍业务库，数据量一大就抽不完；只按 update_time 拉增量，又会漏掉物理删除。CDC（Change Data Capture，变化数据捕获）解决的就是“业务库变了，数仓怎么知道、怎么可靠地搬过来”。

1. 为什么需要 CDC

批量同步有两种常见做法：

做法	怎么取数	问题
全量同步	每次 `SELECT *` 整表	数据量大时慢、给业务库压力大
时间戳增量	`WHERE update_time > 上次时间`	漏物理删除；同一秒的边界数据可能重复或丢失

CDC 直接读数据库的变更日志（MySQL 的 binlog、Oracle 的 redo、PostgreSQL 的 WAL），能拿到 INSERT、UPDATE、DELETE 三种事件，并且带上变更顺序。它适合订单、用户、商品这类需要持续同步的核心表。

2. 核心概念

2.1 两类 CDC

基于日志：解析 binlog/WAL，实时性好、对业务库压力小，能捕获删除。需要数据库开启日志并授予权限。
基于查询：轮询表或时间戳字段，实现简单，但延迟高、难捕获删除，边界容易出错。

生产上常见的组合是：先做一次全量快照，再从快照位点开始追增量日志。全量和增量必须对齐到同一个位点，否则中间会丢数据或重复。

2.2 binlog 与位点

MySQL 的 binlog 有三种格式：

STATEMENT：记录 SQL 语句。函数、时间等可能导致主从结果不一致。
ROW：记录每一行变更前后的值。CDC 通常要求 ROW。
MIXED：由数据库自动选择。

每个事件有位置信息。传统复制用文件名 + 偏移量（mysql-bin.000001, 154），GTID 模式下用全局事务 ID。同步任务要持久化“已经处理到哪个位点”，重启后从位点继续，而不是从头再来。

2.3 幂等与去重

CDC 链路至少一次投递很常见：网络重试、任务重启都可能让同一事件到达两次。落库必须幂等：

有主键的表用 UPSERT / MERGE，同一主键重复写入结果一致。
分区表按“业务日期 + 批次”覆盖写，重跑同一批次先清理再写。
事件表用 event_id 去重，保留每个主键最新一条。

2.4 删除怎么处理

不要物理删除数仓里的历史行。更稳妥的做法是加一个 is_deleted 标记，或者把删除事件写进一张变更日志表，再由下游按需过滤。这样既能追溯，也不会因为一次误删把历史数据清空。

3. 动手：用 DuckDB 模拟一次增量合并

先建一张 CDC 事件表，模拟业务库发过来的变更：

CREATE OR REPLACE TABLE ods_order_cdc (
    event_id INTEGER,
    order_id VARCHAR,
    op VARCHAR,
    amount DECIMAL(12, 2),
    event_ts TIMESTAMP
);

INSERT INTO ods_order_cdc VALUES
    (1, '1001', 'I', 120.00, TIMESTAMP '2024-01-01 10:00:00'),
    (2, '1001', 'U', 150.00, TIMESTAMP '2024-01-01 10:05:00'),
    (3, '1002', 'I', 80.00,  TIMESTAMP '2024-01-01 10:06:00'),
    (4, '1002', 'D', NULL,   TIMESTAMP '2024-01-01 10:10:00'),
    (5, '1003', 'I', 260.00, TIMESTAMP '2024-01-02 09:00:00'),
    (6, '1001', 'U', 150.00, TIMESTAMP '2024-01-02 09:05:00');

3.1 取每个订单的最新状态

CREATE OR REPLACE TABLE order_current AS
WITH ranked AS (
    SELECT
        order_id,
        op,
        amount,
        event_ts,
        ROW_NUMBER() OVER (
            PARTITION BY order_id
            ORDER BY event_ts DESC, event_id DESC
        ) AS rn
    FROM ods_order_cdc
)
SELECT order_id, amount, event_ts
FROM ranked
WHERE rn = 1 AND op <> 'D';

SELECT * FROM order_current ORDER BY order_id;

结果只剩订单 1001 和 1003。订单 1002 的最新事件是 D，被过滤掉。ORDER BY event_ts DESC, event_id DESC 里的 event_id 是必要的：同一秒内可能有多条事件，单靠时间戳无法确定顺序。

3.2 按分区覆盖，保证重跑幂等

离线数仓更常见的做法是按天分区覆盖。假设 2024-01-02 的增量要写入 dwd_order_inc：

CREATE OR REPLACE TABLE dwd_order_inc (
    dt DATE,
    order_id VARCHAR,
    amount DECIMAL(12, 2)
);

BEGIN TRANSACTION;
DELETE FROM dwd_order_inc WHERE dt = DATE '2024-01-02';

INSERT INTO dwd_order_inc
SELECT
    DATE '2024-01-02' AS dt,
    order_id,
    amount
FROM ods_order_cdc
WHERE event_ts >= TIMESTAMP '2024-01-02 00:00:00'
  AND event_ts <  TIMESTAMP '2024-01-03 00:00:00'
  AND op <> 'D'
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY order_id ORDER BY event_ts DESC, event_id DESC
) = 1;
COMMIT;

重跑同一天时，先删掉该分区再写，结果不会翻倍。注意 DELETE 和 INSERT 要放在同一个事务里，避免删完还没写入时任务失败，留下空分区。

3.3 批量同步工具配置

DataX 的 MySQL 到 Hive 作业，核心是 reader、writer 和切分键：

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "username": "reader",
            "password": "${MYSQL_PASSWORD}",
            "column": ["order_id", "status", "amount", "update_time"],
            "where": "update_time >= '${bizdate}'",
            "connection": [
              {
                "table": ["orders"],
                "jdbcUrl": ["jdbc:mysql://mysql-host:3306/shop"]
              }
            ]
          }
        },
        "writer": {
          "name": "hdfswriter",
          "parameter": {
            "defaultFS": "hdfs://namenode:8020",
            "fileType": "orc",
            "path": "/warehouse/ods/orders/dt=${bizdate}",
            "fileName": "orders",
            "column": ["order_id", "status", "amount", "update_time"],
            "writeMode": "append"
          }
        }
      }
    ]
  }
}

密码用变量传入，不要写进 JSON 提交到 Git。bizdate 也要由调度系统传入，不能写死。

3.4 Flink CDC 的最小链路

Flink CDC 可以直接把 MySQL 的 binlog 读成流，再写到 Kafka 或数据湖：

CREATE TABLE mysql_orders (
    order_id STRING,
    status STRING,
    amount DECIMAL(12, 2),
    update_time TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'mysql-host',
    'port' = '3306',
    'username' = 'reader',
    'password' = '${MYSQL_PASSWORD}',
    'database-name' = 'shop',
    'table-name' = 'orders',
    'server-time-zone' = 'Asia/Shanghai'
);

CREATE TABLE kafka_orders (
    order_id STRING,
    status STRING,
    amount DECIMAL(12, 2),
    update_time TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'ods_orders',
    'properties.bootstrap.servers' = 'kafka:9092',
    'key.format' = 'json',
    'value.format' = 'json'
);

INSERT INTO kafka_orders
SELECT order_id, status, amount, update_time
FROM mysql_orders;

upsert-kafka 会按主键把更新和删除写成 upsert 消息，下游消费时不会把同一订单的多次变更当成多条新订单。

4. 全量 + 增量的衔接

一条可靠的初始化流程：

1. 记录当前 binlog 位点 P
2. 用一致性快照导出全量数据（带事务或锁，保证快照对应位点 P）
3. 全量写入 ODS，标记快照完成
4. 从位点 P 开始消费 binlog，写增量表
5. 下游按主键合并全量和增量

如果第 2 步和第 3 步之间业务库继续写入，快照和增量会重叠。重叠不可怕，幂等合并可以解决；可怕的是漏数据。所以位点要在快照开始前记录，不能在快照结束后才记录。

5. 常见坑

5.1 只按 update_time 拉增量

物理删除不会更新 update_time，永远拉不到。业务库如果用软删除（is_deleted），要确认删除操作会更新这个字段。

5.2 同一秒的数据重复或丢失

用 >= 拉增量但不记录上次最大时间，或者用 > 但事务提交顺序和时间戳不一致，都会出问题。CDC 用位点而不是时间戳，正是为了避开这个边界。

5.3 事件乱序

同一订单的更新事件可能因为分区、网络等原因乱序到达。合并时按 event_ts 和事件序号排序，取最新一条，而不是“最后到达的一条”。

5.4 schema 变更

源表加字段、改类型，CDC 任务可能直接失败或写出 null。上线前约定变更流程：先兼容新旧字段，再切换，最后清理旧字段。Flink CDC 新版本支持部分 schema evolution，但下游存储也要支持。

5.5 没有监控

至少监控三项：同步延迟（源库当前时间 - 已处理事件时间）、失败次数、数据量突增突降。没有监控的 CDC 任务，等业务发现报表不对时往往已经错了好几天。

6. 在线练习

题目	考点
196. 删除重复的电子邮箱	去重与保留最小主键
182. 查找重复的电子邮箱	重复数据识别
1173. 即时食物配送 I 🔒	状态与时间条件过滤

7. 面试延伸

面试题：全量同步和 CDC 增量同步怎么选？

是什么：全量简单但慢、压力大；CDC 实时、能捕获删除，但需要日志权限和位点管理。

解决什么问题：数据量增长后，全量窗口越来越长；CDC 把同步变成持续的小批量变更。

我的实操边界：本章用 DuckDB 模拟 CDC 合并，并用 Flink CDC 语法写过链路，未在生产 MySQL 上长期运行。

面试题：CDC 怎么保证不丢不重？

是什么：不丢靠持久化位点和至少一次投递；不重靠主键幂等、事件去重或分区覆盖。

解决什么问题：网络重试和任务重启必然带来重复，幂等比“绝不重复”更容易实现。

我的实操边界：本章演示了分区覆盖和最新事件合并，未实现两阶段提交。

面试题：物理删除怎么同步？

是什么：日志型 CDC 能捕获 DELETE；查询型同步要靠软删除字段或定期全量比对。

解决什么问题：数仓里保留历史，同时让下游知道这条记录已经失效。

我的实操边界：本章用 op = 'D' 过滤当前表，未实现墓碑表和保留策略。

8. 本章验收

能说出基于日志和基于查询的 CDC 各自优缺点。
能解释 binlog 位点、GTID 在增量同步中的作用。
能用窗口函数从事件表取每个订单的最新有效状态。
能写出“先删分区再写入”的幂等 SQL，并说明为什么要放在事务里。
能说出物理删除的两种处理方式及其代价。
能列出至少三个 CDC 同步监控指标。