Flink CDC 3.0 MySQL到Doris实时同步配置与排错实战

简介：由尚硅谷研究院编写的 Flink CDC 3.0 实操讲义，面向具备一定 Flink 基础的大数据研发人员，聚焦变更数据捕获原理与 MySQL 到 Doris 的实时同步落地。内容从 CDC 概念切入，对比基于查询与基于 Binlog 两类方案在执行模式、延迟表现和数据库压力上的差异，再延伸到 flink-cdc-connectors 组件的能力边界；随后用案例串起数据准备、DataStream 与 Flink SQL 两种写法，并完整给出 Streaming ETL 的环境搭建、配置文件编写、任务启动与测试验证步骤，同时点明开启 Binlog、设置检查点等容易出错的配置项。整包仅 1 个 docx 文件，约 145KB，篇幅紧凑，命令与代码示例集中，便于按章节对照练习。已有 1789 人学习，适合希望独立完成实时同步链路部署、保障数据一致性的工程师作为案头参考。

1. 从一次只跑了三十秒的同步任务说起：Flink CDC 3.0 这套环境卡在哪

三台测试机，MySQL 8.0 里一张两万行的表，目标是把它实时同步进 Doris，链路用 Flink CDC 3.0。任务提交上去，三十秒后 Flink WebUI 里那个 Job 变成 FAILED，日志最后一行说读不到 binlog。查了半天，MySQL 的binlog_format还是默认值，server-id也没配。改完重启，任务能起来了，可 Doris 里只有新插入的数据，历史两万行一行没来——startupOptions写成了latest。

Flink CDC 3.0 把 MySQL 到 Doris 的实时同步压成了一份 YAML，看着简单，实际每个参数都咬着另一头：MySQL 的 binlog 开关、Flink 的 checkpoint、Doris 的 FE 端口和副本数，任何一头没对上，任务不是起不来就是悄悄丢数据。这篇记录的就是我在这套环境里从装包、配 YAML 到验证跑通的全过程，适合手上有一台 MySQL、一套 Doris、想把增量同步搭起来的人照着走。

2. 环境清点：Flink 1.18、CDC 3.0 与 MySQL 8.0 的版本对照

装之前先把版本钉死。这套链路上有四个东西会互相挑版本：Flink 本体、Flink CDC 的连接器、MySQL 驱动、Doris。版本错一个，报错信息往往指向别处，白折腾半天。我一般会先在一张纸上（现在是一个 Markdown 表）把要装的版本、端口、机器角色全列出来，装的时候照着对，出问题也好回溯。

2.1 版本对照表与选型理由

下面这张表是我在这套环境里实际跑通的组合，直接从项目文档里的依赖和实测输出里对齐出来的，不是随手写的：

Flink 选 1.18.0 而不是更新的版本，原因是 DataStream 那套依赖里flink-table-planner_2.12、flink-table-runtime、flink-table-api-java-bridge都跟着${flink-version}走，1.18 这条线跟 CDC 3.0 的 connector 组合是稳的；跨大版本升级时，table planner 的 Scala 后缀（_2.12）经常是第一个炸的地方，编译期就报找不到类。

MySQL 驱动必须显式引 8.0.31，别指望 Flink 自带的。CDC 连接器读 binlog 走的是它自己的协议实现，但 JDBC 部分（比如全量快照阶段）仍然要用驱动，版本低了会出现Unknown system variable或者时区解析异常。我一般会把mysql-connector-java和flink-connector-mysql-cdc的版本写死在 pom 里，不放进 dependencyManagement 让它自动继承。

提示：Flink 1.18 官方支持 JDK 8/11/17，但 CDC 3.0 的发行版脚本对 JDK 17 的模块化参数没做适配，稳妥起见统一 JDK 8，别在同一台机器上混着装两个 JDK 又不设JAVA_HOME。

2.2 MySQL 侧三件事：server-id、log-bin、binlog_format

CDC 读不到数据，九成栽在 MySQL 配置上。修改/etc/my.cnf，在[mysqld]段落下加这几行：

[mysqld] # 每个 MySQL 实例唯一，主从和 CDC 都靠它区分 server-id = 1 # 开启 binlog，值会作为 binlog 文件前缀，最终文件形如 mysql-bin.000001 log-bin = mysql-bin # 必须是 row，statement/mixed 模式下 CDC 拿不到行级变更 binlog_format = row # 只对这两个库记录 binlog，减少无谓的磁盘写入 binlog-do-db = test binlog-do-db = test_route # 记录整行前后镜像，UPDATE 时才能拿到完整字段 binlog_row_image = FULL

