简介：一份面向具备一定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的 jar，pipeline 那套用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 write，1.2.x 之后这些能力才齐；MySQL 侧的 Binlog 参数如果是默认值，连上去只会读到全量快照，增量一条都捕获不到。所以先把表和机器摆清楚，再动手解压。

2.1 一张对照表把四台服务的角色定死

我习惯先画一张表，把每个组件装在哪台机器、开哪个端口、干什么用写清楚，装的时候照着对，出问题也知道去哪台机器的哪个日志里找。

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 必须是 ROW，STATEMENT/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，但如果之前有人为了省空间改成 MINIMAL，update 语句在 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.yaml

Flink 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 # 状态后端用 hashmap，1.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.defaultFS，HADOOP_CLASSPATH把 Hadoop 的 jar 塞进 classpath，两个都缺，checkpoint 就一定写不出去。改完配置分发到所有节点，bin/start-cluster.sh起集群，WebUI 能打开 8081 就说明运行时没问题。

3. 装 Flink CDC 3.0：tar 包、pipeline 连接器和 Maven 依赖

Flink CDC 3.0 的安装不是"装一个软件"，而是三件互不相干的东西：一个解压即用的 bin 包（提供flink-cdc.sh）、两个 pipeline 连接器 jar（MySQL 源和 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/...。这两个包名对应的坐标也不一样：

我一般的做法是拿到 jar 先跑一次jar tf | grep，再决定 pom 和 import 怎么写，不凭印象。这一步花十秒，能省掉半小时的编译报错排查。

3.3 DataStream 作业的 pom.xml 与打包

DataStream 方式适合需要在流里做加工的场���，依赖要自己引全。Flink 1.18.0 加 CDC 3.0.0，我把关键依赖列出来，其余可以按需增删。

<properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> <flink.version>1.18.0</flink.version> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties> <dependencies> <!-- Flink 运行时，打包时 scope 用 provided，集群里已有 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink.version}</version> </dependency> <!-- FlinkSQL 方式才需要下面三个，DataStream 可以不加 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-runtime</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java-bridge</artifactId> <version>${flink.version}</version> </dependency> <!-- CDC 3.0 的 MySQL 连接器，坐标已迁到 org.apache.flink --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>3.0.0</version> </dependency> <!-- MySQL 驱动版本要和服务器对上，8.0.31 服务端配 8.0.31 驱动最省心 --> <dependency> <groupId>com.mysql</groupId> <artifactId>mysql-connector-j</artifactId> <version>8.0.31</version> </dependency> </dependencies>

flink-table-planner_2.12这个_2.12后缀是 Scala 版本，Flink 1.18 官方包默认按 2.12 编译，写错后缀会出现NoSuchMethodError这种莫名其妙的运行时错误。打包用maven-assembly-plugin，打成jar-with-dependencies，把 CDC 连接器和 MySQL 驱动一起塞进去，提交时不用再往集群 lib 里扔 jar：

<build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-assembly-plugin</artifactId> <version>3.0.0</version> <configuration> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> </configuration> <executions> <execution> <id>make-assembly</id> <phase>package</phase> <goals><goal>single</goal></goals> </execution> </executions> </plugin> </plugins> </build>

phase绑到package，意味着敲一次mvn clean package就同时产出普通 jar 和带依赖的胖包，不用再多跑一条命令。

4. 三条路写同一个同步：DataStream、FlinkSQL、yaml pipeline 怎么选

同一个 MySQL 到 Doris 的同步，Flink CDC 3.0 给了三种写法。选型标准其实很简单：要在流里加工就用 DataStream，纯 SQL 就能表达就用 FlinkSQL，只是"库到库搬数据"就用 pipeline 的 yaml。三者的坑完全不同，我按实际写过的顺序记一遍。

4.1 DataStream：MySqlSource.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 MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("hadoop103") .port(3306) .databaseList("test") .tableList("test.t1") .username("root") .password("000000") .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); // 4. 接入数据流 DataStreamSource<String> mysqlDS = env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), "MySqlSource"); // 5. 直接打印，验证阶段看 TaskManager 日志最直观 mysqlDS.print(); // 6. 提交 env.execute(); } }

startupOptions这一项是整段代码里最需要想清楚的，它有五个取值，行为差别极大：

我一般首次跑用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 FlinkSQL：mysql-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 pipeline：MySQL 到 Doris 的 StreamingETL 配置

这才是 3.0 真正新增的东西。不用写一行 Java，一份 yaml 描述 source、sink、route，flink-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: UTC+8 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 的每个并行度都要以独立的从库身份连 MySQL，parallelism: 1只占一个，写5400-5404留了余量；如果并行度调到 2 而范围只有一个值，第二个子任务会连不上。table.create.properties.replication_num在单 BE 的测试环境必须设成 1，Doris 建表默认 3 副本，副本数大于 BE 数量会直接建表失败，报错信息里是一串 tablet 创建失败，跟 CDC 本身没关系。

路由变更单独写一段，把源库表映射到目标端另一个库：

# 追加在 mysql-to-doris.yaml 里，route 是顶层键 route: - source-table: test_route\.t1 sink-table: doris_test_route.t1

source-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 端连不上 FE，CDC 会一直重试而不失败退出。

# 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.yaml

5. 排查与避坑：五条卡了我半天的记录

这套环境我从头装过三遍，下面五条是同一种现象反复出现的，按"看到什么 → 为什么 → 怎么解"记，遇到时直接对号。

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.cdc，pom 里的 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 write，pipeline 侧还要显式打开删除开关（sink.enable-delete）。三者缺一个，delete 事件就会被静默丢弃。解决：确认目标表是 Unique Key 模型且开了 merge on write，sink 配置补上删除开关，然后重新同步。这也是为什么验证环节一定要造一条 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 恢复，看那条数据有没有补进来：

# 打 savepoint，JobId 从 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 里有数据就当成功了，结果上线第二天一次集群重启，binlog 位点没存住，整库重跑了一遍全量，把报表库压得半小时打不开。从那以后我每次部署这套环境，savepoint 恢复这一步都强制走一遍，哪怕只是本地验证环境，也不省这两分钟。希望帮到你。

本文还有配套的精品资源，点击获取