简介尚硅谷研究院出品的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 的对比直接决定选型维度基于查询Sqoop/DataX基于 BinlogCanal/Debezium执行模式BatchStreaming能否捕获全部变更否是延迟高低对数据库压力明显增加基本无典型场景离线批量抽取实时数仓、缓存同步这张表的信息量在于能否捕获所有数据。基于查询的 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.0Flink CDC 版本 3.0.0MySQL 8.0.31。依赖坐标如下直接放进 pom.xml 即可properties maven.compiler.source8/maven.compiler.source maven.compiler.target8/maven.compiler.target flink-version1.18.0/flink-version project.build.sourceEncodingUTF-8/project.build.sourceEncoding /properties dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-java/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-planner_2.12/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-api-java-bridge/artifactId version${flink-version}/version /dependency dependency groupIdcom.ververica/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.0.0/version /dependency dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId version8.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. 开启 CheckpointFlink 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 MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(hadoop103) .port(3306) .databaseList(test) .tableList(test.t1) .username(root) .password(000000) .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); // 5. 接入 Flink 流处理 DataStreamSourceString 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 代表 insertu 代表 updated 代表 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 对应 STRINGINT 对应 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: UTC8 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-idMySQL 主库会直接拒绝多余的连接。解决办法就是把 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: UTC8如果只同步该库下的部分表可以写成更精确的正则tables: test_route.\.(t1|t2)这种路由机制的价值在于新增表时不需要改代码只要新表名能被正则匹配任务会自动建表并开始同步。但它也有边界——正则只匹配表名不匹配库名跨库同步时需要写多条 source 规则或者在 tables 字段里用逗号分隔多个模式。4.4 任务启动与结果验证启动前要做几件事Flink 配置中开启 CheckpointDoris 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-binmysql-bin、binlog_formatrow、binlog-do-dbtest后重启 MySQL 报错退出日志提示 server-id 缺失。原因只配置了 log-bin 但没有配置server-id或者配置的值与其他节点重复。MySQL 5.7 之后只要开启 binlog 就必须设置全局唯一的 server-id否则拒绝启动。解决在 my.cnf 中显式配置server-id1并确保集群内各节点 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: UTC8同时 MySQL 连接的 JDBC URL 里显式加serverTimezoneAsia/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 和一条 updateINSERT 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 任务都强制走一遍这三步检查确认数据一致才敢更新下游报表逻辑。希望帮到你。本文还有配套的精品资源点击获取