简介：《Flink CDC 3.0：MySQL数据实时捕获与同步至Doris的全流程解析及应用案例》由尚硅谷研究院编写，面向具备Flink基础的大数据研发与实时同步工程师，帮助读者理解CDC原理并落地MySQL到Doris的Streaming ETL。文档先厘清变更数据捕获概念，对比基于查询与基于Binlog方案在延迟、完整性和数据库压力上的差异，再说明flink-cdc-connectors组件的能力。实操从MySQL建库建表与数据准备讲起，覆盖DataStream和Flink SQL两种方式，并完整演示MySQL至Doris链路的环境搭建、配置编写、任务启动与测试，适合需要打通实时入仓链路、验证数据一致性及排查同步延迟的读者。压缩包共1个docx文件，约145KB，按CDC简介与案例实操两章组织，兼顾原理与工程实践，结构紧凑便于边读边练。已有1789人学习，可作为掌握开启Binlog、设置检查点等配置细节与排错思路的参考。
1. 从一张订单表说起：MySQL 到 Doris 的实时同步到底难在哪
Flink CDC 3.0 这套东西，我最开始接触是因为一个很具体的麻烦：业务方要求 MySQL 里刚改的订单状态，五分钟内必须在 Doris 的报表里看到。批处理顶不住这种需求，一条 StreamingETL 链路就得自己搭，而 Flink CDC 是这条链路上最省事的一环——不用额外部署采集组件、不用维护消息队列做中转、不用自己解析 Binlog 日志格式。它把 MySQL 的增量变更直接变成 Flink 里的一条流，再往 Doris 里写，全量和增量一次搞定，断点续传靠 Checkpoint 兜底。
这份资料的价值在于它把三件事串成了一条完整的线：CDC 的原理与两种执行模式的取舍、DataStream 和 Flink SQL 两种编码方式怎么接 MySQL、以及 3.0 新增的 YAML Pipeline 怎么把 MySQL 整库同步到 Doris。适合已经能跑起一个 Flink 作业、但还没真正在生产链路上趟过实时同步坑的开发者。下面按「选型 → 环境 → 编码 → 避坑 → 改造」的顺序拆。
2. 选型与架构拆解：查询式 CDC 与 Binlog 式 CDC 差在哪条链路上
2.1 一张对比表背后的执行模式差异
CDC 全称 Change Data Capture，核心动作只有一句话：盯住数据库的变动（insert / update / delete），按发生顺序记下来，喂给下游消费。听起来简单，但「怎么盯」这件事决定了整条链路的延迟、数据库压力和完整性。
传统做法是基于查询：定时跑一条select * from t where update_time > 上次时间，捞增量。Sqoop、DataX 走的就是这个路子，属于 Batch 模式。它的硬伤有三个——删不掉的数据它看不见（物理删除在源表里已经没了）、更新延迟取决于轮询间隔、每次全表扫都给源库加一份压力。
另一种是基于 Binlog：MySQL 把每次数据变更写进二进制日志，采集端伪装成从库去订阅这份日志。Canal、Maxwell、Debezium 都是这个思路，属于 Streaming 模式。它能拿到完整的 before/after 镜像，包括删除操作，延迟在毫秒到秒级，而且读 Binlog 对业务库几乎无感。
| 维度 | 基于查询的 CDC | 基于 Binlog 的 CDC |
|---|---|---|
| 代表产品 | Sqoop、DataX | Canal、Maxwell、Debezium、Flink CDC |
| 执行模式 | Batch | Streaming |
| 能否捕获全部变更 | 否，删除和中间态会丢 | 是，含 before/after 镜像 |
| 变化延迟 | 高，取决于轮询周期 | 低，接近实时 |
| 对源库压力 | 有，周期性全表或范围扫描 | 低，读 Binlog 流 |
| 断点续传 | 靠水位字段，容易漏 | 靠位点（offset/binlog position） |
选型结论很直接：只要下游对时效的要求是分钟级以内，或者业务表存在物理删除，就不该在查询式 CDC 上纠缠。剩下的问题只是「用哪个 Binlog 采集器」，这就是 Flink CDC 要回答的。
2.2 flink-cdc-connectors 在整条链路里的位置
Flink CDC 不是一个独立服务，它是 Flink 的一个 Source 组件。内部用 Debezium 做 Binlog 解析引擎，外面套一层 Flink 的 Source API，把解析出来的变更事件变成 Flink 数据流里的记录。这样做的好处是省掉了「采集器 → 消息队列 → 计算引擎」这条经典三段式链路里的中间环节，采集和计算在同一个作业里完成。
关键设计在状态管理上。作业启动时先做一次全量快照（snapshot），把当前表里的存量数据读出来；快照读完的位置和之后的 Binlog 位点，都会作为算子状态参与 Flink 的 Checkpoint。所以恢复的时候不是「从头再来」，而是接着上次的位置继续读。这也是为什么后面代码里 Checkpoint 必须开——不开 Checkpoint，这套断点续传的能力等于没有。
从数据形态看，MySqlSource拿到的是 Debezium 的SourceRecord，需要反序列化成下游能认的格式。最省事的是JsonDebeziumDeserializationSchema，直接输出 JSON 字符串，带op字段标识操作类型（c 插入、u 更新、d 删除、r 快照读）。要落库到 Doris，通常还得在中间做一层解析，或者干脆走 3.0 的 Pipeline 让它内置处理。
2.3 3.0 的 Pipeline 模型：把 Source 和 Sink 装进一个 YAML
3.0 最大的变化是引入了 Pipeline 这个概念，把「采集端」和「写入端」拆成独立的 connector jar，再用一份 YAML 把它们装配起来。作业不再需要你写 Java 代码，flink-cdc.sh读配置文件、拼出 Flink 作业图、提交到集群。
这个模型里有三个角色：source connector 负责读源库（MySQL），sink connector 负责写目标端（Doris、Kafka、Paimon 等），pipeline 负责中间的 schema 演进、表名路由和数据转换。这样拆的收益是可组合——同一份源配置，换一个 sink 配置就能写到不同目标，不用改代码；代价是灵活性受限于 YAML 支持的选项，复杂的业务转换还是得回到 DataStream。
三种方式的定位可以这么理解：DataStream 最灵活，适合要做复杂处理或自定义序列化的场景；Flink SQL 最轻，适合快速验证和简单同步；YAML Pipeline 最工程化，适合整库同步这种配置量大的活儿。后面三种都会过一遍，按你的场景挑。
3. 环境准备：MySQL Binlog 开关与 Doris 集群的一次性改造
3.1 建库建表与造一批能验证增删改的数据
先准备两个库：一个test做主同步，一个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) ); -- t3 故意用不同的列名和类型，用来验证 schema 演进 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));接着灌初始数据。注意test和test_route里的 t1、t2 主键故意错开，后面看 Doris 里的结果时，一眼就能分辨出哪条记录是从哪个库同步过去的。
USE test; INSERT INTO t1 VALUES ('1001','u001'),('1002','u002'),('1003','u003'); INSERT INTO t2 VALUES ('1001','u001'),('1002','u002'),('1003','u003'); INSERT INTO t3 VALUES ('1001','F'),('1002','F'),('1003','M'); USE test_route; INSERT INTO t1 VALUES ('1001','u001'),('1002','u002'),('1003','u003'); INSERT INTO t2 VALUES ('1004','u004'),('1005','u005'),('1006','u006'); INSERT INTO t3 VALUES ('1001','F'),('1002','F'),('1003','M');建表和造数这两步看着琐碎，但后面的所有验证都依赖它。特别是test_route.t2用了 1004-1006 这组主键，当你看到 Doris 里doris_test_route.t2出现这三条记录时，就能确认 route 规则确实生效了，而不是碰巧同步到了一个同名表。
3.2 Binlog 参数：四个必须改对的地方
MySQL 8.0 默认关闭 Binlog，这一项不改，Flink CDC 启动时会直接报「无法读取 binlog」之类的错。编辑my.cnf（不同发行版路径可能是/etc/my.cnf或/etc/mysql/mysql.conf.d/mysqld.cnf），在[mysqld]段落下加配置。
[mysqld] # 集群内唯一即可，CDC 会用它伪装成从库，重复会顶掉别的从库 server-id = 1 # 开启 binlog 并指定文件前缀，实际文件是 mysql-bin.000001 这种 log-bin = mysql-bin # 必须是 row，statement/mixed 拿不到 before/after 镜像，Debezium 直接解析不了 binlog_format = row # 只记录这两个库，减少日志量；不写则全库记录 binlog-do-db = test binlog-do-db = test_route改完systemctl restart mysqld，用show variables like 'binlog_format'和show master status各确认一次。前者必须是ROW，后者要能看到正在写的 binlog 文件名。
四个参数里最容易翻车的是binlog-do-db。它的语义是「按当前默认库判断」，如果一条 SQL 先use test再写test_route.t1，这条变更不会被记录。所以要么在连接层面强制指定库，要么干脆去掉这个限制、全库开 Binlog，用binlog-ignore-db排除系统库反而更稳。
注意：
binlog_format改成row之后，单条变更的日志体积会比 statement 大不少。大表批量更新的场景下要留意磁盘水位，必要时配合binlog_expire_logs_seconds缩短保留时间。
3.3 Doris 侧：FE/BE 起起来，建库属性别忘
Doris 这边要起两个角色：FE 负责元数据和查询规划，BE 负责实际存储。先bin/start_fe.sh --daemon起 FE，再用 MySQL 客户端连上 FE 的 9030 端口执行ALTER SYSTEM ADD BACKEND "node02:9050"把 BE 挂进来，然后bin/start_be.sh --daemon。BE 加不进去，后面建表会卡在副本数不足。
# 通过 MySQL 协议连 FE，端口是 9030 不是 3306 mysql -uroot -P9030 -hnode01 # 建目标库，名字故意和源库区分开，方便验证 route CREATE DATABASE IF NOT EXISTS test; CREATE DATABASE IF NOT EXISTS doris_test_route;建表这一步交给 Flink CDC 自动做，但有个前提：Doris 侧建表时副本数默认是 3，单机或两节点环境会一直处于副本不健康状态，写入直接失败。这就是后面 YAML 里table.create.properties.replication_num: 1的由来——环境有几个 BE，就设几（不超过）。同理light_schema_change要设为 true，否则源表加列时 Doris 这边会走重量级 Schema Change，同步任务会长时间阻塞。
3.4 Flink 侧：Checkpoint 与 connector jar 都要到位
Flink 1.18 的集群配置里要打开 Checkpoint 间隔，否则 Pipeline 模式的作业根本提交不上去——3.0 的 Pipeline 内部依赖 Checkpoint 做全量到增量的阶段切换。
# conf/flink-conf.yaml execution.checkpointing.interval: 5000然后是 connector 的 jar。Pipeline 模式下，flink-cdc-pipeline-connector-mysql-3.0.0.jar和flink-cdc-pipeline-connector-doris-3.0.0.jar都要放进$FLINK_HOME/lib下。放错位置（比如放进flink-cdc-3.0.0/lib）是最常见的启动失败原因，因为最终拉起作业的是 Flink 集群本身，它只认自己 lib 下的包。
4. 三种接入方式：DataStream、Flink SQL 与 YAML Pipeline 落地 MySQL→Doris
4.1 Maven 依赖与打包插件
DataStream 方式需要一套完整依赖。Flink 相关的五个包版本统一用 1.18.0，CDC connector 单独用 3.0.0，MySQL 驱动用 8.0.31 和源库版本对齐。
<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 运行时与 Table API，SQL 方式同样依赖这几个 --> <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> <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> <!-- CDC 核心包，DataStream 和 SQL 都靠它 --> <dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>3.0.0</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>8.0.31</version> </dependency> </dependencies> <build> <plugins> <!-- 打成带依赖的胖包，否则集群上找不到 CDC 的类 --> <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>几个参数值得说明。flink-table-planner_2.12的后缀是 Scala 版本，必须和集群用的 Flink 安装包一致（Flink 1.18 默认是 2.12），版本对不上会在作业提交时抛NoSuchMethodError，报错信息完全看不出是 Scala 版本问题——这类问题后面避坑章节还会提到。maven-assembly-plugin的jar-with-dependencies会生成xxx-jar-with-dependencies.jar，提交作业时用的就是它，不是默认的瘦包。
4.2 DataStream 主程序：六步走完一个 CDC 作业
主程序的结构很固定，六个 TODO 分别是环境、Checkpoint、Source、DataStream、Sink、执行。中间 Checkpoint 那段配置最多，也最容易被略过，但恰恰是它决定了作业能不能断点续传。
public class FlinkCDCDataStreamTest { public static void main(String[] args) throws Exception { // 1. 流处理环境，并行度先给 1，方便看日志 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. Checkpoint：CDC 的 binlog 位点存在这里，不开等于没有断点续传 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); // 故障重启策略：1 分钟内最多失败 3 次 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); // 状态后端与存储位置 env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://node01:8020/flink-cdc/ck"); System.setProperty("HADOOP_USER_NAME", "flink"); // 3. 构建 MySQL Source MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("node03") .port(3306) .databaseList("test") .tableList("test.t1") .username("root") .password("Mysql@123") .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); // 4. 从 Source 生成数据流 DataStreamSource<String> mysqlDS = env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), "MySQLSource"); // 5. 输出，生产上换成 Doris/Kafka Sink mysqlDS.print(); // 6. 执行 env.execute("flink-cdc-mysql-test"); } }逐段说明。enableCheckpointing(3000L, EXACTLY_ONCE)里的 3000 是毫秒，作业跑起来后每 3 秒做一次快照，间隔太大会让恢复时重放的数据变多，太小则给 HDFS 造成压力，5 秒左右是个折中值。RETAIN_ON_CANCELLATION很关键——手工 cancel 作业后，如果不保留最后一次 Checkpoint，重启时只能从头做全量，几十 G 的表会直接把你按在地上。HADOOP_USER_NAME这行是给 HDFS 写权限用的，容器里跑可以换成 Kerberos 认证。
MySqlSource的 builder 里，databaseList和tableList都支持正则，test\\.t.*这种写法可以把整个库的表都纳进来。startupOptions是语义最重的参数，取值直接决定作业第一次启动时的行为：
| 取值 | 首次启动行为 | 适用场景 |
|---|---|---|
initial() | 先做全量快照，再接着读最新 binlog | 首次接入，要历史数据 |
earliest() | 不做快照，从 binlog 最早位置读 | binlog 保留期内能覆盖全部历史 |
latest() | 不做快照，从 binlog 末尾读 | 只要增量，不要历史 |
specificOffset() | 从指定 binlog 位点读 | 手工接续某个断点 |
timestamp() | 从指定时间戳开始读，更早的事件丢弃 | 按时间补数 |
initial()是最常用的，但要注意它做快照时会短暂给源库加读锁（Flink CDC 3.0 默认用无锁算法读快照，但如果表没有主键或用了某些配置，会退化到加锁模式），业务高峰期接入需要提前评估。
4.3 提交、Savepoint 与从检查点恢复
打包后上传到集群，用flink run提交。注意-c后面跟的是完整类名。
# 提交作业 bin/flink run -m node01:8081 -c com.demo.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test.jar # 手工打一个 Savepoint，路径指向 HDFS bin/flink savepoint <JobId> hdfs://node01:8020/flink-cdc/save # 在 WebUI 里 cancel 作业，然后在 MySQL 里做增删改 # 从 Savepoint 恢复 bin/flink run -s hdfs://node01:8020/flink-cdc/save/savepoint-xxxxxx \ -c com.demo.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar验证恢复是否真的生效，看 TaskManager 日志里有没有重新做全量快照。如果日志显示直接进入 Binlog 读取阶段，说明位点恢复成功；如果又开始刷Snapshot相关日志，那就是位点没接上，多半是 Checkpoint 目录没权限或者 Savepoint 路径写错了。这一步我见过不止一次「以为恢复了、其实重跑了全量」的情况，日志一定要翻。
4.4 Flink SQL 写法与取舍
同样的需求用 SQL 表达，代码量能压到二十行以内。核心是一张带connector = 'mysql-cdc'的建表语句。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); env.enableCheckpointing(3000L); // SQL 方式同样必须开 CK 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' = 'node03',\n" + " 'port' = '3306',\n" + " 'username' = 'root',\n" + " 'password' = 'Mysql@123',\n" + " 'database-name' = 'test',\n" + " 'table-name' = 't1'\n" + ")"); tableEnv.sqlQuery("SELECT * FROM t1").execute().print();PRIMARY KEY NOT ENFORCED这句不能省，Flink SQL 需要靠主键信息判断 upsert 语义，缺少它会退化成 append 模式，更新和删除都表达不了。connector名称在 3.0 里仍然是mysql-cdc，没有随大版本改名，这点容易记错。
SQL 方式的取舍很清楚：写起来快，适合做字段裁剪、简单过滤、多表 join 这类纯 SQL 能表达的转换；但一旦要做复杂序列化、调用外部服务补数据，就得退回 DataStream。另外 SQL 方式在 3.0 里仍然是「单表配置」，整库同步要写一堆建表语句，这时候就该上 Pipeline 了。
4.5 YAML Pipeline：MySQL 整库同步到 Doris 的完整配置
3.0 的 Pipeline 模式把整库同步变成了改配置文件。在 Flink CDC 安装目录下建一个job文件夹，写一份mysql-to-doris.yaml。
source: type: mysql hostname: node03 port: 3306 username: root password: "Mysql@123" # 正则匹配整库，注意反斜杠转义 tables: test\.* # 伪装成从库的 server-id 区间，不能和已有从库冲突 server-id: 5400-5404 server-time-zone: UTC+8 sink: type: doris fenodes: node01:8030 username: root password: "Mysql@123" # 源表加列时走轻量 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: 1配置分三块。source里的tables是整库同步的核心，test\.*匹配test库下所有表，注意点号要转义，写成test.*会被当成正则通配。server-id给一段范围而不是单值，是因为每张表会分配一个独立的读取实例，范围太小会报「server id 被占用」。server-time-zone要和源库时区一致，不一致会导致时间类型字段偏移 8 小时，这个坑很隐蔽。
sink里的fenodes指的是 Doris FE 的 HTTP 端口 8030，不是 MySQL 协议端口 9030，写错会报连接超时。parallelism在 pipeline 层指定，比在 Flink 集群层面调更精细。
启动命令只有一行，前提是 Flink 集群的 Checkpoint 已经配好、connector jar 已经放进 lib。
# 提交 pipeline 作业 bin/flink-cdc.sh job/mysql-to-doris.yaml作业跑起来后，去 Doris 里show tables from test应该能看到 t1、t2、t3 三张表被自动创建，数据也已经灌进去。接着在 MySQL 里对 t1 做一次 update、对 t2 做一次 delete、给 t3 加一列，再刷新 Doris，这三次操作分别验证了更新同步、删除同步和表结构变更同步。
4.6 路由变更：把 test_route 落到 doris_test_route
默认情况下 Pipeline 会保持库名表名一致地同步过去。要把test_route库落到doris_test_route，靠 route 规则。
source: type: mysql hostname: node03 port: 3306 username: root password: "Mysql@123" tables: test_route\.* server-id: 5400-5404 server-time-zone: UTC+8 sink: type: doris fenodes: node01:8030 username: root password: "Mysql@123" table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 route: # 源表 -> 目标表，支持正则 - source-table: test_route\.t.* sink-table: doris_test_route.t.* pipeline: name: Route MySQL to Doris parallelism: 1route段的source-table和sink-table都支持正则表达式，写test_route\.t.*到doris_test_route.t.*表示「库内所有 t 开头的表，平移到另一个库」。验证方式很简单：同步完成后 Doris 的doris_test_route.t2里应该有 1004、1005、1006 三条，而test.t2里没有——因为这条数据只存在于test_route库。
路由的价值在于做环境隔离和多租户分流：不同业务库同步到 Doris 的不同 database，或者把分库分表的多张源表汇聚成一张宽表。
5. 避坑与排查：这五处配置不改，任务跑不过十分钟
5.1 启动阶段：连接、类缺失与权限
现象：作业提交后立刻失败，日志里是java.lang.ClassNotFoundException: com.ververica.cdc.connectors.mysql.source.MySqlSource。原因：打的是瘦包，或者 connector jar 没放进$FLINK_HOME/lib。Pipeline 模式下最常见——把flink-cdc-pipeline-connector-*.jar放进了 Flink CDC 自己的 lib 目录，但真正拉起作业的是 Flink 集群。解决：DataStream 用jar-with-dependencies打包；Pipeline 把两个 pipeline connector jar 复制到$FLINK_HOME/lib，重启集群。
现象：报Access denied for user，或者能连上但读不到数据。原因：MySQL 用户没有REPLICATION SLAVE和REPLICATION CLIENT权限。CDC 需要伪装成从库读 Binlog，普通账号不够。解决：GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user'@'%'; FLUSH PRIVILEGES;。另外 MySQL 8.0 默认用caching_sha2_password，某些老驱动不支持，必要时间改成mysql_native_password。
现象：报no viable alternative at input或者连不上 FE。原因：Doris 的fenodes写成了 9030。8030 是 FE 的 HTTP 端口，9030 是 MySQL 协议端口，sink 用的是前者。解决：改成node01:8030，多 FE 用逗号分隔。
5.2 运行阶段：Checkpoint、重复消费与写入失败
现象：作业跑着跑着 Checkpoint 全部超时，最后触发重启。原因：状态存储指向 HDFS 但没配HADOOP_USER_NAME，或者 Checkpoint 目录权限不对；另一种是全量阶段数据量大，快照做的 barrier 对齐被拖慢。解决：先hdfs dfs -chmod -R 777 /flink-cdc/ck排除权限问题；再把execution.checkpointing.timeout从默认 10 分钟适当放大；超大表分批接入，不要一次同步几十张。
现象：Doris 里出现重复数据，或者主键重复报错。原因：源表没有主键，Pipeline 无法生成 upsert 语义，只能按 append 写入；或者 sink 端的表模型和源表不匹配。解决：优先给源表加主键；实在加不了，在 Doris 侧建 Unique Key 模型表并手工指定 key 列，同时评估数据去重策略。
现象：源表加了一列，Doris 同步任务直接卡住或者报 schema 不兼容。原因：light_schema_change没开，Doris 走的是重量级 Schema Change，期间表不可写，同步任务阻塞。解决：建表时就把table.create.properties.light_schema_change: true加上。已经建了表的，手工执行ALTER TABLE ... SET ("light_schema_change" = "true")后重新提交。加列操作本身的类型也要兼容，源表从VARCHAR(255)扩到VARCHAR(512)一般没问题，从INT改VARCHAR这种就会被拒。
注意：这三个运行期问题的共同点是「日志里能看到，但报错信息指向性很弱」。排查习惯是先把 Flink WebUI 的 Checkpoint 页面打开，看是超时还是失败，再去看 TaskManager 日志里具体的 exception，别一上来就怀疑配置写错。
6. 二次开发：把 demo 管道改成能上生产的形态
跑通 demo 和能上生产之间还隔着一层工程化改造。我一般按这几个方向动手。
参数外置。不要把主机名密码硬编码在 YAML 里。写一个小脚本，用环境变量渲染模板生成最终 job 文件，敏感信息走配置中心或启动时注入。
#!/usr/bin/env bash # gen_job.sh —— 从模板渲染出实际 job 文件 TEMPLATE=job/mysql-to-doris.yaml.tpl OUT=job/mysql-to-doris.yaml sed -e "s|__SRC_HOST__|${SRC_HOST}|g" \ -e "s|__SRC_PWD__|${SRC_PWD}|g" \ -e "s|__DORIS_FE__|${DORIS_FE}|g" \ -e "s|__DB__|${DB_NAME}|g" \ "$TEMPLATE" > "$OUT" # 提交 $FLINK_CDC_HOME/bin/flink-cdc.sh "$OUT"sed替换是最轻的做法，模板里把可变项写成__SRC_HOST__这类占位符。这样一套模板能生成 dev / staging / prod 三份配置，避免「改了测试环境忘了改生产」这类事故——我真见过因为一份 YAML 里replication_num忘了改，生产 Doris 建了一批副本不健康的表，查了半天才发现。
多表模板化与批量接入。整库同步的表数量上去之后，server-id范围要跟着扩，按「表数 + 2」预留。另外不是所有表都适合实时同步，日志表、流水表这类只增不改的，可以拆到另一条并行度更高的管道里，避免拖慢主链路的 Checkpoint。
监控埋点。Pipeline 模式没有自己的指标接口，只能从 Flink 侧拿。最低成本的方案是定时拉 Flink REST API 的/jobs/:jobid/checkpoints，看最近一次 Checkpoint 的持续时间和大小，持续变大说明有反压或者状态泄漏。再加一个数据一致性对账脚本，每天跑一次源表与目标表的行数和主键集合比对，比盯着 Checkpoint 更实际。
| 改造点 | 做法 | 验证方式 |
|---|---|---|
| 配置外置 | 模板 + 环境变量渲染 | 用测试环境变量跑一遍，确认无硬编码 |
| 敏感信息 | 启动时注入，不落盘 | grep -r password job/应无结果 |
| 多表接入 | 正则 + route 映射 | 抽样比对源表与目标表行数 |
| 状态监控 | Flink REST API 拉 CK 指标 | 连续 24 小时 CK 无超时 |
| 数据对账 | 每日主键集合比对 | 差异率为 0 才认为链路健康 |
验证方法上，我更信「破坏性测试」：故意在同步过程中 kill 掉 TaskManager，看作业能不能自己从 Checkpoint 恢复并且不丢不重。这个测试过不了，其他指标再好看都是纸面功夫。EXACTLY_ONCE 只覆盖到 Flink 内部状态，Sink 端要真正实现精确一次，还得靠 Doris 的 Stream Load 事务标签去重，这一点在跨版本升级时要特别留意。
从那以后我每次接一条新的实时同步链路，都会先把「全量阶段暂停、增量阶段 kill 一次、加列一次」这三件事走一遍，再谈上线。这三步能挡住八成的配置问题。希望帮到你。
本文还有配套的精品资源，点击获取
)