【流式数据处理】Debezium 与 Change Data Capture

lakehouse 第 19 章 从入湖侧讲清楚了 Flink/Iceberg sink 怎么提交、CDC 怎么借 equality delete 做 upsert。但 CDC 管道在到达 Flink 或 Iceberg 之前，还有一整段「数据库 → Kafka」的传输层：谁读 binlog/WAL、事件长什么样、快照和增量怎么切换、断点存在哪。这一段通常由 Debezium + Kafka Connect 承担。

读完 lakehouse/19，读者仍可能卡在这些问题：

• 为什么 Kafka topic 里同时出现 op=r 和 op=c，下游该怎么区分快照与真实写入？
• before/after 在 UPDATE/DELETE 里各代表什么，缺字段时是不是 bug？
• connector 重启后从哪恢复——是 Kafka consumer offset，还是 Debezium 自己的 binlog 位点？
• 表加了列之后，旧事件和新事件的 schema 怎么对齐？

本文只讲 CDC 采集与事件模型，不重复 lakehouse/19 的表格式提交协议，也不展开 Flink checkpoint 内部（见本系列第 10、15 篇）。目标是把 Debezium 产出的事件读懂、把 Connect 架构里的断点与 schema 主题管清楚，并为入湖 upsert 准备好主键、顺序、幂等前提。

环境说明：本机为 Arch Linux on WSL2（内核 6.6.87.2）、i9-12900K / 32 GB，未安装 JVM、Kafka Connect、MySQL、Debezium。下文 JSON 示例来自 Debezium MySQL Connector 官方文档（标注「经删减」）；配置片段按官方文档可复现，不粘贴未执行环境的命令输出，也不伪造吞吐/延迟数字。

版本锚定：Debezium 3.x / 2.x 稳定版文档；Kafka Connect 随 Kafka 3.x（KRaft）；MySQL connector 以 binlog + GTID 场景为主叙述，PostgreSQL 差异在文中标注。

一、CDC 管道在整体链路中的位置

典型实时入湖链路可以切成四段：

flowchart LR
  DB[(OLTP<br/>MySQL / PG)] -->|binlog / WAL| DBZ[Debezium<br/>Kafka Connect Source]
  DBZ -->|change events| K[Kafka topics]
  K --> ENG[Flink / Spark /<br/>Connect Sink]
  ENG --> LAKE[("Iceberg / Hudi / Delta")]

Debezium 的角色是 log-based CDC：持续读取数据库事务日志（MySQL binlog、PostgreSQL logical decoding WAL 等），把每条行级变更封装成统一信封，写入 Kafka。它不保证下游 exactly-once——那是 Kafka + Flink + 表格式三层叠加的结果（本系列第 14、15 篇）。Debezium 保证的是：在 connector 正常运行且 offset 未丢失的前提下，按日志顺序 emit 变更，并在 envelope 里携带足够的源库位点与行镜像供下游还原语义。

二、变更事件信封（Envelope）

每条 Debezium 变更事件（change event）在 Kafka 里通常是一条 JSON / Avro 记录。核心结构固定为四块：

外层还有 ts_ms（Connect 处理时间戳）、可选 transaction（事务边界元数据，需显式开启）等。Debezium 文档称这一结构为 envelope——下游无论用 Flink SQL、Spark 还是 Iceberg Connect sink，首先都要按 envelope 解析语义，而不是直接把 Kafka 字节当 Parquet 列。

2.1 INSERT（op: c）

创建行：before 为 null，after 为完整新行。source.snapshot 通常为 "false"（来自 streaming）。

官方文档风格的示例（经删减）：

{
  "before": null,
  "after": {
    "id": 1002,
    "email": "alice@example.com",
    "name": "Alice"
  },
  "source": {
    "version": "2.5.0.Final",
    "connector": "mysql",
    "name": "inventory",
    "ts_ms": 1705305600000,
    "snapshot": "false",
    "db": "inventory",
    "table": "customers",
    "server_id": 223344,
    "file": "mysql-bin.000003",
    "pos": 154,
    "row": 0,
    "thread": 7
  },
  "op": "c",
  "ts_ms": 1705305601050
}

入湖侧把 after 当作待 upsert 的新版本行；主键字段必须出现在 after 里（见第九节）。

2.2 UPDATE（op: u）

更新行：before 与 after 同时非空（除非 column 过滤或 REPLICA IDENTITY 限制导致 PG 旧值不全）。Debezium 用这一对镜像表达「改了什么」——下游 equality delete + insert 入湖时，通常用 after 写新行、用主键（有时加上 before 里变化前的键）定位旧行。

