4、MySQL CDC Connector详解：MySQL CDC Source配置、Server ID与并行读取、Debezium配置项、表结构变更处理

好，咱们进入第四章。这一章我打算把 MySQL CDC Connector 掰开了揉碎了讲清楚。说实话，这个 Connector 是整个增量数据管道里最核心、也最容易踩坑的一环。很多同学在社区里问我：“为什么我的 CDC 任务跑着跑着就挂了？” 或者 “为什么数据延迟越来越大？” 答案往往就藏在这一章里。

我会从四个维度来拆解：Source 配置、Server ID 与并行读取、Debezium 配置项、表结构变更处理。每个点我都会结合自己踩过的坑来讲，希望能帮你少走弯路。

4.1 MySQL CDC Source 配置：从入门到实战

先看一个最基础的配置。我个人习惯用 YAML 或者 Java API 来定义 Source，但不管哪种方式，核心参数就那些。

// 一个典型的 MySQL CDC Source 配置
DataStreamSource<String> mysqlSource = env.addSource(
    MySqlSource.<String>builder()
        .hostname("192.168.1.100")
        .port(3306)
        .databaseList("shop")          // 监控的数据库
        .tableList("shop.orders")      // 监控的表，格式：db.table
        .username("cdc_user")
        .password("cdc_pass")
        .serverTimeZone("Asia/Shanghai")
        .startupOptions(StartupOptions.initial())
        .deserializer(new JsonDebeziumDeserializationSchema())
        .build()
);

这里有几个关键点，我挨个说一下：

  • hostname / port：别用 localhost，生产环境一定要写具体的 IP 或域名。我曾经见过有人配了 localhost，结果 Flink 集群和 MySQL 不在同一台机器上，任务直接起不来。
  • databaseList / tableList：支持正则表达式。比如 shop.* 可以匹配所有以 shop 开头的数据库。但注意，tableList 的格式必须是 数据库名.表名。
  • startupOptions：有 initial()、latest()、timestamp()、specificOffset() 四种。我建议初次上线用 initial()，它会先做一次全量快照，然后无缝切换到增量。如果只想要增量，用 latest()。
  • deserializer：JsonDebeziumDeserializationSchema 是最常用的，输出 JSON 字符串。如果你需要强类型，可以用 DebeziumDeserializationSchema 自定义。
我的小技巧： 在测试环境，我会把 serverTimeZone 显式设成 Asia/Shanghai。如果不设，默认用 JVM 时区，有时候会导致时间字段解析出错。这个坑我踩过两次，后来就长记性了。

4.2 Server ID 与并行读取：别再踩这个坑了

Server ID 是 MySQL CDC 里最容易出问题的地方。说白了，每个 CDC 连接在 MySQL 看来都是一个“从库”，需要有一个唯一的 Server ID。如果多个连接用了相同的 Server ID，MySQL 会直接断开连接。

先看配置：

.serverId("5400-5403")  // 给每个并行子任务分配一个唯一的 Server ID

这里 5400-5403 表示分配 4 个 Server ID（5400, 5401, 5402, 5403），对应并行度 4。如果你设了并行度 4，但只给了 3 个 ID，任务会报错。

关于并行读取，我想多说两句。Flink CDC 的并行度体现在两个层面：

  • 全量阶段：多个并行子任务同时读取不同的表或分片。比如你有 10 张表，并行度设 5，每个子任务读 2 张表。
  • 增量阶段：所有并行子任务共享一个 binlog 连接，因为 binlog 是单线程写入的，并行度再高也没用。
注意： 并行度不是越高越好。全量阶段并行度太高，可能会把 MySQL 的 IO 打满。我建议从 2-4 开始试，观察 MySQL 的 CPU 和 IO 压力再调整。

我曾经在一个项目里，把并行度设成了 16，结果 MySQL 的磁盘 IO 直接飙到 100%，业务查询全堵住了。后来降到 4，任务跑了 3 个小时才完成全量。所以，并行度要跟 MySQL 的硬件能力匹配。

4.3 Debezium 配置项：隐藏的调优开关

Flink CDC 底层依赖 Debezium 来解析 binlog。Debezium 有很多配置项，但大部分用默认值就行。我挑几个最常用的讲一下。

