数据源连接器之MySQL CDC：Binlog原理与Flink实战

说到实时数据同步，MySQL CDC 是绕不开的核心技术。我个人做流式元数据同步项目时，第一关就是搞定 MySQL 的 Binlog。说白了，没有 Binlog，实时同步就是空谈。今天咱们就把它彻底讲透。

MySQL Binlog 到底是什么？

Binlog，全称 Binary Log，是 MySQL 的二进制日志。它记录了所有数据变更操作——INSERT、UPDATE、DELETE，甚至表结构变更。你想想看，数据库每发生一次写操作，Binlog 就会追加一条记录。

我刚开始接触时有个误区：以为 Binlog 和 Redo Log 是一回事。其实完全不同。Redo Log 是 InnoDB 存储引擎层面的，用于崩溃恢复；Binlog 是 MySQL Server 层面的，用于主从复制和数据恢复。咱们做 CDC，盯的就是 Binlog。

Binlog 的三种格式

格式 记录方式 特点 CDC 适用性
STATEMENT 记录 SQL 语句 体积小，但上下文依赖强 不推荐，数据一致性难保证
ROW 记录每行变更前后数据 体积大，但精确可靠 强烈推荐，Flink CDC 首选
MIXED 混合模式 MySQL 自动选择 可用，但不如 ROW 稳定

关键结论：做 Flink CDC 同步，Binlog 格式必须设置为 ROW。否则你拿到的数据可能不完整，甚至出现主键冲突。我在生产环境踩过这个坑，后来统一强制 ROW 格式。

Binlog 的核心机制

Binlog 文件是顺序写入的，每个文件写满后会滚动生成新文件。MySQL 内部维护一个 binlog position，CDC 客户端可以从任意位置开始消费。这就是 Flink 能做到断点续传的基础。

嗯，这里要注意：Binlog 不会永久保留。MySQL 有自动清理机制，通过 expire_logs_days 参数控制。我建议至少保留 3 天，给 CDC 任务留足容错时间。

Flink MySQL CDC Connector 配置与使用

Flink 官方提供的 mysql-cdc 连接器，底层封装了 Debezium 引擎。它监听 Binlog，把数据变更转成 Flink 能处理的 DataStream。说白了，你只需要配好连接信息，剩下的框架帮你搞定。

环境准备

先确认 MySQL 开启了 Binlog。登录 MySQL 执行：

SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';

如果 log_bin 是 OFF，或者 binlog_format 不是 ROW，需要修改 MySQL 配置文件并重启。我曾经遇到一个客户，Binlog 没开，折腾了半天才发现——这种低级错误最耽误时间。

另外，CDC 读取 Binlog 需要一个有权限的账号：

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'cdc_user'@'%' IDENTIFIED BY 'your_password';
FLUSH PRIVILEGES;

Maven 依赖

在 pom.xml 中添加：

<dependency>
    <groupId>com.ververica</groupId>
    <artifactId>flink-connector-mysql-cdc</artifactId>
    <version>2.4.0</version>
</dependency>

注意版本要和 Flink 匹配。我用的是 Flink 1.17 + CDC 2.4，稳定运行了大半年。

核心配置参数

参数 说明 推荐值
hostname MySQL 主机地址 实际 IP 或域名
port MySQL 端口 3306
database-name 要监听的数据库 支持正则，如 db_.*
table-name 要监听的表 支持正则，如 user_.*
server-id CDC 客户端唯一 ID 5400-5400 范围，避免冲突
scan.startup.mode 启动模式 initial / latest-offset
debezium.snapshot.mode 快照模式 initial（全量+增量）

我的经验：server-id 必须全局唯一。如果多个 CDC 任务用了相同 ID，MySQL 会认为连接冲突，导致其中一个任务断开。我曾经排查了整整一天才发现是 ID 重复的问题。

代码实战：单表同步

下面是一个完整的 Flink CDC 示例，监听 MySQL 的 orders 表，实时输出变更数据：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