server-id这一项最容易被忽略，单机开发时它好像无所谓，但 CDC 连接器本质上是把自己伪装成一个从库去拉 binlog，Flink 侧还会在 YAML 里再声明一段server-id区间。两处冲突的表现是任务反复重连、日志里刷A slave with the same server_uuid/server_id。binlog-do-db也要留意：它按「当前默认库」判断，跨库更新（UPDATE test.t1 ...而当前库是别的）不会进 binlog，遇到「明明改了却没同步」先查这个。binlog_row_image项目文档没写，但这是我必加的一项，默认FULL时无所谓，有些环境被调成MINIMAL，UPDATE 出来的 JSON 里只有主键，下游 Doris 更新会丢字段。

改完必须重启 MySQL，systemctl restart mysqld或者按你环境的服务名来。重启后用下面两条命令确认生效，别改完就往下走：

-- 三个值必须分别是 ON / ROW / FULL SHOW VARIABLES WHERE Variable_name IN ('log_bin', 'binlog_format', 'binlog_row_image', 'server_id');

log_bin返回ON才说明 binlog 真的开了；binlog_format返回ROW；server_id返回你配的那个数字。如果log_bin还是OFF，常见原因是配置写进了[client]段、或者 MySQL 8 用了 include 目录而你没改对文件，mysqld --verbose --help | grep my.cnf能看到它实际加载的顺序。

2.3 三台机器的角色与端口规划

主机名我用node01/node02/node03，自己在/etc/hosts里对好 IP。这套分布是我在一台 16G 内存的机器上拆虚拟机时常用的最小拓扑：

Doris 这边有两个端口必须分清：9030是 MySQL 协议端口，你用mysql -P9030连的是它；8030是 FE 的 HTTP 端口，fenodes要填的是这个。我在 YAML 里曾经照着旧笔记填了7030，任务起来后卡在connect timed out，翻fe.conf里的http_port才发现对不上。所以装之前先把 FE 的实际 HTTP 端口确认一遍，写在纸上。

目录规划也顺手定好，后面所有命令都基于它：Flink 放/opt/module/flink-1.18.0，CDC 发行版解压到/opt/module/flink-cdc-3.0.0，MySQL 数据目录不动，Doris 的 FE/BE 各自在自己的安装目录下。CDC 发行版单独一个目录，别和 Flink 混着装——它的lib/里塞的是 pipeline connector，跟 Flink 的lib/不是一回事，混在一起会出现类冲突。

2.4 把测试库和初始数据先造出来

环境清点完，数据先造好，后面验证才有对照物。登录 MySQL，建两个库，各建三张结构近似但有一处差异的表——t3用sex而不是name，这是故意的，后面测路由和加列时用得上：

-- 源库 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'); -- 路由测试库，主键刻意错开，方便看出是哪张表同步过来的 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 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');

test和test_route两个库的表结构一致、数据部分重叠，是为了后面验证「同一个 MySQL 实例、不同库同步到 Doris 不同库」时能一眼看出串没串库。主键统一用VARCHAR(255)而不是自增 INT，也是刻意的：CDC 的 JSON 输出里主键是字符串，如果你下游建表时写成 INT，Doris 侧会因为类型不匹配把整条记录丢掉，而且报错藏在 BE 的日志里，不明显。

这些准备动作做完，MySQL 侧就算齐了。接下来才是真正装 Flink CDC 3.0。

3. 装 Flink CDC 3.0：从 tar 包到 pipeline connector 落进 lib

Flink CDC 3.0 有两种用法，装法完全不同，必须先想清楚用哪种。一种是 DataStream / FlinkSQL 方式，把flink-connector-mysql-cdc当普通 Maven 依赖引到自己的工程里，打成 fat jar 提交给 Flink 集群；另一种是 3.0 主推的 Pipeline 方式，用官方发行版里的bin/flink-cdc.sh直接提交一份 YAML，连接器 jar 提前放进发行版的lib/。两种方式可以共存，但装的东西不一样。

3.1 解压发行版并核对目录结构

先装 Flink 1.18，再解压 CDC 发行版：

# 解压 Flink，放到统一目录 tar -zxvf flink-1.18.0-bin-scala_2.12.tgz -C /opt/module/ # 解压 CDC 发行版，单独一个目录 tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ # 看下解压出来的结构，lib 和 bin 是重点 ls -l /opt/module/flink-cdc-3.0.0/