配置项 默认值 说明 我的建议
debezium.snapshot.mode initial 全量快照模式：initial / when_needed / never / schema_only 生产环境用 initial，测试环境用 schema_only 加速
debezium.snapshot.fetch.size 1024 全量阶段每次读取的行数 如果表数据量大，可以调到 5000-10000，减少网络往返
debezium.snapshot.locking.mode minimal 全量快照时的锁策略：minimal / none / extended 用 minimal 或 none，避免长时间锁表
debezium.binlog.buffer.size 1024 binlog 事件缓冲区大小 如果延迟高，可以调到 2048 或 4096
debezium.max.queue.size 8192 Debezium 内部事件队列大小 如果出现背压，适当调大，但别超过 65536

配置方式很简单，在 MySqlSource 的 builder 里加：

.debeziumProperties(new Properties() {{
    put("snapshot.mode", "initial");
    put("snapshot.fetch.size", "5000");
    put("snapshot.locking.mode", "none");
    put("binlog.buffer.size", "2048");
}})

核心原则： 全量阶段追求速度，增量阶段追求稳定。全量时调大 fetch.size 和并行度；增量时关注 max.queue.size 和 binlog.buffer.size，防止背压导致延迟。

4.4 表结构变更处理：DDL 来了怎么办？

这是 CDC 里最头疼的问题。MySQL 的表结构变了（比如加了一列），Flink CDC 怎么处理？

默认情况下，Flink CDC 会忽略 DDL 事件。也就是说，你在 MySQL 里加了列，Flink 这边不会自动感知，数据照常按旧 schema 解析。这会导致什么问题？

  • 如果新增的列有默认值，旧数据正常，新数据会多出一个字段，但 Flink 的 schema 里没有，反序列化会报错。
  • 如果删了一列，Flink 解析时发现字段缺失，也会报错。

那怎么处理呢？我总结了三种方案：

  1. 重启任务：最粗暴的方式。修改 DDL 后，重启 Flink 任务，重新拉取 schema。适合低频变更的场景。
  2. 使用 Schema Registry：比如 Confluent Schema Registry，CDC 事件里带上 schema 版本，下游根据版本号动态解析。适合微服务架构。
  3. 自定义反序列化器：在 DebeziumDeserializationSchema 里捕获 DDL 事件，手动更新 schema。适合对实时性要求高的场景。

我个人推荐第二种方案。为什么呢？因为第一种太暴力，第三种太复杂。Schema Registry 是业界标准做法，Flink 社区也有现成的集成方案。

避坑指南： 我曾经在一个项目里，业务方频繁修改表结构（一天改三次），我用了方案一，每次都要重启任务，导致数据断流。后来换成 Schema Registry，问题才解决。所以，如果 DDL 变更频繁，一定要上 Schema Registry。

另外，还有一个细节：DDL 事件本身也会被写入 binlog。如果你用 JsonDebeziumDeserializationSchema，DDL 事件会被序列化成 JSON，但 op 字段是 c（create）或 a（alter）。你可以在下游过滤掉这些事件，或者单独处理。

4.5 本章知识体系总览

为了让你更直观地理解这一章的内容，我画了一张图，把 MySQL CDC Connector 的核心模块和流程串起来。

MySQL CDC Connector 核心架构图 MySQL CDC Connector 核心架构 MySQL 数据库 全量快照读取 Binlog 增量监听 Server ID 分配 Debezium 引擎 DDL 变更处理 Flink CDC Source 说明： 全量快照阶段：多个并行子任务通过不同 Server ID 同时读取表数据。 增量阶段：所有子任务共享一个 Binlog 连接，DDL 事件通过 Debezium 引擎处理。

从这张图你可以看到，MySQL CDC Connector 的核心流程是：MySQL 产生数据 → 全量快照或 Binlog 监听 → Debezium 引擎解析 → Flink CDC Source 输出。Server ID 管理、DDL 处理是贯穿始终的横切关注点。

4.6 本章小结

这一章的内容比较多，我帮你梳理一下重点：

  • Source 配置：注意 databaseList 和 tableList 的格式，startupOptions 选对模式。
  • Server ID：必须唯一，并行度与 Server ID 数量匹配。
  • Debezium 配置：全量阶段调 fetch.size，增量阶段调 max.queue.size。
  • DDL 处理：默认忽略，建议上 Schema Registry 或自定义反序列化器。
← 上一章 📖 返回目录 下一章 →