public class MySQLCDCSingleTable {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("192.168.1.100")
            .port(3306)
            .databaseList("ecommerce")
            .tableList("ecommerce.orders")
            .username("cdc_user")
            .password("cdc_pass")
            .serverId("5400")
            .deserializer(new JsonDebeziumDeserializationSchema())
            .build();

        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source")
            .print();

        env.execute("MySQL CDC Single Table Sync");
    }
}

启动后，对 orders 表做任何 INSERT、UPDATE、DELETE，控制台都会打印 JSON 格式的变更事件。你想想看，这就是实时同步的起点。

启动模式详解

scan.startup.mode 有两个常用值：

  • initial：先做全量快照，再切换到增量 Binlog 监听。适合首次启动。
  • latest-offset：直接从当前 Binlog 位置开始，不读历史数据。适合已有全量数据、只需增量同步的场景。

避坑指南：我曾经在生产环境用 latest-offset 启动，结果发现丢了一部分数据。原因是任务启动前有数据写入，但 Binlog 位置已经跳过了。后来我改成 initial，先做快照保证数据完整，再切增量。除非你确定历史数据不需要，否则别省这一步。

Binlog 到 Flink 的数据流转

为了让你更直观地理解整个过程，我画了一张流程图：

MySQL CDC 数据流转流程 MySQL 数据库 Binlog 文件 Debezium 引擎 解析 Binlog 事件 Flink CDC Source DataStream 事件流 ... 全量快照阶段： 首次启动时，Flink CDC 先读取表全量数据，记录当前 Binlog 位置，然后切换到增量监听。 增量监听阶段： Debezium 持续读取 Binlog，将 INSERT/UPDATE/DELETE 事件转为 ChangeEvent，Flink 按事件时间处理。 Checkpoint 容错： Flink 定期保存 Binlog 消费位置（offset）。任务重启时，从上次 Checkpoint 位置恢复，保证 Exactly-Once。

从图上你能看到，整个链路分三步：MySQL 产生 Binlog → Debezium 解析成事件 → Flink 消费并处理。其中 Checkpoint 机制保证了数据不丢不重。我个人觉得，理解这个流程比背配置参数更重要——参数可以查文档，但架构思维得自己建立。

多表监听与正则匹配

实际项目中，往往需要监听多张表甚至整个数据库。Flink CDC 支持正则表达式：

.databaseList("ecommerce")
.tableList("ecommerce.orders|ecommerce.users|ecommerce.products")

或者用通配符：

.databaseList("ecommerce")
.tableList("ecommerce.order_.*")

嗯，这里要注意：tableList 的格式是 数据库名.表名，中间用点号分隔。多个表用竖线 | 连接。我习惯把需要同步的表名写在一个配置文件里，启动时动态加载，方便运维调整。

常见问题与避坑

  • Binlog 太大撑爆磁盘？ 设置 expire_logs_days=3，配合 max_binlog_size=500M 控制单个文件大小。
  • CDC 任务重启后重复消费？ 开启 Flink Checkpoint，设置 env.enableCheckpointing(5000)，保证 Exactly-Once 语义。
  • MySQL 主从切换怎么办？ CDC 连接的是主库地址。如果发生切换，需要手动更新连接配置并重启任务。目前 Flink CDC 还不支持自动跟随主从切换。
  • DDL 变更怎么处理？ Flink CDC 默认只捕获 DML（数据变更），不捕获 DDL（结构变更）。如果需要同步表结构变化，得额外配置 Debezium 的 include.schema.changes 参数。

我曾经踩过的坑：有一次 MySQL 做了大表 ALTER TABLE，耗时 10 分钟。这期间 Binlog 暂停写入，CDC 任务一直等待，导致下游数据延迟飙升。后来我加了一个超时监控，如果 Binlog 超过 30 秒没有新事件，就触发告警。这种问题不常见，但遇到了就很头疼。

小结

MySQL CDC 的核心就三件事：Binlog 格式配成 ROW、Flink CDC 连接器配好参数、Checkpoint 保证容错。你只要把这三步走稳，实时同步的基础就搭好了。剩下的就是根据业务需求做数据转换和写入下游——那些咱们后面章节再聊。

← 上一章 📖 返回目录 下一章 →