简介面向具备一定 Flink 基础、希望掌握实时数据同步链路的研发人员这份由尚硅谷研究院编写的 Flink CDC 3.0 文档以 MySQL 变更捕获与写入 Doris 为主线系统梳理 CDC 概念、基于查询与基于 Binlog 两类方案的差异以及 flink-cdc-connectors 组件的使用方式。资源包共 1 个 docx 文件约 145KB内容集中为可跟随操作的技术文档涵盖数据准备、DataStream 与 Flink SQL 两种实现路径并给出 MySQL 到 Doris 的 Streaming ETL 环境搭建、配置文件编写、任务启动与测试步骤。已有 1789 人学习适合用作项目落地前的方案比对和实操参照。读者可从中获得开启 Binlog、设置检查点等关键配置要点理解全量加增量读取的流程并借助完整示例快速搭建从源库到目标库的同步任务减少联调与排错成本。1. 版本先对齐Flink CDC 3.0 这条链路装错一步就白跑把一个 Flink CDC 3.0 的作业从 MySQL 实时同步到 Doris听上去是写个 yaml 敲一行bin/flink-cdc.sh的事真动手才发现是四五个组件互相限制版本Flink 1.18.0 的 classpath 里必须能找到flink-connector-baseflink-connector-mysql-cdc3.0.0 的坐标一旦和 Flink 版本错开作业会在提交阶段就抛 NoSuchMethodMySQL 8.0.31 的binlog_format不是 ROWsource 初始化直接失败Doris 1.2.4 的 FE、BE 没起全sink 建表也会卡住。我见过最多的失败情形是 MySQL 的log-bin根本没开程序跑了三十秒退出日志里只有一行含糊的 connect 异常让人对着配置文件反复怀疑自己。这套环境适合已经能跑 Flink 作业、准备把业务库变更接到 Doris 做实时报表的开发者下面从装什么版本一路写到装坏了怎么救。2. 环境清单与 MySQL 侧准备Flink 1.18.0、CDC 3.0.0、MySQL 8.0.31 的版本对齐这一章先把“装什么、装在哪、版本怎么搭”确定下来。装环境最怕的不是步骤多而是装完才发现某个组件的版本天生不适配回头重装一遍之前配好的 checkpoint 目录、Doris 端口全要重来。2.1 四个组件的版本对应关系这套链路里真正会互相制约的是 Flink 和 CDC 连接器MySQL 与 Doris 更多是配合的角色。CDC 3.0.0 这一版把 DataStream、FlinkSQL、Pipeline 三种用法统一收进一个发行包Pipeline 模式依赖 Flink 1.18 的运行时所以 Flink 侧我一般固定在 1.18.0Scala 二进制选 2.12别用 2.13 的包去配_2.12后缀的 planner类加载时找不到类又不报清楚的原因。组件版本在链路里的角色装错的典型表现JDK8编译与运行Flink 1.18 起对 JDK 17 有额外参数要求先用 8Flink1.18.0scala_2.12作业运行时与 CDC 3.0 的 base 包版本不一致提交即失败flink-connector-mysql-cdc3.0.0读 MySQL 全量与增量低版本没有 Pipeline 相关类mysql-connector-java8.0.31JDBC 驱动用 5.1.x 连 MySQL 8 报认证插件错误MySQL8.0.31数据源binlog 未开source 初始化失败Doris1.2.4目标库FE/BE 未起全建表卡住或副本不足表里最后两行是很多人忽略的MySQL 和 Doris 看起来“本来就在跑”但 Flink CDC 对它们是有要求的。MySQL 必须是 ROW 格式的 binlogDoris 必须至少有一个 BE 存活并且副本数不超过 BE 数量。2.2 Flink 1.18.0 与 Doris 1.2.4 的最小安装改动Flink 解压完不要急着提交作业Pipeline 模式靠 checkpoint 记录 binlog 位点没有 checkpoint 配置任务重启就是从头发一遍或者干脆不落位点。把下面几行追加到conf/flink-conf.yaml分发到集群其它机器后再启动。tar -zxf flink-1.18.0-bin-scala_2.12.tgz -C /opt/module/ cd /opt/module/flink-1.18.0 cat conf/flink-conf.yaml EOF # Pipeline 模式必须打开间隔按业务量调5 秒是起步值 execution.checkpointing.interval: 5000 # 状态后端与 checkpoint 落盘位置放 HDFS 才能扛住 TaskManager 挂掉 state.backend: hashmap state.checkpoints.dir: hdfs://node101:8020/flink-cdc/checkpoints EOF bin/start-cluster.shexecution.checkpointing.interval的单位是毫秒配 5000 表示 5 秒一次state.checkpoints.dir写在配置里比写在代码里更省事因为 Pipeline 模式的作业根本走不到你的代码。这一段最容易忽略的是 HDFS 路径的可写权限后面避坑章会专门讲。Doris 侧按 FE、BE 顺序起起来之后用 MySQL 客户端连查询端口确认后端节点注册状态。cd /opt/module/doris/fe bin/start_fe.sh --daemon cd /opt/module/doris/be bin/start_be.sh --daemon # 9030 是 Doris 的查询端口7030 是 FE 的 HTTP 端口别混 mysql -uroot -p000000 -P9030 -hnode101 -e show backends\Gshow backends里的Alive必须是 trueAvailCapacity不能是 0。这一步没过就往下走后面 sink 建表一定会报副本数不满足报错信息还很长容易误判成权限问题。2.3 MySQL 建库建表与 Binlog 参数测试数据按教程里的结构建就够验证全链路主库test放t1、t2、t3三张表另一套test_route放同名表用来验证路由。表结构简单到只有主键加两个字段正是为了排除字段类型带来的干扰。-- 主库模拟业务库 CREATE DATABASE IF NOT EXISTS 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)); -- 第二套库后面验证路由和多库同步 CREATE DATABASE IF NOT EXISTS test_route; USE test_route; 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 test.t1 VALUES (1001,zhangsan),(1002,lisi),(1003,wangwu); INSERT INTO test.t3 VALUES (1001,F),(1002,F),(1003,M);建表时主键不能省CDC 的 update、delete 事件要靠主键定位目标行没有主键的表在 Doris 侧也建不出合适的模型。真正让新手卡住的是 binlog。默认安装的 MySQL 8 是不开 binlog 的CDC 读的是 binlog 而不是查询表所以这一段必须配。改my.cnf加四行然后重启 MySQL。[mysqld] # 集群内唯一多个 CDC 作业或主从复制时重复会直接握手失败 server-id 1 # 开启 binlog值作为文件前缀 log-bin mysql-bin # Flink CDC 只认 ROWSTATEMENT/MIXED 都会报错 binlog_format row # 只捕获这两个库生产上按实际库名收窄 binlog-do-db test binlog-do-db test_routebinlog-do-db是按“当前默认库”过滤的用错库名执行更新事件不会写进 binlog表现就是“我明明改了数据Doris 没变”。这类问题没有日志告警只能靠自己记住这条规则。2.4 三行 SQL 确认 binlog 真的开了重启 MySQL 之后别相信配置文件写对了用 SQL 回读变量。SHOW VARIABLES LIKE log_bin; -- 期望 ON SHOW VARIABLES LIKE binlog_format; -- 期望 ROW SHOW MASTER STATUS; -- File、Position 都要有值SHOW MASTER STATUS返回空结果说明 binlog 没真正生效八成是my.cnf放错位置有的发行版读mysqld.cnf或者没重启服务。这三条命令建议写进自己的检查清单后面所有疑难杂症都从这三条开始排查。3. DataStream 方式接入pom 依赖、MySqlSource 参数与 Checkpoint 断点续传DataStream 写法是最能把原理看清的一种全量快照怎么切、位点怎么存、重启怎么续全部在代码里摊开。生产上如果只是同步一两张表到 Doris用后面的 Pipeline 模式更省事但只要涉及自定义加工最后还是回到 DataStream。3.1 pom 里必须出现的依赖与作用域依赖写错的表现通常是ClassNotFoundException或NoSuchMethodError两者都很费时间。原则是Flink 自带的东西打providedCDC 自己的东西和 JDBC 驱动打进去。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 !-- Flink 运行时集群上已有provided 避免和集群 jar 打架 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink-version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink-version}/version scopeprovided/scope /dependency !-- 后缀 _2.12 必须和 Flink 发行包的 scala 版本一致 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-table-planner_2.12/artifactId version${flink-version}/version scopeprovided/scope /dependency !-- CDC source 依赖的桥接包漏了会在 Source 构建时报找不到工厂 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-base/artifactId version${flink-version}/version /dependency !-- CDC 本体3.0.0 才带 Pipeline 相关类 -- dependency groupIdcom.ververica/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.0.0/version /dependency !-- MySQL 8 必须用 8.x 驱动5.1.x 会报认证插件不支持 -- dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId version8.0.31/version /dependency /dependencies打provided这件事我踩过坑早期图省事把flink-streaming-java一起打进 fat jar结果集群上出现两份 Flink 类作业能提交运行到算子链初始化时抛IllegalStateException日志里根本看不出是 jar 冲突。打包插件用maven-assembly-plugin的jar-with-dependenciesphase绑到package出一条带依赖的 jar 就够。3.2 MySqlSource 构建参数逐项拆解Builder 里每一个参数都会影响行为尤其是startupOptions选错就是“为什么我的作业只读增量、历史数据一条没有”。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; MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(node103) // MySQL 地址别写 localhost .port(3306) .databaseList(test) // 捕获的库和 binlog-do-db 保持一致 .tableList(test.t1) // 库名.表名支持 test\\.t.* 正则 .username(root) .password(000000) // 反序列化成 JSON方便 print 和环境验证时肉眼看 .deserializer(new JsonDebeziumDeserializationSchema()) // initial先全量快照再接着读最新 binlog .startupOptions(StartupOptions.initial()) .build();参数取值说明databaseListtest必须被 MySQL 的 binlog 过滤规则包含tableListtest.t1支持test\\.t.*正则写错表名不报错但读不到startupOptionsinitial首次启动做全量快照后转增量startupOptionslatest只读启动之后的变更没有历史数据startupOptionsearliest从 binlog 头读要求历史 binlog 未被清理deserializerJsonDebeziumDeserializationSchema输出含 before/after/op 的 JSONearliest是新手最容易误用的如果 MySQL 的 binlog 已经被expire_logs_days清掉作业从第一天就绪不了报错信息非常隐晦。我一般先用initial把全量收上来后续扩容重跑再换latest。3.3 Checkpoint 与状态后端断点续传的唯一可靠方式CDC 读取的 binlog 位点是存在状态里的没有 checkpoint 就没有断点续传作业重启等于从头再来一遍Doris 侧可能重复写入。所以开 checkpoint 不是可选项。StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 单表同步先单并行度排查简单 // 3 秒一次 CKEXACTLY_ONCE 保证位点不丢 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务被 cancel 时保留最后一次 CK这是升级、改配置后的后悔药 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 失败率重启一天内最多 3 次每次间隔 1 分钟 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage(hdfs://node101:8020/flink-cdc); // 访问 HDFS 的身份不设就用提交作业的 Linux 用户多半没写权限 System.setProperty(HADOOP_USER_NAME, deploy);EXACTLY_ONCE在 CDC 场景下的含义是位点和数据一致但 Doris 侧是否幂等还要看表模型CheckpointingMode.AT_LEAST_ONCE更快代价是重启后可能重复几条。RETAIN_ON_CANCELLATION这三个词值得牢记没有它你改完一个参数重启就只能重新全量读一遍。3.4 打包、提交与 Savepoint 恢复本地 run 一遍确认能读到数据再打成 jar 提交到集群。提交时指定 JobManager 地址和入口类jar 后面跟的是程序参数不是 Flink 参数。# 打包产出一个带全部依赖的 jar mvn clean package -DskipTests # 提交到 standalone 集群8081 是 WebUI 端口 bin/flink run -m node101:8081 \ -c com.demo.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test.jar # 记下 JobId 之后先做一次 savepoint 再动数据 bin/flink savepoint JobId hdfs://node101:8020/flink-cdc/save # 带 savepoint 重启-s 后面是这个目录里带 savepoint-xxx 的那一层 bin/flink run -s hdfs://node101:8020/flink-cdc/save/savepoint-xxxxxx \ -c com.demo.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar-s的路径必须指到具体的savepoint-xxxxxx目录而不是它的父目录指错了 Flink 会当成没有状态直接从startupOptions重新开始。验证方式很直接在 MySQL 里 update 一行看 TaskManager 日志里打印的 JSON 是op:u而不是又一轮全量op:r。4. FlinkSQL 与 Pipeline 两种写法把 MySQL 变更落到 Doris 1.2.4DataStream 灵活但代码量大更常见的情况是“整库同步到 Doris”这时候 FlinkSQL 适合做验证CDC 3.0 的 Pipeline 模式适合做上线。4.1 FlinkSQL 建一张 MySQL CDC 表FlinkSQL 的好处是不用管序列化建表带connector mysql-cdc就能查。注意主键要写NOT ENFORCEDFlink 不校验但下游需要它来推 update、delete。StreamTableEnvironment tableEnv StreamTableEnvironment.create(env); tableEnv.executeSql( CREATE TABLE t1 (\n id STRING PRIMARY KEY NOT ENFORCED,\n name STRING\n ) WITH (\n connector mysql-cdc,\n hostname node103,\n port 3306,\n username root,\n password 000000,\n database-name test,\n table-name t1\n )); // 直接查结果会随 MySQL 变更持续更新 tableEnv.sqlQuery(SELECT * FROM t1).execute().print();connector名字是mysql-cdc而不是mysql写错会提示找不到 factory报错信息会列出所有可用 connector扫一眼就能确认 CDC 的 jar 有没有加载上。本地 IDE 跑直接运行 main 就行打包到集群提交方式和 DataStream 完全一样。4.2 Pipeline 模式yaml 配置与依赖 jar 放置CDC 3.0 新增的 Pipeline 模式不需要写代码解压发行包把两个 pipeline connector 的 jar 放进lib目录剩下就是写 yaml。tar -zxf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ cd /opt/module/flink-cdc-3.0.0 # 这两个 jar 不放进来启动会报 connector 无法识别 cp /path/to/flink-cdc-pipeline-connector-mysql-3.0.0.jar lib/ cp /path/to/flink-cdc-pipeline-connector-doris-3.0.0.jar lib/ mkdir -p job cd jobyaml 分三段source描述 MySQLsink描述 Dorispipeline描述作业本身。source: type: mysql hostname: node103 port: 3306 username: root password: 000000 tables: test\.t.* # 正则匹配 t1、t2、t3 server-id: 5400-5404 # 区间形式多并行度时逐个分配 server-time-zone: UTC8 # 和 MySQL 时区一致否则时间字段差 8 小时 sink: type: doris fenodes: node101:7030 # FE 的 HTTP 端口不是 9030 username: root password: 000000 # 源表加列时自动在 Doris 侧做轻量 schema change table.create.properties.light_schema_change: true # 副本数必须小于等于 BE 数量单机测试写 1 table.create.properties.replication_num: 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1fenodes写 8030 还是 7030 取决于 FE 配置很多发行版是 8030教程里是 7030连不上先用show frontends确认 HTTP 端口。server-id用区间是 3.0 的写法MySQL 侧server-id 1是服务端自己的编号两者不冲突但也不能在同一实例上撞车。4.3 启动顺序与路由变更启动顺序有讲究Doris 先就绪Flink 集群次之最后起 CDC 任务。目标库要先建出来Pipeline 不会替你建库只会建表。# 1. Doris 侧建目标库 mysql -uroot -p000000 -P9030 -hnode101 -e create database if not exists test # 2. Flink 集群已启动且开了 checkpoint # 3. 提交同步任务 cd /opt/module/flink-cdc-3.0.0 bin/flink-cdc.sh job/mysql-to-doris.yaml任务起来后去 Doris 里show tables会看到t1、t2、t3自动建出来字段和主键都跟着源表走。这时候在 MySQL 里 insert、update、delete 各来一条刷新 Doris 就能看到变化。路由变更解决的是“源库名和目标库名不一样”的场景比如test_route要落到doris_test_route。在 yaml 里追加一段route用正则把源表映射到目标表。# 接在 source / sink 之后 route: - source-table: test_route\.t.* sink-table: doris_test_route.t.*路由规则是正则替换source-table里的点必须转义成\.否则会被当成任意字符——这个坑非常隐蔽表现是表全被同步到一个乱七八糟的名字下或者干脆没同步。5. 避坑排查五个让任务三十秒就退出的现场下面这些是我在装这套环境时真撞过的按“现象 → 原因 → 解决”记遇到同类问题可以直接对着查。一、启动就报 binlog 相关错误现象作业提交后立即失败日志里有The MySQL server is not configured to use a row level binary log或类似提示。 原因binlog_format不是 ROW或者log-bin压根没开。有的环境是配置文件改了但服务没重启。 解决回到 2.4 节那三条 SQL 回读变量log_bin必须是 ON、binlog_format必须是 ROW。确认后在 MySQL 上执行FLUSH LOGS;再重试别直接改代码怀疑连接器。二、Checkpoint 一直失败作业反复重启现象WebUI 里 checkpoint 全是 FAILED日志出现Permission denied或Filesystem closed。 原因setCheckpointStorage指到 HDFS但提交作业的 Linux 用户在 HDFS 上没有写权限HADOOP_USER_NAME没设或者设成了不存在的用户。 解决在 main 方法最前面加System.setProperty(HADOOP_USER_NAME, deploy)并确认hdfs dfs -ls /flink-cdc能看到这个用户建的目录。临时验证也可以先换成file:///tmp/flink-ck排除 HDFS 因素。三、savepoint 恢复后数据重复或从头读现象带-s重启Doris 里出现重复行或者日志显示又在做全量快照。 原因-s路径指到了父目录Flink 认为没有可用状态或者startupOptions用的是initial无状态启动必然重跑全量。 解决路径要精确到savepoint-xxxxxx这一层。全量阶段本身不幂等Doris 侧如果是 Duplicate 模型重复行会实打实存在这种表结构要改 Unique 模型。四、Doris 建表失败报副本数相关错误现象Pipeline 启动后 sink 阶段报错Doris 里表没建出来。 原因replication_num写成了 3但只有 1 个 BE 存活副本数超过可用 BE 数量时建表直接拒绝。 解决show backends看存活数量测试环境统一写replication_num: 1。生产上先扩 BE 再改副本数不要反过来。五、时间字段差 8 小时现象MySQL 里2024-01-01 10:00:00Doris 里变成02:00:00或者反过来。 原因server-time-zone没配或者配成UTC而 MySQL 是东八区Debezium 按 UTC 解析。 解决yaml 里写server-time-zone: UTC8DataStream 方式则在 builder 上加.serverTimeZone(Asia/Shanghai)。这条不改后面所有按时间做的对账都白做。6. 装完怎么验证全量、增量、断点续传三道关装完能启动不代表同步对了。我通常分三道关验全量快照的行数对不对、增量变更能不能实时追上、断点续传会不会丢或重。三道都过这套环境才算真的装好。第一关验全量。MySQL 侧和 Doris 侧各查一次行数数字必须一致。注意 Doris 用 9030 端口连别用 7030。# 源端 mysql -uroot -p000000 -h127.0.0.1 -e select count(*) from test.t1 # 目标端端口和地址都不同 mysql -uroot -p000000 -P9030 -hnode101 -e select count(*) from test.t1第二关验增量覆盖 insert、update、delete 三种事件加列这种 DDL 也顺手测一下因为light_schema_change是否生效只有加列能看出来。-- 在 MySQL 上依次执行每执行一条就去 Doris 刷新一次 INSERT INTO test.t1 VALUES (1004,zhaoliu); UPDATE test.t1 SET name lisi_new WHERE id 1002; DELETE FROM test.t1 WHERE id 1003; ALTER TABLE test.t1 ADD COLUMN age INT;预期是 Doris 侧先多一行、再改一行、再少一行最后多出一个age列。如果 insert 生效但 delete 不生效八成是 Doris 表模型建成了 Duplicate主键没起作用。第三关验断点续传这一步最能暴露配置问题。做法是造一次 savepoint停掉作业期间在 MySQL 里改几条数据再带 savepoint 重启看新增的几条有没有被补上、老数据有没有被重复写。验证项命令期望结果全量行数两端count(*)数字一致增量 insertMySQL 插一条后查 Doris立刻可见增量 deleteMySQL 删一条后查 Doris对应行消失DDL 加列MySQLALTER TABLEDoris 自动加列且不报错位点续传flink savepoint后重启只补变更不重跑全量checkpointbin/flink list -m node101:8081状态为 RUNNINGCK 有成功记录还有一个容易被忽略的验证点作业跑起来之后把 TaskManager 进程 kill 掉一个看它能不能从 checkpoint 自动恢复。这一步不做等真出故障的时候才发现重启策略配错了损失的是线上数据。从那以后我每次装这套链路不管多急都强制按“只同步一张表、插三条数据、删一条、重启一次”的顺序走一遍跑通了再放大到整库。玄学问题几乎都出在跳过这一步。希望帮到你。本文还有配套的精品资源点击获取