Flink CDC 3.0 实战：MySQL 到 Doris 实时数据同步

简介：尚硅谷研究院出品的Flink CDC 3.0实战文档，面向具备一定Flink基础的大数据研发人员，系统讲解如何利用Flink CDC实时捕获MySQL变更数据并同步至Doris，覆盖Streaming ETL完整链路。文档先厘清CDC核心概念，对比基于查询与基于Binlog两种模式的差异，再深入flink-cdc-connectors组件的使用，并通过DataStream与Flink SQL两种方式给出可复现的实操案例，包含数据准备、环境搭建、配置编写、任务启动与验证测试等环节，同时点明开启Binlog、设置检查点等关键配置，帮助读者规避常见坑点、确保数据一致性与任务稳定性。压缩包共1个docx文档，大小仅145KB，篇幅凝练但配有完整代码示例和操作指南，可直接作为项目部署参考。已有1788人浏览学习，适合需要快速上手MySQL到Doris实时同步的工程师使用。

1. Flink CDC 3.0 实战定位：一条流把 MySQL 变更打进 Doris

业务库里的一条 update，往往要等小时级批任务跑完才会体现在分析报表里。Flink CDC 3.0 要解决的正是这个痛点：通过解析 MySQL Binlog，把插入、更新、删除转成流式事件，再同步到 Doris 分析引擎。它的价值不只是"实时"，而是让全量快照与增量日志无缝衔接，配合 3.0 新增的 YAML Pipeline 能力，整库同步和自动建表都省掉了手写代码。这套资源的编排逻辑是从 CDC 概念讲起，再落 DataStream、FlinkSQL 和 Doris 同步三个实操场景，适合已有 Flink 基础、但没系统做过 CDC 落地的研发人员。

2. 先立理论：基于查询与基于 Binlog 的 CDC，为什么 3.0 选后者

2.1 两种 CDC 的本质差异：延迟、压力、捕获能力

CDC 全称 Change Data Capture，核心动作是监测并捕获数据库的变动，把插入、更新、删除按发生顺序完整记录下来，交给消息中间件或其他服务消费。这里最容易混的一点是：不是所有叫 CDC 的工具都实时，实现路径不同，效果天差地别。

基于查询的 CDC 属于"拉"模式，代表产品 Sqoop、DataX。它们周期性去查源表，通过比对主键或时间戳找出变化行。问题在于两个：一是高延迟，轮询间隔再短也有窗口；二是给数据库带来额外压力，大表全扫一遍成本极高。基于 Binlog 的 CDC 属于"推"模式，代表产品 Canal、Maxwell、Debezium。它们作为 MySQL 的从节点订阅 Binlog，数据库把变更主动推过来，延迟在秒级甚至毫秒级，且几乎不增加数据库压力，还能捕获删除和结构变更。

两类 CDC 的对比，直接决定选型：

这张表的信息量在于"能否捕获所有数据"。基于查询的 CDC 只能看到查询时刻的数据差异，中间过程的中间态丢失；基于 Binlog 则能完整回放一条记录的多次变更，这对实时数仓的准确性和可追溯性非常关键。

2.2 Flink CDC 3.0 架构变化：从 Source 组件到 Pipeline 引擎

Flink 社区维护的 flink-cdc-connectors 组件，把 MySQL、PostgreSQL 等数据库的读取能力封装成了可直接接入 Flink 的 Source。2.x 时代常用的做法是引入 flink-connector-mysql-cdc，然后通过 MySqlSource 构建 SourceFunction，在 DataStream API 里消费。这种方式没问题，但有几个痛点：整库同步要自己写循环、schema 变更要手工处理、代码量不小。

3.0 是这条产品线的一次大改版。表面上它仍然提供 MySqlSource 给 DataStream 和 SQL 用，但真正的新东西是引入了 Pipeline 同步机制。用户不再需要写 Java 代码，而是编写一个 YAML 配置文件，描述 source、sink、pipeline 三段，再用 bin/flink-cdc.sh 命令直接启动同步任务。

这种设计对生产环境的吸引力在于：整库同步时，表名和库名都支持正则表达式匹配，新表加入源库后，任务能自动识别；同步过程中目标端表结构也能随源端变化，自动执行加列等 DDL。这正是"变更数据捕获"的完整闭环——不只数据要实时，结构变化也要实时。