bin/下只有flink-cdc.sh一个入口脚本，lib/下是发行版自带的若干 jar（包含提交任务用的客户端和基础依赖），conf/里放日志配置。这个发行版不包含 Flink 运行时，它只是把你的 YAML 翻译成一个 Flink Job 然后提交给你指定的集群，所以 Flink 集群必须先起来。我在解压后习惯先跑一次bin/flink-cdc.sh --help，能打出参数说明就说明 JDK 和脚本权限都没问题；如果报JAVA_HOME is not set，先export JAVA_HOME=/usr/local/jdk8再试。

给脚本加执行权限，这一步别省：chmod +x bin/flink-cdc.sh。有些解压工具会丢掉权限位，表现就是Permission denied，看着像脚本坏了，其实就是没加 x。

3.2 两个 pipeline connector 落进 lib 并校验

Pipeline 模式下，MySQL source 和 Doris sink 是两个独立的连接器 jar，必须手动放进lib/：

# 两个 jar 拖进 lib，注意版本号要对齐发行版 cp flink-cdc-pipeline-connector-mysql-3.0.0.jar \ /opt/module/flink-cdc-3.0.0/lib/ cp flink-cdc-pipeline-connector-doris-3.0.0.jar \ /opt/module/flink-cdc-3.0.0/lib/ # 确认两个 jar 都在，且没有同名不同版本的残留 ls /opt/module/flink-cdc-3.0.0/lib/ | grep -E 'mysql|doris'

ls输出里应该正好两行。如果出现mysql-1.4.0之类的旧包，一定要删掉——CDC 加载连接器时按type字段匹配，两个版本同时存在时先加载哪个不确定，表现是参数不认识或者连不上库，非常难查。我第一次装的时候 lib 里躺着上一版留下的connector-mysql，YAML 里写server-time-zone一直报「Unsupported options」，删掉旧包立刻就好了。

注意：Doris connector 的 jar 不要顺手也丢进 Flink 的lib/。Pipeline 任务是跑在 Flink 集群里的，但连接器由 CDC 客户端分发，重复放置会出现同一类被两个 ClassLoader 加载，抛LinkageError。

3.3 DataStream 方式的 Maven 依赖怎么写

如果你的场景需要在流里做转换，而不是原样搬运，就得走 DataStream。pom 里的关键几项：

<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 核心，provided 与否看你提交方式，本地跑就 compile --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink-version}</version> </dependency> <!-- FlinkSQL 方式必须引 table 三件套，Scala 后缀跟集群一致 --> <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 的 MySQL source，包名是 com.ververica 不是 org.apache.flink --> <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>

打包插件用maven-assembly-plugin的jar-with-dependencies，否则提交到集群会报ClassNotFoundException: com.ververica.cdc.connectors.mysql.source.MySqlSource。flink-connector-base也要引上，1.18 里 Source API 的基础类在这个包里，漏了编译期就过不去。

flink-table-planner_2.12的下划线后缀要和集群的 Scala 版本对上。集群是flink-1.18.0-bin-scala_2.12，这里就必须是_2.12，写成_2.11编译能过但运行时抛NoSuchMethodError，属于典型的「编译期不炸运行期炸」。

3.4 DataStream 代码骨架：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); // 任务取消时保留最后一次 CK，否则恢复时只能从头再来 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 3. 状态后端与 CK 存储位置 env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://node01:8020/flinkCDC"); System.setProperty("HADOOP_USER_NAME", "atguigu"); // 4. 构造 MySQL Source MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("node01") .port(3306) .databaseList("test") .tableList("test.t1") .username("root") .password("000000") .deserializer(new JsonDebeziumDeserializationSchema()) // initial：先全量快照，再追增量，最常用的模式 .startupOptions(StartupOptions.initial()) .build(); // 5. 接入流并打印 env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQLSource") .print(); env.execute(); } }

enableCheckpointing(3000L, EXACTLY_ONCE)里的 3000 是间隔毫秒数，别设太小，HDFS 上小文件会爆炸；也别设太大，任务挂了重来的增量就多。我做实时同步一般 5 秒到 10 秒。RETAIN_ON_CANCELLATION是必开的，不然你在 WebUI 点一下 Cancel，CK 目录被清空，想从 Savepoint 恢复就没了。checkpointStorage指向 HDFS 而不是本地，是因为 TaskManager 在 node03 上，本地路径恢复时找不到。