{
  "before": {
    "id": 1002,
    "email": "alice@example.com",
    "name": "Alice"
  },
  "after": {
    "id": 1002,
    "email": "alice@example.com",
    "name": "Alice Smith"
  },
  "source": {
    "connector": "mysql",
    "name": "inventory",
    "ts_ms": 1705305700000,
    "snapshot": "false",
    "db": "inventory",
    "table": "customers",
    "file": "mysql-bin.000003",
    "pos": 892
  },
  "op": "u",
  "ts_ms": 1705305700120
}

若下游只关心当前状态而非变更 diff，常用 ExtractNewRecordState SMT 把 envelope 压平为「仅 after 列 + __op 头」——但入湖 upsert 若要做 delete，仍需保留 op 或 tombstone 语义（lakehouse/19 第四节）。

2.3 DELETE（op: d）

删除行：after 为 null，before 携带被删行的最后镜像。Iceberg equality delete 入湖时，关键是从 before（或 Kafka key）提取主键，写 delete 文件。

{
  "before": {
    "id": 1002,
    "email": "alice@example.com",
    "name": "Alice Smith"
  },
  "after": null,
  "source": {
    "connector": "mysql",
    "name": "inventory",
    "snapshot": "false",
    "db": "inventory",
    "table": "customers",
    "file": "mysql-bin.000003",
    "pos": 1204
  },
  "op": "d",
  "ts_ms": 1705305800300
}

2.4 快照读（op: r）

初始快照（blocking snapshot）与增量快照（incremental snapshot）阶段，Debezium 把表扫描结果 emit 为 READ 操作，即 op: r。before 为 null，after 为行内容；source.snapshot 为 "true"、"last"、"incremental" 等，标记事件来自快照而非 binlog streaming。

增量快照事件示例（Debezium 文档 Example 1，经删减）：

{
  "before": null,
  "after": {
    "pk": "1",
    "value": "New data"
  },
  "source": {
    "snapshot": "incremental"
  },
  "op": "r",
  "ts_ms": "1620393591654"
}

常见误区：把 r 当成「普通 insert」再写一遍，会导致入湖重复行。处理方式有三类：

1. 下游按 source.snapshot 过滤，快照阶段与 streaming 阶段用不同 sink 策略；
2. 用 upsert 按主键覆盖（幂等，lakehouse/19）；
3. 配置 ReadToInsertEvent SMT 把 r 改写成 c（仅当业务明确需要统一 op 码）。

MySQL connector 文档还说明：默认快照 emit r；若希望快照行表现为 c，可启用 ReadToInsertEvent SMT。

2.5 截断与其它 op 码

MySQL TRUNCATE 等 DDL 可能产生 op: t（truncate）类事件（视 connector 与表类型而定）。下游若只做行级 upsert，必须显式处理 truncate——否则湖表保留被截断表的历史行。PostgreSQL connector 对 TRUNCATE 的支持与配置相关；生产上常在下游把 truncate 映射为分区 overwrite 或人工运维，而不是假设 Debezium 自动等价于 DELETE 全表。

三、before / after 的工程语义

3.1 为什么需要成对镜像

关系型数据库日志里，一条 UPDATE 可能只记录「改了哪些列」。Debezium 在 connector 内维护表 schema 内存模型（见第五节 schema history），把日志条目 decode 成完整行镜像。对下游而言：

3.2 PostgreSQL 的 REPLICA IDENTITY

PostgreSQL logical replication 下，若表为 REPLICA IDENTITY DEFAULT，UPDATE/DELETE 的 before 可能只有主键列，非主键旧值缺失。需要完整 before 镜像时，应把表设为 REPLICA IDENTITY FULL（代价是 WAL 体积增大）。这是 CDC 入湖前必须在源库确认的 DBA 项，不是 Debezium 单方能补的。

3.3 列过滤与 schema 演进

Debezium 支持 column.include.list / column.exclude.list。过滤后 before/after 只含子集列，但 主键列必须保留，否则无法 upsert。schema 加列后，新事件 after 多字段、旧事件少字段——Avro/Schema Registry 场景靠 schema id 演进；JSON 场景靠下游「缺列填 null」策略。

四、source 元数据：位点、快照标记与顺序

source 块是 Debezium 的断点凭证，也是下游判断事件先后顺序的辅助依据（最终顺序仍由 Kafka 分区内 offset 保证）。

4.1 常用字段