2.3 启动模式 initial 到底发生了什么

DataStream 方式里有一段.startupOptions(StartupOptions.initial())，这是 CDC 新手最容易忽视的配置。initial 模式表示任务第一次启动时，先对监控的表做一次全量快照，快照结束再无缝切换为读取最新 Binlog。

这里有一个容易误解的点：initial 不是简单地"先查一遍再听日志"。它内部会先把当前 Binlog 位点记录下来，再开始快照，保证快照期间的增量变更不丢。这种做法比先全量后增量的两段式方案可靠得多，但代价是任务启动时间变长，大表快照可能要跑很久。

其他几种模式也值得心里有数：earliest从 Binlog 最早位置开始读，前提是 Binlog 还完整保留；latest只从启动时刻之后的变更开始读，错过历史数据；specificOffset和timestamp则是从指定位点恢复，常用于故障恢复。生产环境里，首次上线用 initial，断点续传从 Savepoint 起，一般不会错。

3. DataStream 与 FlinkSQL：把 MySQL 变更读成流

3.1 环境与依赖准备

实操的第一步是准备工程依赖。Flink 版本 1.18.0，Flink CDC 版本 3.0.0，MySQL 8.0.31。依赖坐标如下，直接放进 pom.xml 即可：

<properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> <flink-version>1.18.0</flink-version> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink-version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink-version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink-version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_2.12</artifactId> <version>${flink-version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java-bridge</artifactId> <version>${flink-version}</version> </dependency> <dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>3.0.0</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>8.0.31</version> </dependency> </dependency>

依赖里有几个值得注意的点。flink-table-planner 和 flink-table-api-java-bridge 是给 FlinkSQL 用的，如果只跑 DataStream 可以去掉，但建议整套带上，方便在同一工程里切换两种方式。mysql-connector-java 是 JDBC 驱动，连接元数据时需要；flink-connector-mysql-cdc 才是真正负责解析 Binlog 的组件，它内部依赖 Debezium 引擎。打包时用 maven-assembly-plugin 打成 jar-with-dependencies，避免提交集群时缺类。

3.2 DataStream 方式：MySqlSource 构建与参数说明

DataStream 方式的核心代码不长，关键参数却一个都不能错。完整代码结构如下：

import com.ververica.cdc.connectors.mysql.source.MySqlSource; import com.ververica.cdc.connectors.mysql.table.StartupOptions; import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.restartstrategy.RestartStrategies; import org.apache.flink.api.common.time.Time; import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; import org.apache.flink.streaming.api.CheckpointingMode; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.CheckpointConfig; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import java.util.Properties; public class FlinkCDCDataStreamTest { public static void main(String[] args) throws Exception { // 1. 准备流处理环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. 开启 Checkpoint，Flink CDC 将 Binlog 读取位点存在状态中， // 只有基于 Checkpoint 或 Savepoint 重启才能实现断点续传 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); // 3. 设置状态后端，生产环境建议放在 HDFS 上 env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC"); System.setProperty("HADOOP_USER_NAME", "atguigu"); // 4. 构建 MySQL CDC Source MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("hadoop103") .port(3306) .databaseList("test") .tableList("test.t1") .username("root") .password("000000") .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); // 5. 接入 Flink 流处理 DataStreamSource<String> mysqlDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySqlSource"); // 6. 打印输出 mysqlDS.print(); // 7. 执行任务 env.execute(); } }

这段代码里，.databaseList("test")和.tableList("test.t1")决定了监控范围。databaseList 支持逗号分隔多个库，tableList 的格式必须是"库名.表名"。这里有一个常见误区：tableList 写 "test.t1" 只同步这一张表，如果写 "test.*" 才能匹配 test 库下所有表。

deserializer 使用了 JsonDebeziumDeserializationSchema，把 Debezium 的 SourceRecord 转成 JSON 字符串。输出格式里包含 before、after、op 三个关键字段：op 为 c 代表 insert，u 代表 update，d 代表 delete。下游解析时要注意，update 事件里 before 和 after 同时存在。

3.3 FlinkSQL 方式：mysql-cdc connector 注册表

FlinkSQL 方式不需要构建 Source 对象，而是通过 CREATE TABLE 语句声明一张动态表，把 CDC Source 映射成表。这段代码在资源里很短，但生产环境中它是更常用的方式，因为后续所有计算都可以直接用 SQL 表达。

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; public class FlinkSQLTest { public static void main(String[] args) { // 1. 创建流处理环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. 创建表环境 StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); // 3. 注册 MySQL CDC 表 tableEnv.executeSql("CREATE TABLE t1 (\n" + " id STRING PRIMARY KEY NOT ENFORCED,\n" + " name STRING\n" + ") WITH (\n" + " 'connector' = 'mysql-cdc',\n" + " 'hostname' = 'hadoop103',\n" + " 'port' = '3306',\n" + " 'username' = 'root',\n" + " 'password' = '000000',\n" + " 'database-name' = 'test',\n" + " 'table-name' = 't1'\n" + ")"); // 4. 查询并打印 Table table = tableEnv.sqlQuery("select * from t1"); table.execute().print(); } }

SQL 方式里PRIMARY KEY NOT ENFORCED是必须写的，它告诉 Flink 主键由源端保证，Flink 不做约束校验。表的字段类型必须与 MySQL 对应，VARCHAR 对应 STRING，INT 对应 INT。'scan.incremental.snapshot.enabled'这个参数 3.0 默认开启，分片并发读取全量数据，如果表数据量不大也可以显式关闭。

两种方式怎么选，我的经验是：纯同步场景用 SQL 方式，能直接落入 FlinkSQL 生态，方便后续 join 维表和聚合；涉及复杂状更新或者对输出格式有特殊要求时用 DataStream 方式，序列化逻辑更可控。这套资料里两种都给了完整代码，正好可以对照学习。

3.4 检查点与状态后端：断点续传必须做对

刚才 DataStream 代码里 Checkpoint 相关配置占了一半篇幅，这不是凑数。Flink CDC 消费 Binlog 的位置是作为状态存储在 Flink 的 Checkpoint 里的。如果不开 Checkpoint，任务一挂，位点即丢，重启后要么从最新读造成数据缺口，要么从头读造成重复。

配置里需要注意三个细节。setMinPauseBetweenCheckpoints(3000L)表示两个 Checkpoint 之间至少间隔 3 秒，避免频繁快照拖垮性能；RETAIN_ON_CANCELLATION表示任务手动取消时保留最后一次 Checkpoint，这样才能从那个位点恢复；setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC")指定快照存放位置，单机测试可以放在本地文件系统，生产必须放 HDFS 或对象存储。

状态后端这里用了 HashMapStateBackend，它在内存中管理状态，适合中小规模任务。如果状态很大，3.0 之前常用 RocksDB，但需要额外引入依赖。从这套资料看，它的场景重点在演示 CDC 同步链路，HashMap 足够。

4. MySQL 到 Doris 的 3.0 YAML 同步：一条命令跑通全库

4.1 Pipeline 同步和手写代码有什么不同

前两章的方式都是写程序、打包、提交，属于"开发者视角"。Flink CDC 3.0 把整库同步变成了"配置文件 + 一条命令"的操作，这就是 Pipeline 机制。

Pipeline 的定位是开箱即用的数据同步工具，对标的是 DataX 这类离线工具，但它是流式的。它把 flink-cdc-pipeline-connector-mysql 作为源端，flink-cdc-pipeline-connector-doris 作为目标端，两者之间由 3.0 内核串联。启动方式不再是 flink run，而是：

bin/flink-cdc.sh job/mysql-to-doris.yaml

这个命令会启动一个独立的同步任务，任务的具体行为由 YAML 文件控制。这种设计把"开发"门槛降到了"配置"，团队里的数据工程师不需要深入 Flink API 也能维护同步链路。

4.2 同步变更配置详解：source / sink / pipeline 三段式

3.0 的 YAML 配置分为三大段，结构清晰。下面这份配置是这套资料里 MySQL 到 Doris 同步变更的完整示例：

source: type: mysql hostname: hadoop103 port: 3306 username: root password: "000000" tables: test.\\.* server-id: 5400-5404 server-time-zone: UTC+8 sink: type: doris fenodes: hadoop102:7030 username: root password: "000000" table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1

逐段说明。source 段里tables: test.\\.*表示匹配 test 库下的所有表，配置的是正则表达式。这里有一个 YAML 转义问题，正则中的.*在 YAML 里要写成.\\.*，第一次写容易漏。.hostname如果生产用了高可用架构，可以填域名而不是单点 IP。

sink 段是 Doris 专用配置。fenodes是 Doris FE 的地址和 HTTP 端口，默认 8030。light_schema_change设置为 true 后，Doris 支持轻量级 schema 变更，源端 MySQL 加列时，Doris 可以通过快速模式同步；如果关闭，DDL 同步会走 DORIS 的普通 ALTER 流程，耗时更长。replication_num设为 1 是因为测试环境通常只有单副本，生产环境至少设 2 或 3。

pipeline 段的parallelism决定整个任务并发度。这里我踩过一个坑：并行度调大后，如果源端 MySQL 的 server-id 只有一个，多个并行的 Binlog 连接会共用同一个 server-id，MySQL 主库会直接拒绝多余的连接。解决办法就是把 server-id 配成一个区间，比如 5400-5404，对应最多 5 个并发连接。

4.3 路由变更：多库多表如何自动映射

路由变更和同步变更的配置结构相同，区别在于 source 段的表匹配策略。资料里给了 test_route 库的示例：

source: type: mysql hostname: hadoop103 port: 3306 username: root password: "000000" tables: test_route.\\.* server-id: 5400-5404 server-time-zone: UTC+8

如果只同步该库下的部分表，可以写成更精确的正则：

tables: test_route.\.(t1|t2)

这种路由机制的价值在于：新增表时不需要改代码，只要新表名能被正则匹配，任务会自动建表并开始同步。但它也有边界——正则只匹配表名，不匹配库名，跨库同步时需要写多条 source 规则，或者在 tables 字段里用逗号分隔多个模式。

4.4 任务启动与结果验证

启动前要做几件事：Flink 配置中开启 Checkpoint，Doris FE 和 BE 都要处于运行状态，Doris 里先创建好目标数据库。资料里的顺序是：

# 1. 开启 Flink Checkpoint vim conf/flink-conf.yaml # 添加 execution.checkpointing.interval: 5000 # 2. 启动 Flink 集群 bin/start-cluster.sh # 3. 启动 Doris FE 与 BE bin/start_fe.sh bin/start_be.sh # 4. 在 Doris 中创建目标数据库 mysql -uroot -p000000 -P9030 -hhadoop102 create database test; # 5. 启动同步任务 bin/flink-cdc.sh job/mysql-to-doris.yaml

验证阶段最直观的做法是：在 MySQL 的 test 库对应表执行 insert、update、delete，然后刷新 Doris 中 test 库的表。注意 Doris 默认有几十秒的可见性延迟，刚提交的事务不会立刻在查询中可见，要等版本更新完成。资料里测试时执行了"新增列操作"，这是同步链路最敏感的场景，如果 sink 配置里light_schema_change没开，加列操作很可能会把任务打挂。

5. 同步翻车集中地：CDC 实战中的常见问题与排查

5.1 开启 Binlog 后 MySQL 启动失败

现象：按资料修改/etc/my.cnf添加log-bin=mysql-bin、binlog_format=row、binlog-do-db=test后，重启 MySQL 报错退出，日志提示 server-id 缺失。

原因：只配置了 log-bin 但没有配置server-id，或者配置的值与其他节点重复。MySQL 5.7 之后，只要开启 binlog 就必须设置全局唯一的 server-id，否则拒绝启动。

解决：在 my.cnf 中显式配置server-id=1，并确保集群内各节点 server-id 不重复。另外binlog-do-db只指定了 test 和 test_route 两个库，其他库的变更不会进 binlog，如果后续新增同步库注意补充。

5.2 server-id 冲突导致连接被拒或数据错乱

现象：YAML 里配置server-id: 5400，任务并发度调到 4，启动后 TaskManager 日志频繁报连接异常，部分表数据缺失。

原因：Flink CDC 每个并发任务都会向 MySQL 注册一个 binlog 连接，连接使用相同的 server-id 时，MySQL 会认为这是重复的从库连接，后注册的连接被踢掉。

解决：把 server-id 改成区间配置，例如5400-5404。区间内每个 ID 会被分配给不同的并发任务，互不冲突。如果表数量少、并发度低，保持区间长度略大于并发数即可。

5.3 从 Savepoint 重启后读不到增量数据

现象：任务正常运行一段时间后手动 Cancel，再从 Savepoint 恢复，MySQL 中新增的数据没有被捕获。

原因：检查点未开启，或关闭方式不对。Flink CDC 的 Binlog 位点存在 Checkpoint 状态中，任务关闭时如果配置的是DELETE_ON_CANCELLATION，最后一次 Checkpoint 会被清理，Savepoint 自然找不到位点。

解决：确认代码里已执行enableCheckpointing，并设置enableExternalizedCheckpoints(RETAIN_ON_CANCELLATION)。从 Savepoint 恢复时命令是：

flink run -s hdfs://hadoop102:8020/flinkCDC/save/savepoint-xxx \ -c com.atguigu.cdc.FlinkCDCDataStreamTest ./gmall-flink-cdc.jar

注意 Savepoint 路径必须是 Flink 能访问的分布式存储，本地路径在集群模式下行不通。

5.4 Doris 目标表新增列后同步任务失败

现象：MySQL 源表执行 ALTER TABLE ADD COLUMN 后，同步任务日志报错，Doris 表结构未更新。

原因：Doris sink 默认没有开启轻量级 schema 变更，或者 Doris 版本不支持某些 DDL 类型。资料里使用的 Doris 版本是 1.2.4，支持light_schema_change，但必须显式配置。

解决：在 YAML 的 sink 段加上：

table.create.properties.light_schema_change: true table.create.properties.replication_num: 1

如果任务已经跑起来后发现这条配置没加，需要重启任务才能生效。另外，Doris 对 ADD COLUMN 和 MODIFY COLUMN 的支持程度不同，遇到不支持的 DDL 时，先在 Doris 手工执行一遍，再恢复任务同步。

5.5 时区问题：同步到 Doris 的时间字段差 8 小时

现象：MySQL 的 TIMESTAMP 字段在 Doris 中查询，时间比源库快了或慢了 8 小时。

原因：MySQL 驱动和 Flink CDC 在解析 Binlog 时间时使用的时区不一致。MySQL 默认使用会话时区，而 Flink 任务所在 JVM 默认是 UTC 时区。

解决：在 source 段配置server-time-zone: UTC+8，同时 MySQL 连接的 JDBC URL 里显式加serverTimezone=Asia/Shanghai。两处都设置后，时间字段才能对齐。这个问题在部署到容器环境时特别容易遇到，因为容器 JVM 默认时区通常不是 Asia/Shanghai。

6. 验证同步正确性：一条链路三处确认

同步任务跑起来不等于同步正确，我一般在每次 CDC 上线或者改过配置后，强制走一遍三处验证。

第一步看 Flink 日志。TaskManager 日志里能看到 Debezium 引擎打印的快照完成信息，以及后续每一条 binlog 事件的解析日志。重点确认 Source 成功注册、无频繁报错。第二步看 Doris 表结构和数据量。在 Doris 里执行show tables确认源表是否全部自动创建，再对比 MySQL 和 Doris 的行数：

-- MySQL 侧 SELECT COUNT(*) FROM test.t1; -- Doris 侧 SELECT COUNT(*) FROM test.t1;

第三步做变更验证。在 MySQL 里执行一条 insert 和一条 update：

INSERT INTO test.t1 VALUES('1004', 'zhaoliu'); UPDATE test.t1 SET name = 'zhangsan2' WHERE id = '1001';

然后在 Doris 侧查询这两条记录，确认数据和 op 类型都正确落库。如果涉及新增列，再执行ALTER TABLE test.t1 ADD COLUMN age VARCHAR(255);，观察 Doris 表结构是否自动更新。同时建议检查保存的 Checkpoint，确认 Binlog 位点在持续更新。

还有很多细节可以深挖，比如对 JSON 输出做二次处理、多表 join 的 CDC 语义、MySQL 大表快照性能调优。这套资料里都附了对应的工程代码、YAML 模板和操作步骤，照搬下来改一改 hostname 和密码，就能完整复现整个同步链路。我这个习惯也是从那次丢增量数据之后养成的——从那以后，我每次上线 CDC 任务都强制走一遍这三步检查，确认数据一致才敢更新下游报表逻辑。希望帮到你。

本文还有配套的精品资源，点击获取