简介一份面向具备一定Flink基础的大数据研发人员的实操型文档聚焦Flink CDC 3.0的实时数据捕获与同步帮助读者理解CDC原理并完成从MySQL到Doris的Streaming ETL落地。内容先厘清CDC概念及基于查询与基于Binlog两类方案的差异再引入flink-cdc-connectors组件的全量与增量读取能力随后通过完整案例展开数据准备、DataStream与Flink SQL两种实现方式以及MySQL到Doris的环境搭建、配置编写、任务启动与测试验证。文中对开启Binlog、设置检查点等关键配置有明确说明便于读者掌握数据一致性与任务稳定性的保障思路并可按业务需求调整参数。资源包共1个docx文件约145KB结构紧凑适合作为专题学习与项目部署参考。已有1789人学习。1. 一条 yaml 就想跑通 Flink CDC 3.0先认清你装的是哪一版上周又有人拿着一份 2.x 的教程来问import com.ververica.cdc.connectors.mysql.source.MySqlSource;这一行为什么标红。这就是 Flink CDC 3.0 最容易吃的一个亏——3.0 把代码包名从com.ververica.cdc整体迁到了org.apache.flink.cdc依赖坐标也从com.ververica换成了org.apache.flink可网上大部分 MySQL 到 Doris 的实时同步教程还是 2.x 的写法抄过去必然编译不过。更麻烦的是 3.0 同时存在两套形态DataStream/FlinkSQL 那套用flink-connector-mysql-cdc的 jarpipeline 那套用flink-cdc-pipeline-connector-mysql加一个flink-cdc.sh入口jar 放错目录任务起得来但一条数据都不同步。MySQL 业务库往 Doris 报表库做实时同步是现在最普遍的一类需求装不好要么全量快照跑完就静止要么 binlog 位点丢了得从头再来。这篇记录按我实际装过的那套环境写Flink 1.18.0、MySQL 8.0.31、Doris 1.2.4、Flink CDC 3.0.0从版本对齐一直写到怎么验证增量真的进来了。2. 版本对齐Flink 1.18.0、MySQL 8.0.31、Doris 1.2.4 各装在哪装这套东西最容易翻车的地方不在代码在版本和端口的对齐。Flink CDC 3.0 的 pipeline 形态对 Flink 版本有要求1.17 和 1.18 都能用1.16 及以下跑不起来Doris 侧要能建 Unique 模型表、要开 merge on write1.2.x 之后这些能力才齐MySQL 侧的 Binlog 参数如果是默认值连上去只会读到全量快照增量一条都捕获不到。所以先把表和机器摆清楚再动手解压。2.1 一张对照表把四台服务的角色定死我习惯先画一张表把每个组件装在哪台机器、开哪个端口、干什么用写清楚装的时候照着对出问题也知道去哪台机器的哪个日志里找。组件版本主机关键端口在这套流程里的角色JDK1.8hadoop102 / hadoop103无Flink 与 CDC 客户端运行环境Flink1.18.0hadoop102JobManager8081跑 CDC 作业的运行时HDFS3.xhadoop1028020存 checkpoint 和 savepointMySQL8.0.31hadoop1033306数据源需要开 BinlogDoris1.2.4hadoop1029030FE 查询/ 8030FE HTTP目标端pipeline 写入Flink CDC3.0.0hadoop102无解压后由flink-cdc.sh提交Doris 的fenodes填的是 FE 的 HTTP 端口不是 9030 那个 MySQL 协议端口。我见过不止一次把fenodes写成:9030报错是一串ConnectException看日志根本看不出是端口填错。默认 8030如果你的fe.conf里改过http_port以实际值为准。2.2 MySQL 8.0.31把 Binlog 按 CDC 的要求打开Flink CDC 底层走的是 Debezium靠读 Binlog 拿变更。MySQL 8.0 的默认配置里 Binlog 是开着的8.0 起log_bin默认 ON但binlog_format和库级过滤不一定对所以还是要显式写一遍。# /etc/my.cnf MySQL 8.0.31 server-id 1 # 打开 binlog值会作为 binlog 文件前缀 log-bin mysql-bin # CDC 必须是 ROWSTATEMENT/MIXED 拿不到完整的前后镜像 binlog_format ROW # 记录整行数据Debezium 依赖这个做 update/delete 解析 binlog_row_image FULL # MySQL 8 用这个参数替代 expire_logs_days单位秒这里保留 7 天 binlog_expire_logs_seconds 604800 # 只对这两个库产生 binlog按实际情况改 binlog-do-db test binlog-do-db test_route改完必须重启binlog_format这类参数不是动态的sudo systemctl restart mysqld重启后用两条 SQL 确认别靠猜SHOW VARIABLES LIKE binlog_format; -- 期望 ROW SHOW VARIABLES LIKE log_bin; -- 期望 ON SHOW MASTER STATUS; -- 有输出说明 binlog 真在生产binlog_row_image这一项很多人省略默认就是 FULL但如果之前有人为了省空间改成 MINIMALupdate 语句在 Doris 侧会变成只带主键的半行数据字段全对不上这类问题排查起来非常费劲。接着把库表建出来。这套环境里我准备了两个库test做全量增量的主链路test_route专门用来验证路由把源库表映射到目标端另一个库名。CREATE DATABASE test; CREATE DATABASE test_route; 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),(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里同样建这三张表t2故意插入不同主键1004/1005/1006这样路由跑起来以后Doris 里两个库的数据一眼能看出哪份是从哪来的不用猜。还要单独建一个同步账号别拿 root 跑长期任务CREATE USER cdc% IDENTIFIED BY 000000; -- SELECT 用于全量快照两个 REPLICATION 权限用于读 binlog GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO cdc%; FLUSH PRIVILEGES;只给SELECT不给REPLICATION SLAVE是最常见的一类翻车全量阶段数据哗哗进来日志也很干净就是增量阶段一动不动最后在 TaskManager 日志里才翻到一句权限不足。2.3 Flink 1.18.0检查点、Hadoop 依赖和 flink-conf.yamlFlink CDC 把 binlog 位点存在 checkpoint 里检查点没开所谓断点续传就是空话。Flink 1.18 的配置文件仍然是conf/flink-conf.yaml到 1.19 才换成config.yaml下面这几项我每次都会写死。# conf/flink-conf.yaml Flink 1.18.0 execution.checkpointing.interval: 5000 execution.checkpointing.mode: EXACTLY_ONCE # 状态后端用 hashmap1.18 里默认也是它显式写出来便于排错 state.backend: hashmap # checkpoint 落在 HDFS单机磁盘挂了作业还能恢复 state.checkpoints.dir: hdfs://hadoop102:8020/flink-cdc/checkpoints state.savepoints.dir: hdfs://hadoop102:8020/flink-cdc/savepoints # 任务被 cancel 时保留最后一次 CK这是恢复的前提 execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION参数里最容易被忽略的是最后一行。默认行为是作业取消时删掉 checkpoint等你发现有问题想从断点恢复文件已经没了只能从 binlog 最早位置重跑一遍全量。5 秒的间隔对数据量不大的场景够用如果库的写入量很大可以调到 10 秒到 30 秒减少 HDFS 小文件压力。Flink 1.18 的发行包默认不带 Hadoop 依赖写 HDFS 会报ClassNotFoundException: org.apache.hadoop.hdfs.DistributedFileSystem。常见做法是在conf/flink-conf.yaml同级配环境变量# conf/master 或 ~/.bashrc 里加确保提交作业的 shell 也能读到 export HADOOP_CONF_DIR/opt/module/hadoop-3.3.4/etc/hadoop export HADOOP_CLASSPATH$(hadoop classpath)HADOOP_CONF_DIR让 Flink 能读到core-site.xml里的fs.defaultFSHADOOP_CLASSPATH把 Hadoop 的 jar 塞进 classpath两个都缺checkpoint 就一定写不出去。改完配置分发到所有节点bin/start-cluster.sh起集群WebUI 能打开 8081 就说明运行时没问题。3. 装 Flink CDC 3.0tar 包、pipeline 连接器和 Maven 依赖Flink CDC 3.0 的安装不是装一个软件而是三件互不相干的东西一个解压即用的 bin 包提供flink-cdc.sh、两个 pipeline 连接器 jarMySQL 源和 Doris 汇、以及一份给 DataStream 作业用的 Maven 依赖。三件东西对应三种写作业的方式混着用是最常见的错误来源。3.1 解压 flink-cdc-3.0.0-bin.tar.gz 与 lib 目录的摆放规则# 解压到统一的模块目录和 Flink 平级 tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ cd /opt/module/flink-cdc-3.0.0 ls # bin conf lib log ...解压后先看lib目录里有什么。3.0 的 bin 包里自带flink-cdc-dist-3.0.0.jar这是 pipeline 的运行时但连接器不在里面需要自己放进去。把两个连接器 jar 拷到lib# 两个连接器必须放在 lib 下不是 Flink 的 lib cp /opt/software/flink-cdc-pipeline-connector-mysql-3.0.0.jar lib/ cp /opt/software/flink-cdc-pipeline-connector-doris-3.0.0.jar lib/ ls lib/ # flink-cdc-dist-3.0.0.jar # flink-cdc-pipeline-connector-mysql-3.0.0.jar # flink-cdc-pipeline-connector-doris-3.0.0.jar这里有个非常容易犯的错把 connector jar 扔进 Flink 自己的lib目录。pipeline 作业是flink-cdc.sh读 yaml、自己解析 source/sink、再翻译成 Flink 作业提交上去的连接器必须在 CDC 的 lib 下才能被加载器扫到。放进 Flink lib 的结果是提交时报找不到 type 为 mysql 的 connector而 Flink 集群一切正常很容易误判成版本问题。flink-cdc.sh默认会去localhost:8081找 Flink跨机器或端口不同就得显式指定# --flink-home 指向 Flink 1.18.0 安装目录脚本会用它下面的 bin/flink 提交 bin/flink-cdc.sh \ --flink-home /opt/module/flink-1.18.0 \ job/mysql-to-doris.yaml加上--flink-home之后提交走的是你指定的那套 Flink版本和配置都跟着它走省掉到底提交到哪个集群的疑惑。3.2 先确认 jar 里的包名再决定 import 写什么版本差异这件事靠jar tf一眼就能确认比翻文档快。# 看 MySQL CDC 连接器里 MySqlSource 的完整包名 unzip -l /opt/software/flink-connector-mysql-cdc-3.0.0.jar | grep MySqlSource # org/apache/flink/cdc/connectors/mysql/source/MySqlSource.class如果输出是org/apache/flink/cdc/...那么代码里就必须import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;。2.x 的 jar 里则是com/ververica/cdc/...。这两个包名对应的坐标也不一样版本Maven 坐标 groupId代码包名前缀2.4 及以前com.ververicacom.ververica.cdc3.0org.apache.flinkorg.apache.flink.cdc我一般的做法是拿到 jar 先跑一次jar tf | grep再决定 pom 和 import 怎么写不凭印象。这一步花十秒能省掉半小时的编译报错排查。3.3 DataStream 作业的 pom.xml 与打包DataStream 方式适合需要在流里做加工的场依赖要自己引全。Flink 1.18.0 加 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 运行时打包时 scope 用 provided集群里已有 -- 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 !-- FlinkSQL 方式才需要下面三个DataStream 可以不加 -- 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 dependency groupIdorg.apache.flink/groupId artifactIdflink-table-api-java-bridge/artifactId version${flink.version}/version /dependency !-- CDC 3.0 的 MySQL 连接器坐标已迁到 org.apache.flink -- dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.0.0/version /dependency !-- MySQL 驱动版本要和服务器对上8.0.31 服务端配 8.0.31 驱动最省心 -- dependency groupIdcom.mysql/groupId artifactIdmysql-connector-j/artifactId version8.0.31/version /dependency /dependenciesflink-table-planner_2.12这个_2.12后缀是 Scala 版本Flink 1.18 官方包默认按 2.12 编译写错后缀会出现NoSuchMethodError这种莫名其妙的运行时错误。打包用maven-assembly-plugin打成jar-with-dependencies把 CDC 连接器和 MySQL 驱动一起塞进去提交时不用再往集群 lib 里扔 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 goalsgoalsingle/goal/goals /execution /executions /plugin /plugins /buildphase绑到package意味着敲一次mvn clean package就同时产出普通 jar 和带依赖的胖包不用再多跑一条命令。4. 三条路写同一个同步DataStream、FlinkSQL、yaml pipeline 怎么选同一个 MySQL 到 Doris 的同步Flink CDC 3.0 给了三种写法。选型标准其实很简单要在流里加工就用 DataStream纯 SQL 就能表达就用 FlinkSQL只是库到库搬数据就用 pipeline 的 yaml。三者的坑完全不同我按实际写过的顺序记一遍。4.1 DataStreamMySqlSource.builder() 的几个参数先看代码这是一份最小可跑的全量加增量作业import org.apache.flink.cdc.connectors.mysql.source.MySqlSource; import org.apache.flink.cdc.connectors.mysql.table.StartupOptions; import org.apache.flink.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. 流处理环境并行度先给 1保证输出顺序好观察 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. 检查点CDC 靠它保存 binlog 位点不开就没有断点续传 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 一分钟内失败不超过 3 次就自动重启防止抖动直接挂掉 env.setRestartStrategy( RestartStrategies.failureRateRestart(3, Time.days(1L), Time.minutes(1L))); env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage(hdfs://hadoop102:8020/flinkCDC); System.setProperty(HADOOP_USER_NAME, bigdata); // 3. 构建 MySQL 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(); // 4. 接入数据流 DataStreamSourceString mysqlDS env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), MySqlSource); // 5. 直接打印验证阶段看 TaskManager 日志最直观 mysqlDS.print(); // 6. 提交 env.execute(); } }startupOptions这一项是整段代码里最需要想清楚的它有五个取值行为差别极大取值启动行为适用场景initial()先做全量快照再接着读最新 binlog首次接入要历史数据earliest()不做快照从 binlog 最早位置读binlog 保留期内能覆盖全部历史latest()不做快照只读启动之后的新变更只关心增量做旁路监控specificOffset()从指定 binlog 位点开始手工指定恢复位置timestamp()从指定时间戳开始之前的变更跳过按时间补数我一般首次跑用initial()验证通过后如果需要重建就改用latest()加 savepoint 恢复避免再来一次全量把 Doris 打满。databaseList和tableList要写全正则的语义test.t1是精确匹配test\.t.*才是整库表。少写一个反斜杠可能一张表都匹配不上而且不报错就是没数据。System.setProperty(HADOOP_USER_NAME, ...)这行在 HDFS 做了权限校验的环境里必须加否则 checkpoint 写入会抛Permission denied。正规做法是在提交时用-yD传参或者配 Kerberos临时验证阶段先这样写省事。打包上传后提交# 先起 HDFS 和 Flink 集群 start-dfs.sh /opt/module/flink-1.18.0/bin/start-cluster.sh # 提交作业-c 指定入口类 /opt/module/flink-1.18.0/bin/flink run \ -m hadoop102:8081 \ -c com.example.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test.jar提交成功后去 TaskManager 日志看输出能看到I、-U、U这类 Debezium 风格的变更标记说明全量快照和增量切换都正常。-m指定 JobManager 地址集群不在本机时必须带。4.2 FlinkSQLmysql-cdc 建表语句里的主键声明不想写 Java 的FlinkSQL 方式代码量少一半本质是把 Source 的配置搬进 DDL 的WITH里。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); // connector 写 mysql-cdc其余键值对和 DataStream 的 builder 参数一一对应 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 要靠主键判断 upsert 语义漏了会被当成 append-only 流update 和 delete 直接丢二是database-name/table-name是单库单表的精确值要同步整库得写成test\\.t.*这种正则形式和 DataStream 里的tableList语义一致。FlinkSQL 作业打包到集群测试的方式和 DataStream 完全相同flink run -c指到入口类即可。4.3 yaml pipelineMySQL 到 Doris 的 StreamingETL 配置这才是 3.0 真正新增的东西。不用写一行 Java一份 yaml 描述 source、sink、routeflink-cdc.sh负责翻译成 Flink 作业。# job/mysql-to-doris.yaml source: type: mysql hostname: hadoop103 port: 3306 username: root password: 000000 # 正则test 库下所有表 tables: test\.t.* # 每个并行度占一个 server-id范围要和 parallelism 对得上 server-id: 5400-5404 server-time-zone: UTC8 sink: type: doris # FE 的 HTTP 端口不是 9030 fenodes: hadoop102:8030 username: root password: 000000 # 建表时允许轻量 schema 变更源端加列能自动跟上 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几个参数是踩过才知道的。server-id写成一个范围是因为 pipeline 的每个并行度都要以独立的从库身份连 MySQLparallelism: 1只占一个写5400-5404留了余量如果并行度调到 2 而范围只有一个值第二个子任务会连不上。table.create.properties.replication_num在单 BE 的测试环境必须设成 1Doris 建表默认 3 副本副本数大于 BE 数量会直接建表失败报错信息里是一串 tablet 创建失败跟 CDC 本身没关系。路由变更单独写一段把源库表映射到目标端另一个库# 追加在 mysql-to-doris.yaml 里route 是顶层键 route: - source-table: test_route\.t1 sink-table: doris_test_route.t1source-table用库名.表名的正则sink-table是目标端的库.表Doris 里对应的库要提前建好。Doris 侧建库走 MySQL 协议-- 连 Doris FE 的 9030 端口 mysql -uroot -p000000 -P9030 -hhadoop102 CREATE DATABASE test; CREATE DATABASE doris_test_route;启动顺序也有讲究Flink 集群 → Doris FE → Doris BE → 提交 pipeline。少一步就会出现作业 RUNNING 但 Doris 里没表的情况因为 sink 端连不上 FECDC 会一直重试而不失败退出。# Doris 启停按实际安装路径 /opt/module/doris/fe/bin/start_fe.sh --daemon /opt/module/doris/be/bin/start_be.sh --daemon # 提交 pipeline cd /opt/module/flink-cdc-3.0.0 bin/flink-cdc.sh --flink-home /opt/module/flink-1.18.0 job/mysql-to-doris.yaml5. 排查与避坑五条卡了我半天的记录这套环境我从头装过三遍下面五条是同一种现象反复出现的按看到什么 → 为什么 → 怎么解记遇到时直接对号。5.1 连接与权限类任务起来了但没数据现象一全量快照跑完日志安静后续 MySQL 里的 insert 一条都不同步。原因是同步账号权限不足或 Binlog 没真正打开。Debezium 读 binlog 需要REPLICATION SLAVE和REPLICATION CLIENT只给SELECT时全量能过、增量静默失败。解决按前面那条GRANT重新授权然后SHOW GRANTS FOR cdc%;核对再SHOW MASTER STATUS;确认 binlog 有输出。改完要重启作业权限是连接时校验的热改不生效。现象二pipeline 提交后立刻报找不到 type 为 mysql 的 connector。原因是连接器 jar 放错目录了扔进了 Flink 的 lib 而不是flink-cdc-3.0.0/lib。pipeline 的连接器由 CDC 自己的类加载器扫描Flink 的 lib 对它不可见。解决ls lib/确认三个 jar 都在缺哪个补哪个补完不用重启 Flink重新提交即可。现象三DataStream 作业编译报package com.ververica.cdc does not exist。这就是开头那个版本坑手上是 3.0 的 jar代码照 2.x 教程写的。解决unzip -l看一眼 jar 里的实际包名3.0 统一改成org.apache.flink.cdcpom 里的 groupId 同步换成org.apache.flink。两者必须一起改只改一个照样报错。5.2 数据一致性类数据进来了但对不上现象四Doris 里字段数比 MySQL 少源端后来加的列没同步过去。原因是建表属性没开轻量 schema 变更pipeline 只在首次建表时按当时的表结构建后续加列不会自动跟进。解决sink 里加上table.create.properties.light_schema_change: true然后重建目标表重跑一次。源端加列这种变更在业务迭代里很常见第一次部署就该把这个开关写进 yaml别等出了问题再补。现象五MySQL 里删了一条Doris 里还在。Doris 侧的删除依赖 Unique 模型加 merge on writepipeline 侧还要显式打开删除开关sink.enable-delete。三者缺一个delete 事件就会被静默丢弃。解决确认目标表是 Unique Key 模型且开了 merge on writesink 配置补上删除开关然后重新同步。这也是为什么验证环节一定要造一条 delete只测 insert 和 update 是测不出这个坑的。6. 装完怎么验证一条 UPDATE 走通全量和增量的闭环装完不做验证等于没装。我的验证顺序固定四步每步都有明确的看到什么才算过。第一步看全量。作业 RUNNING 之后去 Doris 查行数test.t1应该是 3 条和 MySQL 里一致。这时候去看 Flink WebUI 的 Checkpoint 页面最近一次 checkpoint 必须是 COMPLETED且大小不为 0——checkpoint 一直是 IN_PROGRESS 或者失败说明 HDFS 那条路没通后面所有恢复操作都是空谈。第二步看增量。在 MySQL 里执行一条 update立刻去 Doris 查-- MySQL 侧 UPDATE test.t1 SET name zhangsan_v2 WHERE id 1001; -- Doris 侧几秒内应该能看到新值 SELECT * FROM test.t1 WHERE id 1001;延迟在几秒内属于正常具体取决于 checkpoint 间隔和 Doris 的导入攒批。如果超过一分钟还没变去看 TaskManager 日志里有没有反序列化异常或者 Doris 侧是不是在建表阶段就失败了。第三步看断点续传这是最容易被跳过的一步。给作业打一个 savepoint取消期间往 MySQL 里插一条新数据再从 savepoint 恢复看那条数据有没有补进来# 打 savepointJobId 从 WebUI 拿 bin/flink savepoint JobId hdfs://hadoop102:8020/flinkCDC/save # WebUI 里 cancel 作业后从 savepoint 恢复 bin/flink run \ -s hdfs://hadoop102:8020/flinkCDC/save/savepoint-xxxxxx \ -c com.example.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test.jar恢复后如果那条停机期间插入的数据出现在 Doris 里说明位点保存和续读都是对的。这一步不过线上任何一次发版重启都会丢一段数据而且丢了不会报警。第四步看路由。test_route库的数据应该落到doris_test_route两个库的主键范围不一样一眼能分辨。路由配错的表现是表建在了默认库下数据其实同步过去了只是找错地方。验证项操作通过标准全量快照作业启动后查 Doris 行数与 MySQL 一致检查点WebUI 看 Checkpoint 列表COMPLETED 且大小非 0增量更新MySQL UPDATE 一条Doris 数秒内看到新值增量删除MySQL DELETE 一条Doris 对应行消失断点续传savepoint → cancel → 恢复停机期间的数据被补上路由查doris_test_route表和数据都在预期库下早年间我图快装完看到 Doris 里有数据就当成功了结果上线第二天一次集群重启binlog 位点没存住整库重跑了一遍全量把报表库压得半小时打不开。从那以后我每次部署这套环境savepoint 恢复这一步都强制走一遍哪怕只是本地验证环境也不省这两分钟。希望帮到你。本文还有配套的精品资源点击获取