关键区分：source.ts_ms 是数据库里变更发生的时间；记录顶层的 ts_ms 是 Kafka Connect 处理该事件的时间。做事件时间窗口（本系列第 2 篇）时，应优先 source.ts_ms，并处理 clock skew。

4.2 Kafka 分区内顺序 vs 跨表顺序

Debezium 默认按 表 发 topic（topic.prefix + . + schema + . + table）。单表变更在单分区内有序（若 producer 按主键 hash 到固定分区）；跨表无全局顺序。多表 JOIN 物化或宽表入湖时，要么在 Flink 里按业务键重分区，要么接受最终一致。

五、Snapshot 阶段 vs Streaming 阶段

Debezium connector 生命周期分两阶段：

stateDiagram-v2
  [*] --> Snapshot: 首次启动或无有效 offset
  Snapshot --> Streaming: 快照完成，切入 binlog/WAL
  Streaming --> Streaming: 持续 emit c/u/d
  Streaming --> Snapshot: snapshot.mode=always 等
  note right of Snapshot: emit op=r（或 SMT 改为 c）
  note right of Streaming: source.snapshot=false

5.1 为什么要 snapshot

MySQL binlog、PostgreSQL WAL 都有保留策略——日志会被 purge。Connector 不能假设「从很久以前的位置还能读」。首次启动时，Debezium 通常先做 consistent snapshot（一致快照）：在某一时间点冻结表视图（MySQL 用全局读锁或 table-level lock 变体），导出表数据，再从快照对应的 binlog 位点开始 streaming，避免快照与日志之间的缝。

PostgreSQL 使用 replication slot，原理类似：快照边界与 slot 起始 LSN 对齐。

5.2 Blocking snapshot 工作流（概念）

MySQL connector 文档描述的标准流程（简化）：

1. 若配置了 snapshot，锁定/协调源表（视 snapshot.locking.mode）；
2. 读取当前 binlog 位置，作为 streaming 起点；
3. 扫描 table.include.list 中的表，emit op=r 事件；
4. 释放锁，切换到 binlog streaming；
5. 对快照开始后新建表的 DDL，在 streaming 阶段补 capture。

快照进行中 connector 失败重启，会从快照进度恢复而非重头读 binlog（具体 checkpoint 由 Debezium 内部 offset 记录）。

5.3 Streaming 阶段

快照完成后，connector 只读事务日志：

• INSERT → op=c
• UPDATE → op=u
• DELETE → op=d

此阶段 source.snapshot 为 false。Offset 持续写入 Connect offset storage（第六节）。若 binlog 被 purge 且 offset 指向已不存在的位置，connector 失败；snapshot.mode=when_needed 可触发重新快照（MySQL 文档 Behavior when things go wrong）。

5.4 Incremental snapshot（增量快照）

大表全量 blocking snapshot 可能耗时数小时。Debezium 支持 incremental snapshot：在不长时间锁表的前提下分 chunk 扫描，并与 streaming 并行。增量快照仍 emit op=r，source.snapshot=incremental。文档指出：streaming 与 snapshot 可能乱序到达，connector 内部有 buffer 做碰撞解析，之后才对 Kafka emit——下游仍可能短暂看到「先 update 后 read」的错觉，upsert 按主键可吸收。

触发方式： signaling table、execute-snapshot Kafka 信号 topic 等（Debezium Incremental snapshots 章节）。

六、snapshot.mode 选型

snapshot.mode 决定 connector 启动时是否以及如何跑快照。MySQL connector 官方文档（stable）主要取值如下：

与 offset 的交互（工程上最易踩坑）：

• 误删 connect-offsets 中该 connector 的 key → 视为无 offset → initial 会重跑全量快照，Kafka 与湖侧出现大量重复 r/重复主键；
• never + 无 offset → 从当前 binlog 开始，历史变更永久丢失；
• binlog rotate/purge 后旧 offset 失效 → 需 when_needed 或人工干预重新快照。

PostgreSQL connector 有类似概念，但依赖 replication slot；删 slot 的后果比删 Kafka offset 更严重——可能要从头快照。

七、Kafka Connect 架构

Debezium 以 Kafka Connect source connector 插件形式运行，不单独起进程（除非用 Debezium Engine 嵌入应用，本文不展开）。

flowchart TB
  subgraph Connect Cluster
    W[Connect Worker]
    C[Debezium MySQLConnector]
    T1[Task 0]
    T2[Task 1]
    W --> C
    C --> T1
    C --> T2
  end
  DB[(MySQL)] --> T1
  DB --> T2
  T1 --> K1[Kafka topic<br/>inventory.db.orders]
  T2 --> K2[Kafka topic<br/>inventory.db.customers]
  T1 --> OS[(offset.storage.topic)]
  T1 --> SH[(schema.history.internal.kafka.topic)]

