Flink CDC替换Canal实战：如何实现MySQL到Kafka的99.99%数据一致同步

对数据的实时性要求越来越高。传统的离线数仓（T+1）已无法满足业务对秒级响应的需求，而实时数仓和数据湖（Data Lake）架构正成为主流。然而，如何将业务数据库中的变更数据（Insert/Update/Delete）低延迟、高可靠、无侵入地同步到下游系统，一直是构建实时链路的关键挑战。

CDC（Change Data Capture）技术，即变更数据捕获，在广义上泛指任何能捕获数据变更的技术。它主要分为基于直连查询的CDC和基于Binlog的CDC。

以MySQL为例，传统基于Binlog的CDC方案流程通常如下：通过Canal等工具监听Binlog，将日志数据写入Kafka消息队列，再由Flink消费Kafka数据进行处理和数据同步。整个流程包含多个阶段和组件，链路较长。

而Apache Flink CDC则提供了一种更简洁高效的方案，它能够直接从数据库（如MySQL）获取Binlog日志，供下游进行实时计算和分析，从而简化了架构。

这意味着数据无需再通过Canal与Kafka进行中转，由Flink直接处理数据库的变更流，节省了中间环节。

Flink CDC基于Apache Flink构建，能够直接从数据库的事务日志（如MySQL的binlog）中捕获数据变更，并以流式方式输出到Kafka、Pulsar、Iceberg、Hudi等下游系统。其核心价值体现在：

• 无侵入性：无需修改业务代码或添加触发器，仅通过读取数据库日志即可捕获变更。
• 端到端Exactly-Once语义：结合Flink Checkpoint机制，保障数据不丢不重。
• 统一处理模型：CDC数据以流的形式进入Flink，可无缝对接窗口计算、维表关联、状态管理等高级流计算功能。
• 入湖桥梁：作为连接OLTP系统与数据湖的关键组件，实现“实时入湖”（Real-time Lakehouse）。

因此，Flink CDC堪称是构建现代实时数据架构不可或缺的一环，是实时数据入湖的“第一公里”。

Flink CDC原理与实践

核心原理

Flink CDC的底层依赖于Debezium（开源CDC引擎），通过封装Debezium的Source Connector，将其集成到Flink的SourceFunction中。其工作流程如下：

1. 启动时全量快照：首先对源表进行一致性快照，读取当前全量数据。
2. 切换至增量日志：快照完成后，自动无缝切换到监听数据库的Binlog/Redo Log，捕获增量变更。
3. 统一事件格式输出：所有记录（包括快照和增量）都以统一的RowData或JSON格式输出，包含操作类型（INSERT/UPDATE/DELETE）、时间戳、前后镜像等信息。
4. Checkpoint保障一致性：利用Flink的检查点机制记录读取位置，确保故障恢复后的数据一致性。

注：Flink CDC 2.0+ 引入了“无锁快照”和“并行读取”机制，大幅提升大表初始化性能。

接入常见数据库实践

✅ 通过Flink DataStream API接入MySQL

下面演示如何编写一个Flink CDC应用程序，将MySQL表更改实时推送到Kafka Topic中。需要依赖flink-connector-jdbc和flink-connector-kafka库。

核心代码示例：

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

    Properties properties = new Properties();
    properties.setProperty("bootstrap.servers", "localhost:9092");
    properties.setProperty("group.id", "test-group");

    JdbcSource<RowData> source = JdbcSource.<RowData>builder()
            .setDrivername("com.mysql.jdbc.Driver")
            .setDBUrl("jdbc:mysql://localhost:3306/test_db")
            .setUsername("flink_cdc_user")
            .setPassword("password")
            .setQuery("SELECT id, name, age, email FROM test_table")
            .setRowTypeInfo(Types.ROW(Types.INT, Types.STRING, Types.INT, Types.STRING))
            .setFetchSize(1000)
            .build();

    DataStream<RowData> stream = env.addSource(source);
    // ... 后续处理逻辑，如写入Kafka
}

前提条件：