StartupOptions有五种，我按场景选：

我第一次跑用的是latest()，结果 Doris 里只有新数据，被问「历史数据呢」的时候才发现问题。首次接入无脑用initial()，这是最省事的。

3.5 FlinkSQL 方式：一条 CREATE TABLE 就够

不想写 Java 的话，FlinkSQL 方式更短，建一张逻辑表指向 MySQL：

CREATE TABLE t1 ( id STRING PRIMARY KEY NOT ENFORCED, name STRING ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'node01', 'port' = '3306', 'username' = 'root', 'password' = '000000', 'database-name' = 'test', 'table-name' = 't1' ); SELECT * FROM t1;

PRIMARY KEY NOT ENFORCED里的NOT ENFORCED不能省，这是 FlinkSQL 的语法要求，声明主键但不做校验。connector的值是固定的mysql-cdc，写错成mysql会走到 JDBC connector 上，那是批式查询，完全不是 CDC。这个方式不需要 YAML，也不需要往 lib 里塞 pipeline jar，但它跑在 Flink 集群里，所以 Flink 的 checkpoint 配置照样要开——否则默认行为是每次重启从头快照，数据会重复。

装到这里，两种模式都能跑了。接下来是 3.0 最核心的部分：那份 YAML。

4. 写 mysql-to-doris.yaml：source、sink、pipeline 三段怎么填

管道模式的全部配置就一个 YAML 文件，分三段：source说从哪儿读，sink说往哪儿写，pipeline说这个任务叫什么、开几个并行度。文件放在 CDC 发行版目录下的job/里，我习惯按「来源-目标」命名，比如mysql-to-doris.yaml，日后多了也好认。

mkdir -p /opt/module/flink-cdc-3.0.0/job cd /opt/module/flink-cdc-3.0.0/job vim mysql-to-doris.yaml

4.1 source 段：tables 正则和 server-id 区间

source: type: mysql hostname: node01 port: 3306 username: root password: "000000" # 正则匹配要同步的表，test 库下所有表 tables: test.\.* # CDC 伪装成从库时用的 server-id 区间，不能和 MySQL 自身冲突 server-id: 5400-5404 # 会话时区，不写会按 UTC 解析，时间字段差 8 小时 server-time-zone: UTC+8

tables是正则，test.\.*里那个反斜杠是转义点号，意思是「库名 test，表名任意」。想只同步部分表就写test.t1,test.t2，逗号分隔。写test.*也能跑，但.在正则里匹配任意字符，理论上会把testX也吃进来，严谨点还是转义。

server-id: 5400-5404这个区间是给连接器内部用的，它会从这个范围里挑一个作为自己的 slave id。必须保证：一，不和 MySQL 配置里的server-id = 1重叠；二，同一套环境里多个 CDC 任务的区间不重叠。我见过两个任务都写5400-5404，第二个起来就反复重连，日志刷server id conflict。多个任务就错开，比如5400-5404、5410-5414。

server-time-zone: UTC+8是我踩过坑才加的。不写的时候，Debezium 把 MySQL 的DATETIME当 UTC 解析，Doris 里落下来的时间比实际晚 8 小时；如果 Doris 建表用的是DATETIME，不会有任何报错，就是数据不对，属于最难发现的那类问题。MySQL 服务端是+08:00就一定写上UTC+8。

4.2 sink 段：fenodes 端口和两个建表属性

sink: type: doris # FE 的 HTTP 端口，不是 9030 那个 MySQL 协议端口 fenodes: node01:8030 username: root password: "000000" # 允许 Doris 做轻量 schema change，源端加列时下游能自动跟着加 table.create.properties.light_schema_change: true # 副本数，单 BE 环境必须为 1，否则建表一直 PENDING table.create.properties.replication_num: 1

fenodes填的是 FE 的 HTTP 端口。这里我要专门说一句：这个端口在fe.conf里叫http_port，默认 8030，9030是 query port，你用 MySQL 客户端连的是它。填错的表现是任务启动时connect timed out，或者更隐蔽一点，能连上但后续所有建表、写入操作都超时。验证办法很简单：curl http://node01:8030/api/bootstrap能返回 JSON 就对了。

table.create.properties.*这两个属性会拼进 Doris 的建表语句里。light_schema_change开启后，源端ALTER TABLE ADD COLUMN，Doris 侧能跟着加，不用重建表；不开的话，下游遇到新列会报 schema 不匹配，任务直接挂。replication_num在单 BE 环境必须设 1，Doris 默认副本数是 3，只有一个 BE 时建表语句会一直卡在 PENDING，你在 Doris 里SHOW TABLES看到表名但查不出数据，SHOW PROC '/dbs/...'里能看到副本状态不对。

