4. Debezium实战：MySQL CDC配置、Kafka Connect集成、SnapShot与Streaming模式、Schema变更处理

好，咱们进入实战环节。上一章讲了CDC的理论基础，这一章直接上手Debezium。说实话，Debezium是我在实时数仓项目里用得最多的CDC工具，没有之一。它跟Kafka Connect的集成做得非常成熟，配置起来也相对顺手。

这一章我会带着你一步步把MySQL的CDC跑起来。从最基础的配置，到快照和流式模式的切换，再到最让人头疼的Schema变更——嗯，这块坑不少，我踩过好几次。

4.1 环境准备：先搭好Kafka Connect集群

在配置Debezium之前，你得先有个Kafka Connect集群。我个人习惯用Confluent Platform，但Apache Kafka发行版也一样用。这里我假设你已经有了Kafka和ZooKeeper。

Kafka Connect有两种模式：

• Standalone模式：单进程，适合开发测试
• Distributed模式：多进程，生产环境必选

我建议你直接上Distributed模式。为什么？因为生产环境你不可能容忍单点故障。我曾经在一个项目里图省事用了Standalone，结果Connect进程挂了，CDC停了整整一个周末……从那以后我再也没用过Standalone。

配置一个基本的distributed worker，核心参数如下：

# connect-distributed.properties
bootstrap.servers=localhost:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
offset.storage.topic=connect-offsets
config.storage.topic=connect-configs
status.storage.topic=connect-status
offset.storage.replication.factor=1
config.storage.replication.factor=1
status.storage.replication.factor=1
plugin.path=/usr/share/java,/usr/local/share/kafka/plugins

4.2 MySQL CDC配置：开启Binlog是第一步

Debezium要读取MySQL的变更，前提是MySQL开启了Binlog。这个步骤很多人会忘，尤其是DBA不太熟悉CDC场景的时候。

你需要确保MySQL的my.cnf里有以下配置：

[mysqld]
server-id=223344
log_bin=mysql-bin
binlog_format=ROW
binlog_row_image=FULL
expire_logs_days=7

几个关键点：

• server-id：每个CDC连接必须唯一，不能跟MySQL集群里其他节点重复
• binlog_format=ROW：必须用ROW模式，Statement和Mixed都不行
• binlog_row_image=FULL：记录完整行数据，否则Debezium拿不到更新前后的值

配置完重启MySQL，然后创建一个专门给Debezium用的用户：

CREATE USER 'debezium'@'%' IDENTIFIED BY 'password';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;

4.3 注册Connector：让Debezium跑起来

环境准备好了，接下来注册一个MySQL Connector。通过Kafka Connect的REST API来操作：

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
  http://localhost:8083/connectors/ \
  -d '{
    "name": "mysql-connector-orders",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "database.hostname": "192.168.1.100",
      "database.port": "3306",
      "database.user": "debezium",
      "database.password": "password",
      "database.server.id": "223344",
      "database.server.name": "dbserver1",
      "database.include.list": "order_db",
      "table.include.list": "order_db.orders,order_db.order_items",
      "database.history.kafka.bootstrap.servers": "localhost:9092",
      "database.history.kafka.topic": "schema-changes.order_db",
      "include.schema.changes": "true",
      "snapshot.mode": "initial"
    }
  }'

这里有几个参数我重点说一下：

参数	说明	我的建议
database.server.name	逻辑名称，会作为Kafka Topic的前缀	用有业务含义的名字，别用test123
table.include.list	要监听的表，格式：库名.表名	尽量精确，别用通配符，避免数据泄露
snapshot.mode	启动时的快照策略	生产环境用initial或when_needed
database.history.kafka.topic	存储DDL变更历史的Topic	这个Topic很重要，别随便删

注册成功后，Kafka Connect会自动创建Topic。Topic的命名规则是：{database.server.name}.{database.name}.{table.name}。比如上面配置会生成：dbserver1.order_db.orders。

4.4 SnapShot与Streaming模式：两种工作状态

Debezium启动后，会经历两个阶段：Snapshot和Streaming。理解这两个模式的区别，是用好Debezium的关键。

4.4.1 Snapshot模式：先拍一张快照

当Connector第一次启动时，Debezium会先对表做一次全量快照。说白了就是把当前表里的所有数据读一遍，每条记录作为一个Create事件发到Kafka。

Snapshot有几种模式：

• initial：启动时做快照，之后切到流式
• when_needed：只在需要时做快照（比如Schema变更后）
• never：不做快照，直接开始流式读取
• schema_only：只捕获Schema，不读数据

我个人最常用的是initial。为什么？因为大多数场景下，你既需要历史数据，也需要实时增量。但要注意，如果表很大（比如上亿行），Snapshot可能会持续很久。我遇到过一张10亿行的订单表，Snapshot跑了将近两个小时。

4.4.2 Streaming模式：持续监听变更

Snapshot完成后，Debezium自动切换到Streaming模式。这时候它就像一个MySQL Slave，持续读取Binlog，把INSERT、UPDATE、DELETE转换成事件发到Kafka。

Streaming模式下，每条消息的格式是这样的：

{
  "schema": { ... },
  "payload": {
    "before": null,           // 更新前的数据，DELETE时有值
    "after": {                // 更新后的数据，INSERT和UPDATE时有值
      "id": 1001,
      "customer_name": "张三",
      "total_amount": 299.00,
      "status": "PAID"
    },
    "source": {
      "version": "1.9.7.Final",
      "connector": "mysql",
      "name": "dbserver1",
      "ts_ms": 1695000000000,
      "snapshot": false,      // false表示这是流式数据
      "db": "order_db",
      "table": "orders",
      "server_id": 123,
      "gtid": null,
      "file": "mysql-bin.000023",
      "pos": 45678,
      "row": 0
    },
    "op": "c",                // c=create, u=update, d=delete, r=snapshot
    "ts_ms": 1695000001000
  }
}

你看，每条消息都包含了before和after，还有op字段标识操作类型。这个结构在实时数仓里非常有用——你可以根据op来决定是INSERT、UPDATE还是DELETE到Iceberg表。

4.5 Schema变更处理：最棘手的部分

好，接下来聊一个让很多人头疼的问题——Schema变更。说白了就是MySQL表结构改了，比如加了列、改了列类型，Debezium怎么处理？

Debezium通过Schema History Topic来追踪DDL变更。每次检测到ALTER TABLE，它会：

1. 把DDL语句写入database.history.kafka.topic
2. 更新内部的Schema缓存
3. 后续的消息使用新的Schema结构

但这里有个大坑——Schema兼容性。如果下游消费者（比如Flink或者Kafka Connect Sink）还在用旧的Schema解析消息，就会报错。

处理Schema变更，我有几条建议：

• 使用Schema Registry：Confluent Schema Registry可以管理Schema版本，支持向前/向后兼容
• 设置合理的兼容性策略：我一般用BACKWARD或FULL，确保旧消费者也能处理新消息
• 监控Schema变更频率：如果一天改好几次表结构，说明设计有问题
• DDL变更前通知数据团队：让下游做好准备，避免消费中断

另外，Debezium的include.schema.changes参数可以控制是否把DDL事件也发到Kafka。如果你需要实时同步Schema变更到Iceberg，建议设为true。

4.6 整体架构图

说了这么多，咱们用一张图把整个流程串起来：

这张图把整个流程分成了三个阶段：Snapshot全量读取、Streaming增量读取、以及贯穿始终的Schema变更处理。你想想看，Debezium其实就是在模拟一个MySQL Slave，只不过它把Binlog事件转成了Kafka消息。

4.7 避坑指南

最后，分享几个我实战中遇到的坑：