7.1 Worker、Connector、Task

MySQL connector 通常 1 个 database server → 1 个 task（单线程读 binlog）。PostgreSQL 一般也是单 task。并行度来自多 connector 实例捕获不同库表，而非单 connector 多 task 读同一 binlog。

7.2 配置与 REST 管理

Connect 通过 REST API 注册 connector：

{
  "name": "inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",
    "topic.prefix": "inventory",
    "database.include.list": "inventory",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schema-changes.inventory",
    "include.schema.changes": "true"
  }
}

tasks.max 对 MySQL 实际常为 1；调大不会加速单库 binlog 读取。

7.3 内部 topic 与 Connect 框架 topic

除业务 change event topic 外，还有三类基础设施 topic：

Debezium 另用 schema history topic（connector 级配置，非 Connect 框架内置）：

生产必须给 connect-offsets 与 schema history 设 replication.factor ≥ 3，否则 broker 故障可能丢断点，触发全量重快照。

八、Offset storage：读库位点存在哪

Debezium 的 offset 不是 Kafka consumer group offset。它记录的是源数据库日志坐标，例如：

• MySQL：{"file": "mysql-bin.000003", "pos": 154} 或 GTID 集合；
• PostgreSQL：{"lsn": 24023128, ...}。

Connect 把这些 JSON 以 key = [connector 名, partition 标识] 写入 offset.storage.topic（默认 connect-offsets），value 为位点 + 可选 schema history 进度。

sequenceDiagram
  participant DB as MySQL binlog
  participant T as Debezium Task
  participant OS as connect-offsets
  participant K as data topic
  DB->>T: event at pos 892
  T->>K: emit change event
  T->>OS: commit offset pos 892
  Note over T,OS: task 崩溃重启后<br/>从 OS 读 pos 892 续读

与下游 Flink 的分工：

• Debezium offset：保证 Kafka 里不丢、不重读整库（在 at-least-once emit 语义下仍可能重复 emit 最后一批）；
• Flink checkpoint 中的 Kafka source offset：保证 Flink 算子 exactly-once 或 at-least-once；
• Iceberg commit：保证 湖表可见性边界。

三层独立。清空 Flink checkpoint 不会回退 Debezium；重置 Debezium offset 不会自动重置 Flink。

8.1 重置 offset 的后果

社区 FAQ 常见操作：向 connect-offsets 写 tombstone（key 相同、value null）触发 compaction 删除位点。后果：

• 下次启动按 snapshot.mode 决定是否全量快照；
• 下游若仅 at-least-once，会出现重复历史数据；
• 入湖 upsert 可幂等吸收，append-only 湖表则重复行。

任何「从某个 binlog pos 重放」的运维都必须同时规划 湖侧幂等 与 下游 consumer。

九、Schema history topic：DDL 链为何不能 compact

数据库 schema 随时间变化。同一张表在 t0 可能是 (id, name)，t1 加了 email 列。Binlog 里 t0 时刻的 UPDATE 只有两列，t1 时刻有三列——connector 必须知道每个位点对应的 schema 才能 decode。

Debezium 做法：

1. 在内存维护每张表的当前 schema；
2. 遇到 DDL，解析并更新内存，同时 append 一条记录到 schema.history.internal.kafka.topic；
3. Restart 时从 offset 位点出发，重放 schema history 中截至该位点的全部 DDL，重建内存 schema，再读 binlog。

因此 schema history topic 必须 保留完整 append-only 历史（文档明确：不要 compact）。若 topic 丢失或损坏，需 snapshot.mode=recovery 等模式重建（MySQL 文档 Restoring a database schema history topic）。

9.1 Schema change 事件 topic

若 include.schema.changes=true，Debezium 还会向 {topic.prefix} 或配置的 topic 发送 schema change 事件（含 DDL 语句与 tableChanges 结构），供下游 Registry 或 Flink 动态建表。这与 history topic 不同：history 给 connector 自己恢复用；schema change topic 给应用消费。

Flink CDC 或 KafkaTableSource 监听 schema change 来 ALTER TABLE 湖表——否则新列进不了 Iceberg equality field。

十、Kafka 消息键、topic 命名与 tombstone

10.1 Topic 命名

默认：{topic.prefix}.{database}.{table}，例如 inventory.inventory.customers。可通过 topic.naming.strategy 定制。

10.2 消息 Key

