简介这份由尚硅谷研究院编写的Flink CDC 3.0技术文档面向具备一定Flink基础的大数据研发与实时同步工程师围绕MySQL变更数据捕获到Doris的Streaming ETL链路展开。内容先厘清CDC通过监测数据库插入、更新、删除并写入消息中间件供订阅消费的核心思想对比基于查询与基于Binlog两类方案在执行模式、变化捕获能力和延迟上的差异再引入flink-cdc-connectors组件说明其读取MySQL、PostgreSQL全量与增量数据的方式。案例部分覆盖MySQL建库建表与测试数据准备、DataStream和Flink SQL两种实现、MySQL到Doris的环境搭建、配置文件编写、任务启动与测试并提示开启Binlog、设置检查点等稳定性要点配有完整代码示例与操作指南。资源包仅含1个docx大小约145KB结构紧凑便于按章节查阅。已有1789人学习可作为快速上手项目部署、核对关键配置与验证数据一致性的参考材料。1. 装完就报 ClassNotFoundExceptionFlink CDC 3.0 的包结构和 2.x 不是一回事Flink CDC 3.0 把 MySQL 的变更实时同步到 Doris这套组合现在在实时数仓里出现频率很高但装起来有个反直觉的地方binlog 开着、Doris 的 FE 和 BE 都起来了、任务也提交上去了日志里却甩出一个 ClassNotFoundException。这不是玄学是 3.0 把连接器的打包方式和 2.x 换了一整套jar 名字、依赖坐标、启动入口全变了照旧资料装必然翻车。我第一次装就卡在这里jar 名字对不上白折腾一下午。这套环境Flink 1.18.0 CDC 3.0.0 MySQL 8.0.31 Doris 1.2.4装对了半小时能跑通整库同步装错了就是一路报错。下面按我实际装机的顺序从环境清点、Binlog 开启、CDC 解压、yaml 配置一路记到排错验证。2. 环境清单与版本对齐Flink 1.18.0 CDC 3.0.0 MySQL 8.0.31 Doris 1.2.42.1 版本矩阵哪几个号必须锁死这套环境最容易出问题的地方不是装不上是装上了但版本互相不认。Flink 的 table planner、connector-base 和 CDC 连接器之间是强绑定关系差一个小版本就可能抛 NoSuchMethodError。我一般先把这张表列出来装之前对一遍比装完再查日志省事得多。组件版本作用装错会怎样JDK8 或 11运行 Flink 与 CDC11 以上部分反射调用受限Flink1.18.0流处理引擎与 CDC 3.0 不匹配时 Source 初始化失败Flink CDC3.0.0捕获 MySQL 变更2.x 的 jar 名与 3.0 完全不同MySQL8.0.31数据源5.7 以下部分 binlog 字段解析异常Doris1.2.4目标端版本过低不支持 light_schema_changeScala2.12Flink 二进制包后缀与集群已有包不一致会类冲突提示Flink 的二进制包要选带 scala_2.12 的那个Doris 的 connector 在 3.0 里是按 2.12 编译的混用 2.13 会在运行期才暴露问题。CDC 3.0 的定位和 2.x 有本质区别2.x 是一堆独立的 source connector用 DataStream 或 SQL 接入3.0 引入了 pipeline 的概念把 source 和 sink 用一份 yaml 描述由flink-cdc.sh直接提交成一个 Flink 作业。所以它同时存在两套用法——老的 DataStream/FlinkSQL 写法依然能用新的 yaml pipeline 是 3.0 的主推方式两条路依赖的 jar 不是同一批。2.2 机器规划与前置条件三台机器的分工我按下面这个来不用多够跑通就行hadoop102HDFS NameNode、Flink 集群入口、Doris FE BEhadoop103MySQL 8.0.31、Flink TaskManagerhadoop104Flink TaskManager、备用节点前置的东西不多但少一样后面都得回头补。三台机器先做 SSH 免密JDK 统一装到/opt/module/jdk1.8.0_212JAVA_HOME写进/etc/profile.d/my_env.sh。HDFS 必须有因为 CDC 的 Checkpoint 要落到 HDFS 上本地文件系统只适合单机试玩多并行度下位点恢复会出问题。# 三台机器统一配置环境变量 sudo vim /etc/profile.d/my_env.sh # 写入以下内容 export JAVA_HOME/opt/module/jdk1.8.0_212 export HADOOP_HOME/opt/module/hadoop-3.1.3 export FLINK_HOME/opt/module/flink-1.18.0 export PATH$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$FLINK_HOME/bin # 使配置生效 source /etc/profile.d/my_env.sh这段脚本里FLINK_HOME是关键后面flink-cdc.sh提交任务时会读它去找 Flink 的 lib 目录和配置。HADOOP_HOME里的sbin也要加进 PATH不然start-dfs.sh得写全路径。我见过有同学只加了 bin 没加 sbin启动 HDFS 时一直提示命令不存在还以为是 Hadoop 没装好。HDFS 起来之后先建好 checkpoint 目录并给写权限# 启动 HDFS start-dfs.sh # 建 checkpoint 根目录 hadoop fs -mkdir -p /flinkCDC # 放开权限避免提交任务的用户和写 HDFS 的用户不一致 hadoop fs -chmod -R 777 /flinkCDC注意checkpoint 目录权限是高频坑点。Flink 提交任务用的用户和 HDFS 的超级用户不是同一个时写 checkpoint 直接 Permission denied任务跑几秒就重启日志还藏在 TaskManager 里不显眼。2.3 开启 MySQL Binlogmy.cnf 里那五行CDC 抓 MySQL 变更靠的是解析 binlog所以 MySQL 这一侧必须先开。改之前先确认 MySQL 的配置文件位置源码包安装一般在/etc/my.cnf用mysql --help | grep my.cnf能看到实际加载顺序。# /etc/my.cnf [mysqld] # 数据库实例 id单机随便给个非 0 值 server-id 1 # 开启 binlog这个名字会成为 binlog 文件前缀 log-binmysql-bin # 必须是 rowDebezium 解析的是行级变更 binlog_formatrow # 只记录这两个库多库时按需增加 binlog-do-dbtest binlog-do-dbtest_route逐项说一下为什么这么写。binlog_formatrow是硬要求statement 和 mixed 模式下 binlog 里记的是 SQL 语句而不是行的前后镜像CDC 拿不到变更列的准确值同步过去的数据会是错的而且这种错不会报错属于最难受的一类。binlog-do-db是白名单配了之后只有这两个库的变更会进 binlog其他库改了也不会同步——所以如果你的业务库后面要加记得回来补这一行然后重启。server-id在 CDC 的 yaml 里还有一个server-id: 5400-5404那个是 CDC 伪装成 MySQL 从库的 id 段两者不能撞。改完重启并验证# 重启 MySQL sudo systemctl restart mysqld # 登录验证三个关键参数 mysql -uroot -p000000 -e SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format; SHOW VARIABLES LIKE server_id; 三个值应该分别是 ON、ROW、1。如果log_bin还是 OFF多半是 my.cnf 里的[mysqld]段没写对或者机器上还有别的配置文件覆盖了它用mysqld --verbose --help | grep -A1 Default options看实际加载了哪些。还有一个权限点容易被跳过CDC 连接 MySQL 的账号需要复制权限root 一般都有但生产上单独建账号的话要补-- 授予复制相关权限CDC 读取 binlog 需要 GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO cdc_user%; FLUSH PRIVILEGES;缺REPLICATION SLAVE的表现是任务能起来但读到 binlog 阶段就报错退出错误信息里带Access denied不太直观。2.4 建库建表与初始数据环境验完把测试数据铺好。建两个库是有讲究的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)); -- 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));三张表不是随便凑数。t1 和 t2 结构一样用来测多表同步t3 的字段和它们不同用来测 schema 变更和字段映射。表里都带主键是因为 CDC 的 upsert 语义依赖主键没主键的表在 Doris 侧建表时会退化成 duplicate 模型重复数据消不掉。初始数据各插三条-- test 库 USE test; INSERT INTO t1 VALUES(1001,zhangsan),(1002,lisi),(1003,wangwu); INSERT INTO t2 VALUES(1001,zhangsan),(1002,lisi),(1003,wangwu); INSERT INTO t3 VALUES(1001,F),(1002,F),(1003,M); -- test_route 库id 故意从 1004 开始方便看路由是否生效 USE test_route; INSERT INTO t1 VALUES(1001,zhangsan),(1002,lisi),(1003,wangwu); INSERT INTO t2 VALUES(1004,zhangsan),(1005,lisi),(1006,wangwu); INSERT INTO t3 VALUES(1001,F),(1002,F),(1003,M);两个库的 t2 数据 id 段不一样同步完成后去 Doris 里对比一眼就能看出哪个库进了哪张表路由配错时不会两边都像对的。3. 安装 Flink CDC 3.0.0解压、依赖包与 Checkpoint 打通3.1 装 Flink 1.18.0 与 CDC 3.0.0 二进制包两个包都要装位置分开Flink 是运行引擎CDC 3.0 的二进制包提供flink-cdc.sh和 pipeline 相关的运行时。# 解压 Flink 到 /opt/module tar -zxvf flink-1.18.0-bin-scala_2.12.tgz -C /opt/module/ # 解压 CDC 3.0.0 tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ # 确认目录结构 ls /opt/module/flink-cdc-3.0.0/bin # 应该能看到 flink-cdc.sh ls /opt/module/flink-1.18.0/bin # 应该能看到 flink、start-cluster.sh这里有个认知差flink-cdc.sh不是 Flink 自带的脚本它属于 CDC 3.0 的发行包作用是读 yaml、拼装 Flink 作业、调flink run提交。所以它必须知道 Flink 装在哪靠的就是前面配的FLINK_HOME。如果执行时报找不到 Flink先回头确认这个变量。两个包都解压完先别急着启动把flink-cdc-3.0.0/lib下自带的 jar 看一眼清单会看到flink-cdc-dist-3.0.0.jar这类核心包——这些不需要手动拷flink-cdc.sh提交时会自动带上。3.2 connector jar 放哪lib 目录决定能不能跑起来3.0 的 pipeline 模式下source 和 sink 的 connector 是独立 jar必须放进 Flink 的 lib 目录否则作业提交时报找不到 factory。这一步是 3.0 和 2.x 最大的操作差异。# 把 MySQL 和 Doris 的 pipeline connector 放进 Flink lib cp /opt/module/flink-cdc-3.0.0/lib/flink-cdc-pipeline-connector-mysql-3.0.0.jar \ $FLINK_HOME/lib/ cp /opt/module/flink-cdc-3.0.0/lib/flink-cdc-pipeline-connector-doris-3.0.0.jar \ $FLINK_HOME/lib/ # MySQL 驱动也要放不然连不上 8.0 的库 cp /opt/module/flink-cdc-3.0.0/lib/mysql-connector-java-8.0.31.jar \ $FLINK_HOME/lib/ # 确认拷进去了 ls $FLINK_HOME/lib | grep -E cdc|mysql三个 jar 缺一不可。少了 connector-mysql 会在解析 yaml 的 source 段时报错少了 connector-doris 会在 sink 段报错少了 MySQL 驱动则是运行期连库失败报No suitable driver。这里就是前面说的那个版本坑2.x 时代大家找的是flink-sql-connector-mysql-cdc-2.x.x.jar3.0 改成了flink-cdc-pipeline-connector-mysql-3.0.0.jar名字不对放进去也不生效。提示如果你同时要跑 DataStream 方式的作业那个用的是 Maven 依赖里的flink-connector-mysql-cdc和 pipeline 的 jar 不是一回事别混着拷会类冲突。3.3 打开 Checkpointflink-conf.yaml 里的一行决定成败CDC 的位点信息是以状态形式存在 Checkpoint 里的不开 Checkpoint作业重启就只能从 binlog 当前位点重新开始中间的数据直接丢。所以这一行是必配项。# $FLINK_HOME/conf/flink-conf.yaml # 每 5 秒做一次 checkpoint execution.checkpointing.interval: 5000 # checkpoint 落到 HDFS和代码里指定的路径保持一致 state.checkpoints.dir: hdfs://hadoop102:8020/flinkCDCexecution.checkpointing.interval这个值不能太大太大意味着故障时最多丢这么长时间的数据也不能太小太小会让 TaskManager 频繁做快照影响吞吐。5 秒是这套环境里比较稳的值。state.checkpoints.dir要和后面代码里设的路径一致否则作业自己写的 checkpoint 和集群配置会对不上。配置改完分发到其他节点然后启动集群# 分发配置 xsync $FLINK_HOME/conf/flink-conf.yaml # 启动 Flink 集群 $FLINK_HOME/bin/start-cluster.sh # 检查进程应该有 StandaloneSessionClusterEntrypoint 和 TaskManagerRunner jpsjps里能看到 StandaloneSessionClusterEntrypoint 就说明 JobManager 起来了浏览器打开 hadoop102:8081 能看到 WebUI 就算这一步过了。3.4 启动顺序HDFS → Flink → Doris FE → Doris BE顺序看着是小事实际上搞反了会浪费很多时间排查。Doris 的 FE 启动后靠 BE 上报存活BE 起来前 FE 是Alive: false你在这种状态下建库会失败Flink 的 checkpoint 目录在 HDFSHDFS 没起来作业提交后 checkpoint 全部失败任务反复重启。# 1. HDFS start-dfs.sh # 2. Flink $FLINK_HOME/bin/start-cluster.sh # 3. Doris FE--daemon 让它后台跑 /opt/module/doris/fe/bin/start_fe.sh --daemon # 4. Doris BE /opt/module/doris/be/bin/start_be.sh --daemon # 5. 验证 BE 是否注册成功 mysql -uroot -p000000 -P9030 -hhadoop102 -e SHOW BACKENDS\G最后那条命令里Alive是 true、Join是 true 才算 Doris 侧准备好了。-P9030是 FE 的查询端口和 MySQL 的 3306 不是一回事第一次用很容易写错。连上之后先把目标库建出来-- Doris 侧建库同步时表会自动建库要提前建好 CREATE DATABASE test; CREATE DATABASE doris_test_route;Doris 侧只建库不建表表结构由 CDC 根据 MySQL 的表自动推导并创建。doris_test_route是给路由变更用的名字故意和 MySQL 的test_route不一样路由是否配对一看就知道。4. 三种接入方式DataStream、FlinkSQL 与 YAML Pipeline 的配置差异4.1 Maven 依赖Flink 1.18.0 与 CDC 3.0.0 的搭配写代码的方式用 Maven 工程pom 里的版本号必须对齐。Flink 那堆包的版本统一用 property 管CDC 单独写 3.0.0。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 核心四个包的版本必须一致 -- 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-runtime/artifactId version${flink-version}/version /dependency !-- CDC source注意坐标和 2.x 不同 -- 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 /dependenciesflink-table-planner_2.12后缀那个_2.12是 Scala 版本必须和集群二进制包的 Scala 版本一致错了两边类加载不起来。打包用 assembly 插件打成带依赖的 fat jarbuild plugins plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-assembly-plugin/artifactId version3.0.0/version configuration descriptorRefs descriptorRefjar-with-dependencies/descriptorRef /descriptorRefs /configuration executions execution idmake-assembly/id phasepackage/phase goals goalsingle/goal /goals /execution /executions /plugin /plugins /build打成 fat jar 有个副作用Flink 自己的类也被打进去了提交到集群时和集群上的类冲突会出现ClassCastException或者诡异的序列化错误。常见做法是在 assembly 里加scopeprovided/scope或者打包后手工确认没把flink-core打进去。我一般直接看 target 目录下那个-jar-with-dependencies.jar的体积正常几百 KB 到几 MB如果一下十几 MB基本就是把 Flink 打进去了。4.2 DataStream 方式MySqlSource 的关键参数与 Checkpoint 配置DataStream 的写法适合需要自己加工数据的场景自由度最高。核心是MySqlSource.builder()那一串参数。StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 开启 checkpointCDC 的 binlog 位点靠它保存不开就没法断点续传 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务取消时保留 checkpoint方便从位点恢复 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 失败重启策略1 天内最多 3 次每次间隔 1 分钟 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); env.getCheckpointConfig().setCheckpointStorage(hdfs://hadoop102:8020/flinkCDC); System.setProperty(HADOOP_USER_NAME, deploy); MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(hadoop103) .port(3306) .databaseList(test) .tableList(test.t1) .username(root) .password(000000) .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); DataStreamSourceString mysqlDS env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), MySQLSource); mysqlDS.print(); env.execute();几个参数值得单独说参数取值说明databaseListtest要捕获的库多个用逗号隔开tableListtest.t1必须是「库名.表名」的完整写法startupOptionsinitial首次启动先全量快照再读增量 binlogdeserializerJsonDebeziumDeserializationSchema把 SourceRecord 转成 JSON 字符串setCheckpointStoragehdfs://…/flinkCDC必须与 flink-conf.yaml 中的路径一致startupOptions有几种取值选错的后果差别很大。initial()是第一次启动先做全量快照再读增量最常用earliest()不做快照从 binlog 最早位置读只在确认 binlog 没被清理时用latest()从当前位点开始历史数据完全不要specificOffset()指定具体位点timestamp()从指定时间开始。生产上做首次同步基本都是initial()后续重启靠 checkpoint 恢复不会重复做快照。System.setProperty(HADOOP_USER_NAME, ...)这行看着不起眼是 HDFS 写 checkpoint 的保命符。不设的话作业用提交者的用户名去写 HDFS权限不对就 Permission denied任务反复重启但日志不明显。打包提交# 上传 jar 后提交 $FLINK_HOME/bin/flink run \ -m hadoop102:8081 \ -c com.example.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test.jar-m指定 JobManager 地址-c指定主类全限定名两个都不能省。4.3 FlinkSQL 方式建表语句里的 connector 参数不想写 Java 的话FlinkSQL 更省事一张建表语句就把 source 定义完了。CREATE TABLE t1 ( id STRING PRIMARY KEY NOT ENFORCED, name STRING ) WITH ( connector mysql-cdc, hostname hadoop103, port 3306, username root, password 000000, database-name test, table-name t1 );PRIMARY KEY NOT ENFORCED这个写法很多人第一次见会懵它声明的是逻辑主键不真的在 Flink 里做约束检查作用是告诉 planner 这条流是 upsert 流下游可以按主键更新。少了这一句后面的聚合和去重都不对。对应的 Java 入口很简单StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); StreamTableEnvironment tableEnv StreamTableEnvironment.create(env); tableEnv.executeSql(建表语句); Table table tableEnv.sqlQuery(select * from t1); table.execute().print();SQL 方式和 DataStream 方式提交到集群的方法完全一样都是flink run。区别在于 SQL 方式的 checkpoint 需要在代码里或者集群配置里开光靠 SQL 语句开不了这一点容易漏。4.4 YAML PipelineMySQL 到 Doris 整库同步的配置3.0 最省事的用法是 yaml pipeline一份配置文件就把 source、sink、路由全描述了不用写一行 Java。# job/mysql-to-doris.yaml source: type: mysql hostname: hadoop103 port: 3306 username: root password: 000000 # 正则写法test 库下所有表 tables: test.\.* # CDC 伪装成从库的 id 段不能与 MySQL 的 server-id 重复 server-id: 5400-5404 # 时区必须指定否则时间字段差 8 小时 server-time-zone: UTC8 sink: type: doris fenodes: hadoop102:7030 username: root password: 000000 # 允许轻量 schema changeMySQL 加列时 Doris 侧跟着加 table.create.properties.light_schema_change: true # 单 BE 环境必须设 1默认 3 会建表失败 table.create.properties.replication_num: 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1逐项拆一下几个容易写错的地方。tables: test.\.*里的点是正则元字符必须转义写成test.*会匹配到别的库。server-id: 5400-5404是个范围CDC 会在这个区间里给每个并行度分配一个 id范围太小并行度调大就会失败。server-time-zone: UTC8不写的话DATETIME类型同步到 Doris 会整体偏移 8 小时这种错查起来特别费劲因为数据看着「有」只是不对。replication_num在单 BE 的测试环境里必须改成 1Doris 默认三副本只有一个 BE 时建表直接报错。启动命令就一条# 提交 pipeline 作业 /opt/module/flink-cdc-3.0.0/bin/flink-cdc.sh job/mysql-to-doris.yaml脚本会解析 yaml、生成 Flink 作业图、调flink run提交控制台会打印作业 id。提交后去 WebUI 能看到一个叫Sync MySQL Database to Doris的作业跑起来之后 Doris 的test库里会自动出现 t1、t2、t3 三张表。4.5 路由变更把 test_route 写到另一个库路由route是 3.0 的另一个新东西作用是在同步过程中改表名或库名。典型场景是源库和目标库不同名或者多张表合并到一张。source: type: mysql hostname: hadoop103 port: 3306 username: root password: 000000 tables: test_route.\.* 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 route: # 源表正则匹配sink 端用 保留原表名 - source-table: test_route\.(.*) sink-table: doris_test_route. pipeline: name: Route MySQL to Doris parallelism: 1source-table用正则捕获sink-table里的是占位符表示沿用源端的表名。这样test_route.t1会同步成doris_test_route.t1库名换了表名不变。如果想把多张表合成一张把换成固定表名即可比如doris_test_route.merged_table。路由配错的表现是数据进了错的库或者建表时提示库不存在。排查方法很直接去 Doris 里SHOW TABLES FROM doris_test_route看一眼表没出来就是没匹配上先单独拿源表名去测正则。5. 常见问题排查六个把我卡住的坑5.1 启动阶段类找不到与 jar 冲突现象一提交作业立刻抛ClassNotFoundException: com.ververica.cdc.connectors.mysql.source.MySqlSource。原因是 pom 里用的还是 2.x 的坐标或者 jar 名字按 2.x 的找。3.0 的 DataStream 依赖坐标仍是flink-connector-mysql-cdc但版本必须写 3.0.0pipeline 模式必须用flink-cdc-pipeline-connector-mysql-3.0.0.jar。解决方式是把版本号锁死pipeline 的 jar 确认放进$FLINK_HOME/lib且所有节点的 lib 目录都拷了一份。现象二作业能提交运行时报NoSuchMethodError或ClassCastException。原因是 fat jar 里打进了 Flink 自己的类和集群上的版本冲突。解决方式是把 Flink 相关依赖的 scope 改成provided重新打包后确认 jar 体积正常。我一般打完包会看一眼体积几百 KB 才是对的。5.2 同步阶段数据没过去或者不对现象三作业显示 RUNNINGDoris 里库是空的一张表都没建。先查三处Doris 的目标库有没有提前建CDC 只自动建表不建库fenodes端口是不是 7030不是 90309030 是查询端口Doris 的 root 密码是不是和 MySQL 的搞混了。三个都对还不行就看 Flink 日志里 sink 段的报错通常是Unknown database或者鉴权失败。现象四数据同步过去了但时间字段整体差 8 小时。原因是 yaml 里没写server-time-zone: UTC8。MySQL 的DATETIME不带时区CDC 读到之后按默认时区解释和东八区差 8 小时。补上这一项重启作业历史数据需要重新全量同步一次才会修正。现象五MySQL 里加了列Doris 里没跟着变。table.create.properties.light_schema_change没开。这个参数控制 Doris 侧是否允许轻量 schema change不开的话加列不同步且不会有明显报错。开了之后ADD COLUMN能同步但DROP COLUMN和改类型仍然不支持需要手工处理。5.3 运行阶段Checkpoint 与位点回退现象六作业每隔几秒重启一次日志里是Checkpoint failed或Permission denied。两个原因最常见HDFS 的 checkpoint 目录权限不对写不进去或者没设HADOOP_USER_NAME作业用的用户和目录属主不一致。解决方式是hadoop fs -chmod -R 777 /flinkCDC放开权限同时在代码里加System.setProperty(HADOOP_USER_NAME, deploy)或者在提交时通过-Dyarn参数指定用户。还有一个容易被忽略的点replication_num忘了改成 1。单 BE 的 Doris 默认三副本建表表建不出来作业会卡在建表阶段反复重试日志里是Failed to create table之类看着像权限问题实际是副本数问题。提示这六个坑里前两个和后两个都属于「装的时候多看一眼就能避开」的类型。把版本、jar 位置、权限、副本数四件事在装之前确认一遍能省掉大半排查时间。6. 装完怎么验证四步确认数据真的同步过去了作业 RUNNING 不等于同步成功得按下面的顺序验一遍缺一步都可能留下隐患。第一步看 Checkpoint。打开 hadoop102:8081进作业详情页Checkpoint 标签下应该能看到周期性成功的记录Size 大于 0、Duration 在几百毫秒到几秒之间。如果一直是IN_PROGRESS或者失败先别往下走回去查 HDFS 权限。第二步看表有没有建出来-- 连 Doris 查询端口 mysql -uroot -p000000 -P9030 -hhadoop102 USE test; SHOW TABLES; -- 应该看到 t1、t2、t3 三张表第三步做变更测试这是最实在的验证。在 MySQL 里做一轮增删改然后去 Doris 里查-- MySQL 侧操作 USE test; INSERT INTO t1 VALUES(1004,zhaoliu); UPDATE t1 SET name lisi_new WHERE id 1002; DELETE FROM t1 WHERE id 1003; -- Doris 侧验证 USE test; SELECT * FROM t1 ORDER BY id; -- 期望结果1001/zhangsan、1002/lisi_new、1004/zhaoliu三种变更都能正确反映说明全量快照和增量 binlog 都通了。如果插入有、更新没有多半是主键没设对upsert 退化成 append。第四步验断点续传这是生产上真正救命的能力# 给作业打 savepoint $FLINK_HOME/bin/flink savepoint JobId hdfs://hadoop102:8020/flinkCDC/save # 在 WebUI 里 cancel 作业然后在 MySQL 里再改几条数据 # 从 savepoint 恢复 $FLINK_HOME/bin/flink run \ -s hdfs://hadoop102:8020/flinkCDC/save/savepoint-xxxx \ -c com.example.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test.jar恢复后去 Doris 里查那几条停机期间改的数据如果在说明位点保存和恢复链路是通的如果不在说明 checkpoint 没生效或者恢复时用了latest()之类的启动选项把增量跳过了。第五步补一个 schema 变更测试在 MySQL 里ALTER TABLE t1 ADD COLUMN age INT等几秒去 Doris 里DESC t1看 age 字段有没有出现。这一步过了日常业务加字段就不用停作业了。从那以后我每次装这套环境都强制走一遍这五步验证尤其是第四步的 savepoint 恢复——它平时看不出问题真出事的时候是唯一的后悔药。这套 Flink CDC 3.0 的完整手册里建表 SQL、pom、yaml 模板和排错记录都是现成的对着上面的顺序走一遍比零散搜配置快得多也能少踩几个我踩过的坑。希望帮到你。本文还有配套的精品资源点击获取