4.3 pipeline 段与 Flink 侧 checkpoint 开关

pipeline: name: Sync MySQL Database to Doris # 并行度，先 1 跑通再往上加，加之前确认 MySQL 的 server-id 区间够用 parallelism: 1

name会出现在 Flink WebUI 的 Job 名里，多个任务并用时一定要改得能区分开。parallelism不是随便加：每个并行子任务都要占用一个server-id，server-id: 5400-5404只有 5 个名额，并行度超过 5 就会报「no available server id」。要提并行度就同步把区间放宽，比如5400-5420配parallelism: 16。

YAML 写完还没完，Flink 侧必须开 checkpoint，否则 pipeline 任务跑一段时间就重启、数据重复写：

# conf/flink-conf.yaml，改成你的 Flink 安装目录 execution.checkpointing.interval: 5000

这个文件要分发到所有 Flink 节点（node02和node03），只改一台不生效，因为 JobManager 和 TaskManager 都会读它。改完重启集群。我一般顺手再确认state.backend和state.checkpoints.dir，默认值在本地文件系统上，生产环境指到 HDFS 更稳。

4.4 路由变更：同一个 YAML 换个 tables 就串到另一个库

项目里还有一个「路由变更」的用法，意思是从 MySQL 的test_route库同步，落到 Doris 的另一个库，让两边数据不串。做法很简单，复制一份 YAML 改三处：

source: type: mysql hostname: node01 port: 3306 username: root password: "000000" tables: test_route.\.* # 改这里，指向 test_route 库 server-id: 5410-5414 # 换一段区间，和上一个任务错开 server-time-zone: UTC+8 sink: type: doris fenodes: node01:8030 username: root password: "000000" # 关键：指定落到 Doris 的哪个库 table.create.properties.database: doris_test_route table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 pipeline: name: Sync MySQL test_route to Doris parallelism: 1

table.create.properties.database这个键在文档里没明写，但 Doris connector 支持透传建表属性，库里没有对应名称时会自动建。这里最值得注意的是server-id必须换区间，两个任务如果在同一套 MySQL 上跑，区间重叠就是互相踢下线的结果。Doris 侧记得先CREATE DATABASE doris_test_route，或者依赖自动建库，我倾向手动建，出问题时边界清楚。

5. 排查清单：这套链路翻车的五个典型现场

装和配的过程里，报错集中在五种。下面按「现象 → 原因 → 解决」写，都是我实际遇到过的原样记录，遇到时照这个顺序查，比看日志猜快得多。

5.1 现象：任务能起，但 Doris 里只有任务启动之后的新数据

原因是startupOptions用了latest()，或者 YAML 的 source 没有配全量快照相关的参数，连接器直接跳过历史数据从 binlog 末尾读。这个错误不会报，任务状态一直是 RUNNING，很容易误判成「已经同步完了」。

解决：DataStream 代码里改成StartupOptions.initial()；Pipeline 模式（3.0 的 YAML）默认行为就是先快照后增量，如果你在 source 段显式写了scan.startup.mode: latest-offset，删掉它。改完重跑前先把 Doris 里已经写进去的数据清掉，或者换一个空库，否则你会看到「历史数据 + 重复的新数据」搅在一起，更难判断。验证方式是对比行数：MySQL 里SELECT COUNT(*)和 Doris 里同表的一致，才算快照跑完。

5.2 现象：日志刷「server id conflict」或任务反复重连 MySQL

原因是同一套 MySQL 上跑了多个 CDC 任务，server-id区间重叠；或者 YAML 里的区间和 MySQLmy.cnf里配的server-id撞了。连接器把自己注册成从库，MySQL 发现 id 重复就把先来的踢掉，两边互相踢，日志里就是不停重连。

解决：给每个任务分配不重叠的区间，我一般按 10 递增：任务 A 用5400-5404，任务 B 用5410-5414，任务 C 用5420-5424。同时确认 MySQL 的server-id = 1不落在任何区间内。改完需要在 WebUI 里 cancel 旧任务再重新提交，别用 restart，server-id 是启动时注册的，重启恢复还是老值。

5.3 现象：Doris sink 一直connect timed out，或者建表卡 PENDING

