简介这份资源是尚硅谷研究院编写的Flink CDC 3.0技术文档面向有一定Flink基础、从事大数据处理与实时数据同步的研发工程师帮助读者掌握变更数据捕获原理并落地MySQL到Doris的Streaming ETL链路。文档从CDC概念切入对比基于查询与基于Binlog两种方式的差异讲解flink-cdc-connectors组件的使用并给出数据准备、DataStream与Flink SQL两种实现路径以及MySQL同步至Doris的完整实操涵盖环境搭建、配置文件编写、任务启动与测试同时强调开启Binlog、设置检查点等关键配置。资源包为1个docx文件约145KB内容紧凑、代码示例与操作指南齐全便于快速上手并应用于实际项目。目前已有1788人学习适合希望系统理解Flink CDC并完成稳定数据同步实践的读者参考。1. 从一次凌晨三点的对账说起Flink CDC 3.0 到底解决了什么凌晨三点被叫起来对账MySQL 里订单状态已经改成已退款Doris 报表里还挂着已支付运营那边催着出数。这种场景做实时数仓的同行应该都不陌生——业务库和 OLAP 库之间的数据延迟轻则报表不准重则影响决策。Flink CDC 3.0 就是冲着这类问题来的它把 MySQL 的 Binlog 变更实时捕获出来经过 Flink 做流式 ETL再同步到 Doris 这类分析型数据库整条链路延迟能压到秒级。这份资源是一套完整的 Flink CDC 3.0 实战教程从 CDC 概念讲起覆盖 DataStream API、Flink SQL 两种开发方式最后落到 MySQL 到 Doris 的 Streaming ETL 全流程。适合有 Flink 基础、正在做实时数仓或数据同步的工程师也适合想从 Sqoop/DataX 批同步迁移到流式同步的团队。下面按原理选型 → 动手复现 → 踩坑排查 → 进阶技巧的顺序拆开讲。2. CDC 选型为什么基于 Binlog 的方案才是实时同步的正解2.1 基于查询 vs 基于 Binlog 的本质差异CDCChange Data Capture的核心是捕获数据库的变更。市面上主流方案分两类基于查询的Sqoop、DataX和基于 Binlog 的Canal、Maxwell、Debezium、Flink CDC。这两者的差异不是快一点慢一点而是执行模型根本不同。基于查询的方案本质是批处理靠定时轮询全表或增量字段来拉数据。它有两个硬伤一是无法捕获 DELETE 操作因为被删的行查不到了二是每次查询都要扫表对源库压力大。基于 Binlog 的方案则是流式的MySQL 把每次数据变更按顺序写进 Binlog 文件CDC 工具伪装成从库去订阅这个日志流插入、更新、删除全都能拿到而且对源库几乎无额外压力。对比维度基于查询Sqoop/DataX基于 BinlogFlink CDC执行模式BatchStreaming能否捕获 DELETE否是变化延迟高分钟到小时级低秒级对源库压力大全表扫描小读 Binlog断点续传依赖增量字段依赖 Checkpoint/Savepoint选型结论很直接只要业务对实时性有要求或者需要完整捕获删除操作就必须走 Binlog 路线。Flink CDC 在 Debezium 基础上做了 Flink 原生集成省掉了 Kafka 中间层链路更短。2.2 Flink CDC 3.0 的架构变化Flink CDC 3.0 相比 2.x 最大的变化是引入了 Pipeline 的概念。2.x 时代你要么写 DataStream 代码要么写 Flink SQL都得自己拼 source 和 sink。3.0 提供了flink-cdc.sh加 YAML 配置的方式把 source、sink、pipeline 三段式配置化不用写一行 Java 代码就能跑起一条同步链路。这个变化的意义在于数据同步从开发任务变成了配置任务。以前加一张表的同步要改代码、打包、提交现在改 YAML 重启就行。对于表多、变更频繁的场景运维成本降得很明显。2.3 环境准备与依赖清单动手之前先把环境理清楚。这套流程涉及四个组件MySQL源、Flink计算引擎、Doris目标、HDFSCheckpoint 存储。版本上Flink 用 1.18.0Flink CDC 用 3.0.0MySQL 8.0.31Doris 1.2.4。Maven 依赖里最关键的两个是flink-connector-mysql-cdc3.0.0和mysql-connector-java8.0.31。注意 Flink CDC 3.0 的 DataStream API 包路径已经变成com.ververica.cdc.connectors.mysql.source和 2.x 的com.alibaba.ververica不一样从旧版本迁移的要注意改 import。properties flink-version1.18.0/flink-version /properties dependencies !-- Flink 核心依赖 -- 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 !-- Flink CDC MySQL 连接器 -- dependency groupIdcom.ververica/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.0.0/version /dependency !-- MySQL 驱动 -- dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId version8.0.31/version /dependency /dependencies打包插件用maven-assembly-plugin打 fat jar否则集群上跑会报 ClassNotFound。这个坑后面还会细说。3. DataStream 方式实操从 MySQL 建表到断点续传的完整链路3.1 MySQL 端准备建库建表与 Binlog 开启先在 MySQL 里把测试数据准备好。建两个库test和test_route各建三张表字段简单点方便观察变更。-- 创建数据库 CREATE DATABASE test; CREATE DATABASE test_route; -- 在 test 库建表 USE test; CREATE TABLE t1( id VARCHAR(255) PRIMARY KEY, name VARCHAR(255) ); CREATE TABLE t2( id VARCHAR(255) PRIMARY KEY, name VARCHAR(255) ); CREATE TABLE t3( id VARCHAR(255) PRIMARY KEY, sex VARCHAR(255) ); -- 插入初始数据 INSERT INTO t1 VALUES(1001,zhangsan); INSERT INTO t1 VALUES(1002,lisi); INSERT INTO t1 VALUES(1003,wangwu); INSERT INTO t2 VALUES(1001,zhangsan); INSERT INTO t2 VALUES(1002,lisi); INSERT INTO t2 VALUES(1003,wangwu); INSERT INTO t3 VALUES(1001,F); INSERT INTO t3 VALUES(1002,F); INSERT INTO t3 VALUES(1003,M);test_route库同样建三张表但 t2 表插入不同的数据1004、1005、1006这样后面做路由同步时能看出两个库的数据差异。接下来是 Binlog 配置这一步不做后面全白搭。编辑 MySQL 配置文件sudo vim /etc/my.cnf加入以下配置# 数据库唯一标识集群内不能重复 server-id 1 # 开启 binlog文件名前缀 log-binmysql-bin # binlog 格式必须是 rowstatement 格式拿不到完整行数据 binlog_formatrow # 只捕获指定库的变更按实际业务改 binlog-do-dbtest binlog-do-dbtest_route改完重启 MySQL。注意binlog_format必须是row这是 Flink CDC 能拿到完整变更数据的前提。如果原来是statement或mixed改完要重启才生效。3.2 DataStream 代码Source 构建与 Checkpoint 配置DataStream 方式的核心是构建MySqlSource然后接上 Checkpoint 机制实现断点续传。完整代码如下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; public class FlinkCDCDataStreamTest { public static void main(String[] args) throws Exception { // 1. 准备流处理环境 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. 开启 CheckpointFlink CDC 靠它保存 binlog 位点实现断点续传 // 每 3 秒做一次 CKEXACTLY_ONCE 保证不丢不重 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); // CK 超时 1 分钟 env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); // 两次 CK 之间最小间隔 3 秒 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务取消时保留最后一次 CK方便从 Savepoint 恢复 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 失败重启策略1 天内最多重启 3 次每次间隔 1 分钟 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L) )); // 状态后端用 HashMapStateBackendCK 存到 HDFS env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage(hdfs://hadoop102:8020/flinkCDC); System.setProperty(HADOOP_USER_NAME, atguigu); // 3. 创建 MySQL CDC Source MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(hadoop103) .port(3306) .databaseList(test) // 捕获的数据库 .tableList(test.t1) // 捕获的表不写则捕获整库 .username(root) .password(000000) .deserializer(new JsonDebeziumDeserializationSchema()) // 输出 JSON .startupOptions(StartupOptions.initial()) // 首次全量增量 .build(); // 4. 从 MySQL 读取数据 DataStreamSourceString mysqlDS env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), MySqlSource); // 5. 打印输出 mysqlDS.print(); // 6. 执行任务 env.execute(); } }几个关键参数说明。startupOptions决定启动时的读取策略initial()是先做全量快照再读增量 Binlog适合首次同步latest()只读启动后的变更适合已经同步过、不想重跑全量的场景earliest()从 Binlog 开头读慎用可能重复消费大量历史数据timestamp()和specificOffset()分别按时间戳和指定位点启动用于精确恢复。databaseList和tableList支持正则比如test.\.*表示 test 库下所有表。注意tableList的格式是库名.表名不是只写表名。3.3 打包提交与 Savepoint 恢复验证代码写完打包上传到 Linux然后按顺序启动依赖服务# 启动 HDFSCheckpoint 存储 start-dfs.sh # 启动 Flink 集群 bin/start-cluster.sh # 提交任务 bin/flink run -m hadoop102:8081 -c com.atguigu.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar任务跑起来后TaskManager 日志会先打印 t1 表的全量数据然后进入增量监听状态。这时候在 MySQL 里对 t1 做增删改日志里会实时输出对应的 JSON 变更记录。验证断点续传的步骤先给当前任务创建 Savepoint然后在 WebUI 里 Cancel Job接着往 MySQL 里插几条新数据最后从 Savepoint 重启任务。# 创建 Savepoint bin/flink savepoint JobId hdfs://hadoop102:8020/flinkCDC/save # 从 Savepoint 恢复 bin/flink run -s hdfs://hadoop102:8020/flinkCDC/save/savepoint-xxxxx \ -c com.atguigu.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar恢复后观察日志任务会从 Savepoint 记录的 Binlog 位点继续消费Cancel 期间产生的变更也不会丢。这就是 Checkpoint 机制的价值——没有它任务一挂就得从头再来。4. Flink SQL 与 Pipeline 模式两条更省事的同步路径4.1 Flink SQL 方式DDL 建表即同步如果不想写 Java 代码Flink SQL 方式更直接。核心是用CREATE TABLE语句声明 MySQL CDC 源表然后SELECT查询就自动变成流式同步。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) { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); StreamTableEnvironment tableEnv StreamTableEnvironment.create(env); // 用 DDL 声明 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 )); // 查询即同步 Table table tableEnv.sqlQuery(select * from t1); table.execute().print(); } }注意primary key NOT ENFORCED这个写法Flink SQL 里主键约束不强制校验但 CDC 连接器需要它来识别主键列。connector参数固定写mysql-cdc其余参数和 DataStream 方式一一对应。这种方式适合快速验证和简单同步复杂转换逻辑还是得回到 DataStream。4.2 Pipeline 模式YAML 配置驱动的整库同步Flink CDC 3.0 最省事的是 Pipeline 模式。解压flink-cdc-3.0.0-bin.tar.gz把 MySQL 和 Doris 的连接器 jar 丢进 lib 目录然后写一个 YAML 文件就能跑。source: type: mysql hostname: node01 port: 3306 username: root password: 123 tables: test.\.* # 正则匹配 test 库所有表 server-id: 5400-5404 # 伪装成从库的 server-id 范围 server-time-zone: UTC8 sink: type: doris fenodes: node01:7030 # Doris FE 地址 username: root password: 000000 table.create.properties.light_schema_change: true # 支持轻量级 schema 变更 table.create.properties.replication_num: 1 # 副本数测试环境设 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1tables字段用正则test.\.*匹配整个库加一张表不用改配置。server-id给一个范围Flink CDC 会从中分配避免和 MySQL 主从集群冲突。light_schema_change开启后源表加列会自动同步到 Doris不用手动改目标表结构。启动命令就一行bin/flink-cdc.sh job/mysql-to-doris.yaml跑之前确认两件事Flink 配置里execution.checkpointing.interval已设置比如 5000Doris 的 FE 和 BE 都已启动。然后在 Doris 里建好对应的库任务启动后刷新就能看到数据同步过来。4.3 路由变更一份配置同步多个库test_route库的存在就是为了演示路由功能。当需要把多个源库同步到 Doris 的不同库或者做库名/表名映射时在 YAML 里加 route 配置段即可。核心思路是 source 端用正则匹配多个库sink 端通过 route 规则把test_route.t1映射到 Doris 的指定库表。这样一套配置能管多个库的同步不用为每个库单独写 job。5. 避坑排查那些让我重启过集群的配置细节5.1 Binlog 没开或格式不对任务启动直接报错现象任务提交后立刻失败日志报The MySQL server is not configured to use a row level based binary log。原因MySQL 的binlog_format不是row或者log-bin根本没开。解决检查/etc/my.cnf里log-bin和binlog_formatrow两项改完必须重启 MySQL。另外binlog-do-db如果配了要确认目标库在列表里否则 Binlog 里压根没有这个库的变更。5.2 Checkpoint 没配断点续传变成从头再来现象任务失败重启后之前同步过的数据又全量跑了一遍Doris 里出现重复数据。原因没有开启 Checkpoint或者 Checkpoint 存储路径不可写。Flink CDC 靠 Checkpoint 保存 Binlog 位点没有它就没有断点。解决代码里env.enableCheckpointing()必须调用setCheckpointStorage指向的 HDFS 路径要确保有写权限。用HADOOP_USER_NAME指定用户是个常见做法但生产环境建议配好 Kerberos 或直接用有权限的账号。5.3 fat jar 没打全集群上 ClassNotFound现象本地 IDE 跑得好好的提交到 Flink 集群报ClassNotFoundException: com.ververica.cdc.connectors.mysql.source.MySqlSource。原因依赖没有打进 jar 包集群的 lib 目录里也没有 CDC 连接器。解决Maven 里配maven-assembly-plugin打 fat jar或者把flink-connector-mysql-cdc的 jar 手动放到 Flink 的 lib 目录。Pipeline 模式则是把连接器 jar 放到flink-cdc-3.0.0/lib下别放错位置。5.4 server-id 冲突导致 Binlog 拉取失败现象任务运行一段时间后报A slave with the same server_uuid/server_id as this slave has connected to the master。原因Flink CDC 伪装成从库读 Binlog需要唯一的 server-id。多个 CDC 任务用了相同的 id或者和 MySQL 主从集群的 id 撞了。解决在 source 配置里给server-id指定一个范围如5400-5404Flink CDC 会自动分配。多个任务之间范围不要重叠。5.5 Doris 副本数设太高测试环境写入超时现象同步任务卡在写入阶段Doris 报Failed to create table with replication_num3。原因测试环境 BE 节点不够默认副本数 3 建表失败。解决YAML 里加table.create.properties.replication_num: 1测试环境单副本足够。生产环境按实际 BE 数量设置。6. 进阶技巧让同步链路更稳的几个习惯跑通基础流程之后有几个习惯能显著降低线上翻车概率。第一启动模式按场景选。首次同步用initial()全量加增量一次搞定已经同步过的任务重启用latest()或从 Savepoint 恢复避免重复消费。我见过有人不管什么场景都用initial()结果每次重启都全量重跑Doris 里数据翻了好几倍。第二并行度别盲目调大。Pipeline 模式里parallelism控制同步并行度但源库 Binlog 拉取是单线程的调太大反而增加调度开销。一般按表数量和数据量来表少的话 1 到 2 就够。第三监控 Checkpoint 成功率。Flink WebUI 里 Checkpoint 页面如果频繁失败说明状态后端或网络有问题这时候同步链路随时可能断。建议配告警Checkpoint 连续失败 3 次就介入排查。第四Schema 变更要提前验证。源表加列、改类型在业务迭代中很常见light_schema_change虽然能自动同步但类型映射不一定符合预期。我的习惯是每次 DDL 变更后手动去 Doris 里DESC一下目标表确认字段类型对得上。第五Savepoint 是后悔药但别滥用。Savepoint 能精确恢复位点但创建和恢复都有开销。日常重启用 Checkpoint 自动恢复就行只有版本升级、逻辑大改这种场景才手动打 Savepoint。从那以后我每次上线同步任务前都会强制走一遍Binlog 检查 → Checkpoint 验证 → Savepoint 恢复测试三步确认链路真的能断点续传再交给业务用。希望帮到你。本文还有配套的精品资源点击获取
简介这份资源是尚硅谷研究院编写的Flink CDC 3.0技术文档面向有一定Flink基础、从事大数据处理与实时数据同步的研发工程师帮助读者掌握变更数据捕获原理并落地MySQL到Doris的Streaming ETL链路。文档从CDC概念切入对比基于查询与基于Binlog两种方式的差异讲解flink-cdc-connectors组件的使用并给出数据准备、DataStream与Flink SQL两种实现路径以及MySQL同步至Doris的完整实操涵盖环境搭建、配置文件编写、任务启动与测试同时强调开启Binlog、设置检查点等关键配置。资源包为1个docx文件约145KB内容紧凑、代码示例与操作指南齐全便于快速上手并应用于实际项目。目前已有1788人学习适合希望系统理解Flink CDC并完成稳定数据同步实践的读者参考。1. 从一次凌晨三点的对账说起Flink CDC 3.0 到底解决了什么凌晨三点被叫起来对账MySQL 里订单状态已经改成已退款Doris 报表里还挂着已支付运营那边催着出数。这种场景做实时数仓的同行应该都不陌生——业务库和 OLAP 库之间的数据延迟轻则报表不准重则影响决策。Flink CDC 3.0 就是冲着这类问题来的它把 MySQL 的 Binlog 变更实时捕获出来经过 Flink 做流式 ETL再同步到 Doris 这类分析型数据库整条链路延迟能压到秒级。这份资源是一套完整的 Flink CDC 3.0 实战教程从 CDC 概念讲起覆盖 DataStream API、Flink SQL 两种开发方式最后落到 MySQL 到 Doris 的 Streaming ETL 全流程。适合有 Flink 基础、正在做实时数仓或数据同步的工程师也适合想从 Sqoop/DataX 批同步迁移到流式同步的团队。下面按原理选型 → 动手复现 → 踩坑排查 → 进阶技巧的顺序拆开讲。2. CDC 选型为什么基于 Binlog 的方案才是实时同步的正解2.1 基于查询 vs 基于 Binlog 的本质差异CDCChange Data Capture的核心是捕获数据库的变更。市面上主流方案分两类基于查询的Sqoop、DataX和基于 Binlog 的Canal、Maxwell、Debezium、Flink CDC。这两者的差异不是快一点慢一点而是执行模型根本不同。基于查询的方案本质是批处理靠定时轮询全表或增量字段来拉数据。它有两个硬伤一是无法捕获 DELETE 操作因为被删的行查不到了二是每次查询都要扫表对源库压力大。基于 Binlog 的方案则是流式的MySQL 把每次数据变更按顺序写进 Binlog 文件CDC 工具伪装成从库去订阅这个日志流插入、更新、删除全都能拿到而且对源库几乎无额外压力。对比维度基于查询Sqoop/DataX基于 BinlogFlink CDC执行模式BatchStreaming能否捕获 DELETE否是变化延迟高分钟到小时级低秒级对源库压力大全表扫描小读 Binlog断点续传依赖增量字段依赖 Checkpoint/Savepoint选型结论很直接只要业务对实时性有要求或者需要完整捕获删除操作就必须走 Binlog 路线。Flink CDC 在 Debezium 基础上做了 Flink 原生集成省掉了 Kafka 中间层链路更短。2.2 Flink CDC 3.0 的架构变化Flink CDC 3.0 相比 2.x 最大的变化是引入了 Pipeline 的概念。2.x 时代你要么写 DataStream 代码要么写 Flink SQL都得自己拼 source 和 sink。3.0 提供了flink-cdc.sh加 YAML 配置的方式把 source、sink、pipeline 三段式配置化不用写一行 Java 代码就能跑起一条同步链路。这个变化的意义在于数据同步从开发任务变成了配置任务。以前加一张表的同步要改代码、打包、提交现在改 YAML 重启就行。对于表多、变更频繁的场景运维成本降得很明显。2.3 环境准备与依赖清单动手之前先把环境理清楚。这套流程涉及四个组件MySQL源、Flink计算引擎、Doris目标、HDFSCheckpoint 存储。版本上Flink 用 1.18.0Flink CDC 用 3.0.0MySQL 8.0.31Doris 1.2.4。Maven 依赖里最关键的两个是flink-connector-mysql-cdc3.0.0和mysql-connector-java8.0.31。注意 Flink CDC 3.0 的 DataStream API 包路径已经变成com.ververica.cdc.connectors.mysql.source和 2.x 的com.alibaba.ververica不一样从旧版本迁移的要注意改 import。properties flink-version1.18.0/flink-version /properties dependencies !-- Flink 核心依赖 -- 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 !-- Flink CDC MySQL 连接器 -- dependency groupIdcom.ververica/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.0.0/version /dependency !-- MySQL 驱动 -- dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId version8.0.31/version /dependency /dependencies打包插件用maven-assembly-plugin打 fat jar否则集群上跑会报 ClassNotFound。这个坑后面还会细说。3. DataStream 方式实操从 MySQL 建表到断点续传的完整链路3.1 MySQL 端准备建库建表与 Binlog 开启先在 MySQL 里把测试数据准备好。建两个库test和test_route各建三张表字段简单点方便观察变更。-- 创建数据库 CREATE DATABASE test; CREATE DATABASE test_route; -- 在 test 库建表 USE test; CREATE TABLE t1( id VARCHAR(255) PRIMARY KEY, name VARCHAR(255) ); CREATE TABLE t2( id VARCHAR(255) PRIMARY KEY, name VARCHAR(255) ); CREATE TABLE t3( id VARCHAR(255) PRIMARY KEY, sex VARCHAR(255) ); -- 插入初始数据 INSERT INTO t1 VALUES(1001,zhangsan); INSERT INTO t1 VALUES(1002,lisi); INSERT INTO t1 VALUES(1003,wangwu); INSERT INTO t2 VALUES(1001,zhangsan); INSERT INTO t2 VALUES(1002,lisi); INSERT INTO t2 VALUES(1003,wangwu); INSERT INTO t3 VALUES(1001,F); INSERT INTO t3 VALUES(1002,F); INSERT INTO t3 VALUES(1003,M);test_route库同样建三张表但 t2 表插入不同的数据1004、1005、1006这样后面做路由同步时能看出两个库的数据差异。接下来是 Binlog 配置这一步不做后面全白搭。编辑 MySQL 配置文件sudo vim /etc/my.cnf加入以下配置# 数据库唯一标识集群内不能重复 server-id 1 # 开启 binlog文件名前缀 log-binmysql-bin # binlog 格式必须是 rowstatement 格式拿不到完整行数据 binlog_formatrow # 只捕获指定库的变更按实际业务改 binlog-do-dbtest binlog-do-dbtest_route改完重启 MySQL。注意binlog_format必须是row这是 Flink CDC 能拿到完整变更数据的前提。如果原来是statement或mixed改完要重启才生效。3.2 DataStream 代码Source 构建与 Checkpoint 配置DataStream 方式的核心是构建MySqlSource然后接上 Checkpoint 机制实现断点续传。完整代码如下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; public class FlinkCDCDataStreamTest { public static void main(String[] args) throws Exception { // 1. 准备流处理环境 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. 开启 CheckpointFlink CDC 靠它保存 binlog 位点实现断点续传 // 每 3 秒做一次 CKEXACTLY_ONCE 保证不丢不重 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); // CK 超时 1 分钟 env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); // 两次 CK 之间最小间隔 3 秒 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务取消时保留最后一次 CK方便从 Savepoint 恢复 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 失败重启策略1 天内最多重启 3 次每次间隔 1 分钟 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L) )); // 状态后端用 HashMapStateBackendCK 存到 HDFS env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage(hdfs://hadoop102:8020/flinkCDC); System.setProperty(HADOOP_USER_NAME, atguigu); // 3. 创建 MySQL CDC Source MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(hadoop103) .port(3306) .databaseList(test) // 捕获的数据库 .tableList(test.t1) // 捕获的表不写则捕获整库 .username(root) .password(000000) .deserializer(new JsonDebeziumDeserializationSchema()) // 输出 JSON .startupOptions(StartupOptions.initial()) // 首次全量增量 .build(); // 4. 从 MySQL 读取数据 DataStreamSourceString mysqlDS env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), MySqlSource); // 5. 打印输出 mysqlDS.print(); // 6. 执行任务 env.execute(); } }几个关键参数说明。startupOptions决定启动时的读取策略initial()是先做全量快照再读增量 Binlog适合首次同步latest()只读启动后的变更适合已经同步过、不想重跑全量的场景earliest()从 Binlog 开头读慎用可能重复消费大量历史数据timestamp()和specificOffset()分别按时间戳和指定位点启动用于精确恢复。databaseList和tableList支持正则比如test.\.*表示 test 库下所有表。注意tableList的格式是库名.表名不是只写表名。3.3 打包提交与 Savepoint 恢复验证代码写完打包上传到 Linux然后按顺序启动依赖服务# 启动 HDFSCheckpoint 存储 start-dfs.sh # 启动 Flink 集群 bin/start-cluster.sh # 提交任务 bin/flink run -m hadoop102:8081 -c com.atguigu.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar任务跑起来后TaskManager 日志会先打印 t1 表的全量数据然后进入增量监听状态。这时候在 MySQL 里对 t1 做增删改日志里会实时输出对应的 JSON 变更记录。验证断点续传的步骤先给当前任务创建 Savepoint然后在 WebUI 里 Cancel Job接着往 MySQL 里插几条新数据最后从 Savepoint 重启任务。# 创建 Savepoint bin/flink savepoint JobId hdfs://hadoop102:8020/flinkCDC/save # 从 Savepoint 恢复 bin/flink run -s hdfs://hadoop102:8020/flinkCDC/save/savepoint-xxxxx \ -c com.atguigu.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar恢复后观察日志任务会从 Savepoint 记录的 Binlog 位点继续消费Cancel 期间产生的变更也不会丢。这就是 Checkpoint 机制的价值——没有它任务一挂就得从头再来。4. Flink SQL 与 Pipeline 模式两条更省事的同步路径4.1 Flink SQL 方式DDL 建表即同步如果不想写 Java 代码Flink SQL 方式更直接。核心是用CREATE TABLE语句声明 MySQL CDC 源表然后SELECT查询就自动变成流式同步。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) { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); StreamTableEnvironment tableEnv StreamTableEnvironment.create(env); // 用 DDL 声明 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 )); // 查询即同步 Table table tableEnv.sqlQuery(select * from t1); table.execute().print(); } }注意primary key NOT ENFORCED这个写法Flink SQL 里主键约束不强制校验但 CDC 连接器需要它来识别主键列。connector参数固定写mysql-cdc其余参数和 DataStream 方式一一对应。这种方式适合快速验证和简单同步复杂转换逻辑还是得回到 DataStream。4.2 Pipeline 模式YAML 配置驱动的整库同步Flink CDC 3.0 最省事的是 Pipeline 模式。解压flink-cdc-3.0.0-bin.tar.gz把 MySQL 和 Doris 的连接器 jar 丢进 lib 目录然后写一个 YAML 文件就能跑。source: type: mysql hostname: node01 port: 3306 username: root password: 123 tables: test.\.* # 正则匹配 test 库所有表 server-id: 5400-5404 # 伪装成从库的 server-id 范围 server-time-zone: UTC8 sink: type: doris fenodes: node01:7030 # Doris FE 地址 username: root password: 000000 table.create.properties.light_schema_change: true # 支持轻量级 schema 变更 table.create.properties.replication_num: 1 # 副本数测试环境设 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1tables字段用正则test.\.*匹配整个库加一张表不用改配置。server-id给一个范围Flink CDC 会从中分配避免和 MySQL 主从集群冲突。light_schema_change开启后源表加列会自动同步到 Doris不用手动改目标表结构。启动命令就一行bin/flink-cdc.sh job/mysql-to-doris.yaml跑之前确认两件事Flink 配置里execution.checkpointing.interval已设置比如 5000Doris 的 FE 和 BE 都已启动。然后在 Doris 里建好对应的库任务启动后刷新就能看到数据同步过来。4.3 路由变更一份配置同步多个库test_route库的存在就是为了演示路由功能。当需要把多个源库同步到 Doris 的不同库或者做库名/表名映射时在 YAML 里加 route 配置段即可。核心思路是 source 端用正则匹配多个库sink 端通过 route 规则把test_route.t1映射到 Doris 的指定库表。这样一套配置能管多个库的同步不用为每个库单独写 job。5. 避坑排查那些让我重启过集群的配置细节5.1 Binlog 没开或格式不对任务启动直接报错现象任务提交后立刻失败日志报The MySQL server is not configured to use a row level based binary log。原因MySQL 的binlog_format不是row或者log-bin根本没开。解决检查/etc/my.cnf里log-bin和binlog_formatrow两项改完必须重启 MySQL。另外binlog-do-db如果配了要确认目标库在列表里否则 Binlog 里压根没有这个库的变更。5.2 Checkpoint 没配断点续传变成从头再来现象任务失败重启后之前同步过的数据又全量跑了一遍Doris 里出现重复数据。原因没有开启 Checkpoint或者 Checkpoint 存储路径不可写。Flink CDC 靠 Checkpoint 保存 Binlog 位点没有它就没有断点。解决代码里env.enableCheckpointing()必须调用setCheckpointStorage指向的 HDFS 路径要确保有写权限。用HADOOP_USER_NAME指定用户是个常见做法但生产环境建议配好 Kerberos 或直接用有权限的账号。5.3 fat jar 没打全集群上 ClassNotFound现象本地 IDE 跑得好好的提交到 Flink 集群报ClassNotFoundException: com.ververica.cdc.connectors.mysql.source.MySqlSource。原因依赖没有打进 jar 包集群的 lib 目录里也没有 CDC 连接器。解决Maven 里配maven-assembly-plugin打 fat jar或者把flink-connector-mysql-cdc的 jar 手动放到 Flink 的 lib 目录。Pipeline 模式则是把连接器 jar 放到flink-cdc-3.0.0/lib下别放错位置。5.4 server-id 冲突导致 Binlog 拉取失败现象任务运行一段时间后报A slave with the same server_uuid/server_id as this slave has connected to the master。原因Flink CDC 伪装成从库读 Binlog需要唯一的 server-id。多个 CDC 任务用了相同的 id或者和 MySQL 主从集群的 id 撞了。解决在 source 配置里给server-id指定一个范围如5400-5404Flink CDC 会自动分配。多个任务之间范围不要重叠。5.5 Doris 副本数设太高测试环境写入超时现象同步任务卡在写入阶段Doris 报Failed to create table with replication_num3。原因测试环境 BE 节点不够默认副本数 3 建表失败。解决YAML 里加table.create.properties.replication_num: 1测试环境单副本足够。生产环境按实际 BE 数量设置。6. 进阶技巧让同步链路更稳的几个习惯跑通基础流程之后有几个习惯能显著降低线上翻车概率。第一启动模式按场景选。首次同步用initial()全量加增量一次搞定已经同步过的任务重启用latest()或从 Savepoint 恢复避免重复消费。我见过有人不管什么场景都用initial()结果每次重启都全量重跑Doris 里数据翻了好几倍。第二并行度别盲目调大。Pipeline 模式里parallelism控制同步并行度但源库 Binlog 拉取是单线程的调太大反而增加调度开销。一般按表数量和数据量来表少的话 1 到 2 就够。第三监控 Checkpoint 成功率。Flink WebUI 里 Checkpoint 页面如果频繁失败说明状态后端或网络有问题这时候同步链路随时可能断。建议配告警Checkpoint 连续失败 3 次就介入排查。第四Schema 变更要提前验证。源表加列、改类型在业务迭代中很常见light_schema_change虽然能自动同步但类型映射不一定符合预期。我的习惯是每次 DDL 变更后手动去 Doris 里DESC一下目标表确认字段类型对得上。第五Savepoint 是后悔药但别滥用。Savepoint 能精确恢复位点但创建和恢复都有开销。日常重启用 Checkpoint 自动恢复就行只有版本升级、逻辑大改这种场景才手动打 Savepoint。从那以后我每次上线同步任务前都会强制走一遍Binlog 检查 → Checkpoint 验证 → Savepoint 恢复测试三步确认链路真的能断点续传再交给业务用。希望帮到你。本文还有配套的精品资源点击获取