• 确保MySQL已开启Binlog，且格式为ROW：binlog_format=ROW，binlog_row_image=FULL。
• 用于连接的用户需具备REPLICATION SLAVE， REPLICATION CLIENT， SELECT权限。

运行示例及输出：
通过Flink命令行提交作业后，当源表数据发生变更时，作业会实时输出变更信息。

[INFO] Change data for table: mytable.
[INFO] Record key: {"id": 1}, record value: {"id": 1, "name": "Alice", "age": 25}.
[INFO] Change data for table: mytable.
[INFO] Record key: {"id": 1}, record value: {"id": 1, "name": "Alice", "age": 27}. // 年龄字段从25更新为27

✅ 通过Flink SQL接入MySQL CDC

使用Flink SQL更加简洁，以下是创建CDC源表并查询的示例：

-- 创建MySQL CDC源表
CREATE TABLE mysql_users (
  id INT PRIMARY KEY NOT ENFORCED,
  name STRING,
  email STRING,
  update_time TIMESTAMP(3)
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'flinkuser',
  'password' = 'flinkpw',
  'database-name' = 'test_db',
  'table-name' = 'users'
);

-- 查询并输出到控制台或其它Sink
SELECT * FROM mysql_users;

使用中的常见问题

Flink CDC高频面试题

Q1：Flink CDC和传统Canal / Maxwell有什么区别？

• 集成度：Flink CDC深度集成Flink，CDC数据可直接参与流计算；Canal/Maxwell通常作为独立中间件服务，需额外接入Flink等计算引擎。
• 语义保障：Flink CDC原生支持Checkpoint和Exactly-Once语义；使用Canal等方案时，位点管理需要自行实现，复杂度较高。
• 全量+增量一体化：Flink CDC支持自动切换全量快照与增量日志读取；Canal通常仅支持增量同步。

Q2：Flink CDC如何实现无锁快照？
Flink CDC 2.0引入Chunk-based Snapshot机制：

1. 将表按主键范围分片成多个Chunk。
2. 每个Chunk独立读取，并记录其高低水位位置。
3. 读取过程中允许对表进行并发写入操作，通过后续消费Binlog来补偿快照期间发生的变更。
4. 最终合并快照数据与增量变更日志，保证数据的一致性。

Q3：如何处理DDL变更（如加列）？

• 当前限制：Flink CDC默认不支持动态DDL同步，表结构变化可能导致作业报错或数据解析错误。
• 解决方案：
  • 手动重启作业（适用于低频DDL变更场景）。
  • 使用Schema Registry（如Confluent Schema Registry）结合Avro等支持Schema Evolution的格式。
  • 关注并尝试Flink 1.17+版本中Dynamic Table Options对Schema Evolution的实验性支持。

Q4：Flink CDC能否捕获DELETE操作？
可以。DELETE事件会以op=‘d’的形式输出，并包含删除前的完整行数据（前提是数据库日志为ROW格式且包含before image，如MySQL的binlog_row_image=FULL）。

Q5：如何优化大表CDC同步的性能？

• 升级到Flink CDC 2.3+版本，启用并合理设置parallelism参数。
• 根据源表主键分布情况，适当增加Source算子的并行度。
• 调整Checkpoint间隔，在保证一致性的前提下避免过于频繁的Checkpoint影响吞吐量。
• 对于全量同步阶段，可调整snapshot.fetch.size参数控制每次读取的数据量。

结语

Flink CDC正在成为构建实时数据管道的事实标准。它不仅简化了从数据库到数据湖的同步链路，降低了架构复杂度，还为实时分析、实时风控、实时推荐等场景提供了高质量、低延迟的数据源。随着社区对更多数据库的支持以及Schema Evolution等功能的增强，其在实时数仓与湖仓一体架构中的地位将愈发重要。

建议的学习和实践路径为：从MySQL CDC入手 → 接入Kafka进行数据分发 → 尝试写入Iceberg/Hudi/Paimon等数据湖格式 → 最终构建端到端的实时数据入湖Pipeline。