原因是fenodes填的端口不对。填9030（MySQL 协议端口）会连上然后行为异常，填一个不存在的端口直接超时。另外replication_num大于实际 BE 数量时，建表语句能发出去但永远不成功，Doris 里看得到表、查不到数据。

解决：先curl http://node01:8030/api/bootstrap确认 HTTP 端口通；单 BE 环境把table.create.properties.replication_num设为1；确认 BE 已经ALTER SYSTEM ADD BACKEND注册并且SHOW BACKENDS里Alive是 true。BE 没起来的时候建表也会卡住，别只顾着查配置，先看一眼bin/start_be.sh那台机器进程在不在。

5.4 现象：任务跑十几分钟自己重启，Doris 里出现重复行

原因是 Flink 集群没开 checkpoint。Pipeline 的 Doris sink 依赖 checkpoint 完成两阶段提交，没有 CK，事务一直挂着，Flink 的默认重启策略触发后从头重放，重复数据就进去了。

解决：conf/flink-conf.yaml里加execution.checkpointing.interval: 5000，分发到所有节点并重启集群。DataStream 方式则是在代码里env.enableCheckpointing(...)。改完观察 WebUI 的 Checkpoints 页签，看到有 completed 记录才算生效。这个问题最阴的地方是「看起来一切正常」，等你发现重复数据时已经积了几十万行。

5.5 现象：源端加了一列，同步任务直接挂掉

原因是 Doris 侧没开light_schema_change，新列在目标表里不存在，写入时列数对不上。另一种表现是任务没挂但新列的数据全是 NULL。

解决：sink 段加table.create.properties.light_schema_change: true，并且 Doris 版本要支持（1.2 系列是支持的）。已经建好的表不会自动补上这个属性，需要ALTER TABLE ... SET ("light_schema_change" = "true")或者干脆删表重同步。我一般的做法是首次建表前就把这个属性配上，别等出事了再补。

6. 装完怎么验证：三条命令确认改动真的落到 Doris

任务提交上去显示 RUNNING 不算跑通，得看到数据动。我的验证顺序固定三步，缺一步都不算完。

第一步，提交任务并确认 Flink 侧真的有这个 Job 在跑：

# 在 CDC 发行版目录下提交 cd /opt/module/flink-cdc-3.0.0 bin/flink-cdc.sh job/mysql-to-doris.yaml

脚本会打印一串日志，最后给出 JobID 和 Flink WebUI 地址。拿这个 JobID 去http://node02:8081上对，状态应该是 RUNNING，Checkpoints 页签里能看到定时完成的记录。如果 Job 列表里没有，说明提交阶段就失败了，往上看脚本输出的异常栈，通常是 lib 里缺 connector jar 或者 YAML 缩进错了。

第二步，在 Doris 侧确认库表和全量数据到位：

# 9030 是 Doris 的 MySQL 协议端口，不是 fenodes 的 8030 mysql -uroot -p000000 -P9030 -hnode01 mysql> SHOW DATABASES; mysql> USE test; mysql> SHOW TABLES; mysql> SELECT COUNT(*) FROM t1;

SHOW TABLES里应该看到t1、t2、t3三张表，是连接器按源端结构自动建的。COUNT(*)要和 MySQL 里test.t1的行数一致，这是全量快照完成的标志。如果表在但行数是 0，回看第 5.1 条，大概率是启动模式不对。

第三步，制造一次变更，看它有没有实时落到 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'; ALTER TABLE t1 ADD COLUMN age INT;

三条语句分别验证 INSERT、UPDATE、DELETE，最后一条验证 schema 变更。每执行一条，就去 Doris 里SELECT * FROM t1看一眼，正常的话 INSERT 立刻可见、UPDATE 的name变了、DELETE 那行没了、age列被自动加出来。这四步都能对上，才说明全量 + 增量 + DDL 这条链路真的通了。哪一步没反应，就去看 TaskManager 的日志，tail -f log/flink-*-taskexecutor-*.out，CDC 的读取异常基本都在那里。

我把这套验证流程固定成习惯，是从一次事故之后开始的：那次任务显示 RUNNING，我在 Doris 里随手查了一下有数据，就交付了，两天后业务方问「昨天改的单子怎么没同步」，才发现 checkpoint 根本没开，任务一直重启，看着有数据其实是反复重放的旧快照。从那以后我每次装完这套环境，都强制走一遍「提交 → 对数 → 制造变更」这三步，少一步都不算完。希望帮到你。

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