默认 key 为 主键 JSON。同一主键的变更进入同一 Kafka 分区（若 key-based partitioner），保证分区内 per-key 顺序——这对 upsert 入湖至关重要：同一 uid 的 u/d 若乱序，湖表可能短暂出现旧版本覆盖新版本。

无主键表：key 可能为 null 或整行 hash，顺序语义弱，入湖 upsert 需额外设计（ surrogate key 或全行 equality）。

10.3 Delete 与 compacted topic

若对 change topic 启用 log compaction，DELETE 事件可能以 tombstone（value null）保留 key。下游 Flink 必须启用 deletion 语义解析。未 compact 的 topic 则保留完整 op=d envelope。

十一、Single Message Transform（SMT）与下游简化

Connect 链上常加 SMT 再入 Kafka：

入湖管道若用 ExtractNewRecordState，Iceberg sink 仍须从 header 读 __op 区分 delete。Flink SQL 解析 Debezium JSON 时可用 DebeziumJsonDeserializationSchema 或官方 CDC connector 直接产出 RowKind（+I/+U/-D 等）。

十二、与 lakehouse 第 19 章 upsert 的衔接

lakehouse/19 第四节说明：CDC 入湖要把 c/u/d/r 按主键 reduce 成当前状态，Iceberg 用 equality delete + 新行，Hudi 用 record index，Delta 用 MERGE。Debezium 层必须为这三条机制提供输入前提：

12.1 主键全局唯一

Debezium 只转发源库主键。分库分表若各库都有 id=1，Kafka key 冲突，湖表 upsert 互相覆盖。引擎侧需加 shard id 列或合成全局键（本系列第 17 篇写 Flink 侧 keyBy 与 bucket）。

12.2 顺序与幂等

详细表格式提交与 equality delete 机制见 lakehouse/19 第四节 与 第 10 篇行级删除。

12.3 两条入湖路径对比

flowchart LR
  subgraph PathA["路径 A: Kafka 中转"]
    DB1[(MySQL)] --> DBZ[Debezium]
    DBZ --> K[Kafka]
    K --> FL1[Flink SQL]
    FL1 --> ICE1[Iceberg upsert]
  end
  subgraph PathB["路径 B: Flink CDC 直连"]
    DB2[(MySQL)] --> FL2[Flink CDC source]
    FL2 --> ICE2[Iceberg upsert]
  end

• 路径 A：Debezium 管 snapshot/offset/schema history；Flink 只消费 Kafka，运维边界清晰，多下游可复用 topic。
• 路径 B：无 Kafka 中间层，Flink CDC 内置 snapshot（语义类似 Debezium）；少一跳，但 snapshot 与 Flink checkpoint 耦合，运维工具链不同。

lakehouse/19 的 Flink SQL 示例走路径 B；本文架构图走路径 A。Upsert 表 DDL 与 write.upsert.enabled 对两条路径相同。

十三、生产故障模式（CDC 侧）

这些问题的 入湖侧 后果（小文件、commit 冲突）在本系列第 17 篇从 Flink 作业角度继续展开。

十四、边界

本文不展开：

• Flink checkpoint、两阶段提交 sink 内部（第 10、15 篇）；
• Iceberg IcebergFilesCommitter 与 catalog CAS（lakehouse/11、19）；
• Debezium Engine 嵌入式 API、Outbox Pattern 实现细节；
• Oracle/SQL Server connector 特有权限与 supplement logging 全文。

本文覆盖：

• Debezium envelope 与 op/before/after/source；
• snapshot / streaming / incremental snapshot 阶段划分；
• Kafka Connect worker、offset topic、schema history topic 分工；
• 与 lakehouse/19 upsert 的输入前提与路径选择。

参考资料

1. Debezium Documentation, MySQL Connector（Event structure、Snapshot、snapshot.mode、schema.history.internal.kafka.topic、incremental snapshot）。A 级。
2. Debezium Documentation, PostgreSQL Connector（logical decoding、REPLICA IDENTITY、sequence）。A 级。
3. Apache Kafka Documentation, Kafka Connect（worker、offset/config/status storage、REST API）。A 级。
4. MySQL Reference Manual, Binary Log；PostgreSQL Documentation, Logical Replication。A 级。
5. 本系列：第 15 篇 两阶段提交与 EOS、第 17 篇 流式入湖深化。
6. 跨系列：lakehouse 第 19 章 流式写入与 CDC 入湖、第 10 章 行级删除。

返回 系列目录 · 上一篇 两阶段提交与端到端 Exactly-Once · 下一篇 流式入湖深化