简介：尚硅谷研究院编写的 Flink CDC 3.0 教程文档，面向具备一定 Flink 基础、需要搭建实时数据同步链路的大数据研发人员。内容先从 CDC（变更数据捕获）概念切入，对比基于查询与基于 Binlog 两类方案在执行模式、变化捕获能力、延迟与数据库压力上的差异，再引出 flink-cdc-connectors 组件的用法。案例部分完整覆盖 MySQL 库表与测试数据准备、DataStream 与 Flink SQL 两种接入方式，以及 MySQL 到 Doris 的 Streaming ETL 流程，含环境搭建、配置文件编写、任务启动与结果验证，并点明开启 Binlog、设置检查点等关键细节。

资源包内为 1 个 docx 文件，约 145KB，即上述图文教程全文，可按章节顺序阅读，也可在实操时逐节对照。已有 1789 人学习下载。

读者可借此独立完成从数据源接入、数据流处理到目标库写入的部署，理解各项配置参数的实际作用，并把同步方案迁移到自身业务场景中。

1. Flink CDC 3.0 落地：MySQL 一行变更到 Doris 报表之间缺的那一段

业务库里改了一行订单状态，下游报表要等第二天凌晨才变——这个落差几乎每个做数仓的人都撞过。手上这份 Flink CDC 3.0 的资料，给的就是把这段路补上的完整链路：先用两章把变更数据捕获的两条技术路线讲清楚，再用 DataStream 和 FlinkSQL 各跑一遍 MySQL 到控制台的最小闭环，最后用 3.0 才有的 pipeline 配置文件，把 MySQL 整库直接同步进 Doris。它适合已经能起 Flink 集群、手上有 MySQL 业务库、正在被 T+1 报表拖着走的工程师，也适合刚接触 CDC、想照着命令一行行抄一遍的新手。材料是命令行、配置项、代码混排的形态，没有多余的架构图，我照着跑了两轮，把其中容易翻车的地方单独记了下来。

2. 选型先立住：基于查询与基于 Binlog 的 CDC 差在哪三条线上

2.1 两种捕获模式的执行模型对比

CDC 的核心动作只有一句话：盯住数据库里数据行和表的增删改，把每一次变更按发生顺序记下来，再送给下游订阅消费。听着简单，落到实现上分成两条完全不同的路：一条是定时拿 SELECT 去问数据库「你变了没有」，另一条是让采集程序伪装成一台从库，直接读 MySQL 的 Binlog 事件流。

这两条路的差别不是快慢，而是能不能看见完整的变更历史。基于查询的方案靠时间戳字段或者自增 ID 捞增量，两次查询之间发生的中间状态会被压平，DELETE 语句更是彻底看不见——行没了，你只知道少了，不知道少的是哪一行、什么时候少的。

选型的判断标准其实很朴素：目标端的表如果要求跟源库严格一致，包括删除，那就只有 Binlog 这条路；如果只是每天跑一次全量覆盖、下游能容忍延迟，查询式方案成本更低，没必要上流式。资料里把这条对比放在第 1 章，不是凑篇幅，是因为后面所有的配置复杂度——开 Binlog、配 server-id、开检查点——根子都在这个选择上。

2.2 为什么 MySQL 侧必须开 row 格式的 Binlog

Flink CDC 的 MySQL 连接器底层用的是 Debezium 的引擎，Debezium 解析的是行级变更事件，所以binlog_format必须是row。statement格式只记 SQL 原文，遇到NOW()、UUID()这类函数时，从库重放出来的值跟主库不一致；mixed格式在语句和行之间来回切，解析端要处理的边界情况陡增。这不是性能取舍，是能不能解析出来的问题。

binlog-do-db用来限定只对指定库记日志。它的语义是按「当前默认库」匹配，不是按「被操作的表所属库」匹配，所以跨库语句容易踩空。我一般会把它当成白名单来用，把要同步的库逐条列进去，同时把 binlog 的保留时长调大——binlog_expire_logs_seconds默认值在不少环境里只有几天，任务停一个周末回来，位点就过期了，这是最典型的「重启即全量」事故来源。

server-id是另一处必须提前想清楚的地方。连接器读 binlog 时是以从库身份连上去的，MySQL 主库要求每个从库的server-id唯一。单并行度时连接器会自动生成一个，一旦你把 Flink 任务的并行度调到 2 以上，多个 reader 各自要一个 id，就得手动给一段范围。

2.3 把 MySQL 环境按资料里的参数配好

资料里准备了两套库test和test_route，各建三张表，这个设计的用意是后面验证路由——两张同名表数据刻意错开，同步完一眼就能看出有没有串库。

# 编辑 MySQL 配置文件，开启 Binlog sudo vim /etc/my.cnf # 在 [mysqld] 段落下追加 # server-id：集群内唯一，供 CDC 连接器以从库身份注册 server-id = 1 # log-bin：开启 binlog 并指定文件前缀 log-bin = mysql-bin # binlog_format：必须为 row，Debezium 只能解析行级事件 binlog_format = row # binlog-do-db：白名单，逐条列出需要捕获的库 binlog-do-db = test binlog-do-db = test_route # 建议追加：binlog 保留 7 天，避免任务停机后位点丢失 binlog_expire_logs_seconds = 604800

改完配置重启 MySQL，然后用两条命令自检，确认参数真的生效而不是被 my.cnf 里别处的重复项覆盖：

-- 逐项确认关键参数 SHOW VARIABLES LIKE 'log_bin'; -- 期望 ON SHOW VARIABLES LIKE 'binlog_format'; -- 期望 ROW SHOW VARIABLES LIKE 'server_id'; -- 期望与配置一致 SHOW VARIABLES LIKE 'binlog_expire_logs_seconds';

建库建表这一步，资料里给的结构很简单，两张同步表加一张只有sex字段的表，主要是为了演示多表一起同步：

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)); -- 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 的 t2 主键区间不同 INSERT INTO test.t1 VALUES ('1001','zhangsan'),('1002','lisi'),('1003','wangwu'); INSERT INTO test.t2 VALUES ('1001','zhangsan'),('1002','lisi'),('1003','wangwu'); INSERT INTO test.t3 VALUES ('1001','F'),('1002','F'),('1003','M'); INSERT INTO test_route.t1 VALUES ('1001','zhangsan'),('1002','lisi'),('1003','wangwu'); INSERT INTO test_route.t2 VALUES ('1004','zhangsan'),('1005','lisi'),('1006','wangwu'); INSERT INTO test_route.t3 VALUES ('1001','F'),('1002','F'),('1003','M');

这里有个细节值得留意：id用VARCHAR而不是自增整型。好处是造数据时能精确控制，坏处是位移和分片时无法按主键区间切。实际项目里主键几乎都是自增 BIGINT，Flink CDC 会按主键区间做快照切分，这种情况下的快照并发能力比字符串主键好得多。

3. DataStream 方式跑通：依赖、Source 与检查点三件套

3.1 依赖矩阵：Flink 1.18.0 与 connector 3.0.0 怎么对齐

依赖表看着长，真正决定能不能跑起来的只有几项：Flink 的主版本、Scala 后缀、连接器版本。资料里用的是 Flink 1.18.0 配flink-connector-mysql-cdc3.0.0，表规划器用的是flink-table-planner_2.12，这个_2.12就是 Scala 版本后缀，跟你集群发行包必须一致，写成_2.11会在类加载阶段直接报找不到表工厂。

<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> <!-- 编译期需要，提交到集群时由集群提供，故用 provided --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink-version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink-version}</version> <scope>provided</scope> </dependency> <!-- FlinkSQL 路线必需，注意 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-api-java-bridge</artifactId> <version>${flink-version}</version> </dependency> <!-- CDC Source，groupId 与 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> <build> <plugins> <!-- 用 assembly 打成带依赖的胖包，避免集群缺类 --> <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>

scope这一栏最容易出问题。如果把flink-streaming-java打成 provided，本地 IDE 跑不起来是正常的，因为 IDE 不会自动补集群的类路径；反过来，如果不用 provided 而把 Flink 全家桶打进胖包，提交到集群后会出现同一份 Flink 类被加载两次，典型症状是IllegalStateException: Trying to access closed classloader这类莫名其妙的报错。我的习惯是：Flink 核心和 table 规划器一律 provided，CDC 连接器和 JDBC 驱动打进去，然后本地测试时用 IDEA 的 provided 依赖勾选项临时打开。

3.2 MySqlSource 的 builder 参数与五种启动模式

MySqlSource是整个 DataStream 路线的入口，它的 builder 上有十来个参数，其中真正决定行为的是启动模式。资料里把五种模式的英文注释原封不动贴了出来，这段注释比正文更值得读——它决定了任务第一次启动时到底从哪儿开始读。

新手最容易在这里翻车：以为latest会先同步一遍存量数据，结果任务跑起来下游一张空表，还以为是连接没通。latest的语义就是「装作我刚来，之前的事我不管」，历史数据必须靠initial或者另外跑一次全量。

MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("node01") // MySQL 地址 .port(3306) .databaseList("test") // 捕获的库，可传多个 .tableList("test.t1") // 捕获的表，格式 db.table，支持正则 .username("root") .password("******") // 反序列化器：把 Debezium 的 SourceRecord 转成 JSON 字符串 .deserializer(new JsonDebeziumDeserializationSchema()) // 首次启动：先全量快照，再持续读 binlog .startupOptions(StartupOptions.initial()) .build();

databaseList和tableList要配合着写。只写databaseList("test")会捕获整库所有表，tableList里如果再写test.t1，就只保留 t1——两者是交集关系。tableList用的是正则匹配，写test\\.t.*就能把 t1、t2、t3 一起捕获。生产上我一般显式列表，宁可多写几行，也不要用过宽的正则把无关表卷进来，特别是带大字段的日志表。

还有一个不在 builder 表面但影响很大的参数是serverId。并行度大于 1 时，每个快照分片都要独立注册一个从库身份，必须给它一段范围。不显式指定时，连接器会在默认区间里随机取值，多任务并行跑在同一个 MySQL 实例上就有概率撞车，报错信息是「A slave with the same server_uuid/server_id as this slave has connected to the master」，看着像玄学，其实就是 id 重了。

3.3 Checkpoint 配置与断点续传的完整代码

Flink CDC 把读到的 binlog 位点信息存在状态里，而状态只有在检查点成功时才会持久化。不开检查点，任务一重启就是从零开始——这不是配置项的问题，是整条链路能不能续上的问题。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 每 3 秒做一次检查点，语义为精确一次 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); CheckpointConfig ckConfig = env.getCheckpointConfig(); ckConfig.setCheckpointTimeout(60 * 1000L); // 单次 CK 超时 1 分钟 ckConfig.setMinPauseBetweenCheckpoints(3000L); // 两次 CK 最小间隔，防止拖垮 ckConfig.setMaxConcurrentCheckpoints(1); // 同一时刻只允许一个 CK // 任务被 cancel 时保留最后一次 CK，这是后续 -s 恢复的前提 ckConfig.enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); ckConfig.setCheckpointStorage("hdfs://node01:8020/flinkCDC"); // 失败率重启：1 天内最多 3 次，每次间隔 1 分钟 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); // 状态后端与 HDFS 访问身份 env.setStateBackend(new HashMapStateBackend()); System.setProperty("HADOOP_USER_NAME", "operator");

几个参数值得单独说。setMinPauseBetweenCheckpoints设成跟间隔一样大，是为了防止大状态下检查点写不完就叠下一个，堆积起来会把 TaskManager 的 IO 打满。RETAIN_ON_CANCELLATION是断点续传的后悔药，不开启的话任务一 cancel，HDFS 上的检查点目录会被清掉，想恢复也没得恢复。HADOOP_USER_NAME这一行看着像补丁，但在没配 Kerberos 的测试集群里，不设它经常报权限不足，写 HDFS 直接失败。

检查点间隔也别无脑往小里调。3 秒一次对演示环境没问题，但同步链路本身有延迟，检查点太密会让位点推进速度跟不上数据量，反而拖慢整体吞吐。我一般按数据量给 10 秒到 1 分钟，宁可恢复时多补一点数据，也不要让检查点成为瓶颈。

3.4 打包提交与 Savepoint 恢复验证

代码写完只是第一步，验证「断点续传真的有效」才是这条链路能不能上生产的分水岭。完整的验证动作是这样一串：

# 1. 打包，产物是两个 jar，带依赖的那个才是提交用的 mvn clean package -DskipTests # 2. 启动 HDFS（检查点目录在 HDFS 上） start-dfs.sh # 3. 启动 Flink 集群 bin/start-cluster.sh # 4. 提交任务 bin/flink run -m node01:8081 \ -c com.example.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test-jar-with-dependencies.jar # 5. 观察 TaskManager 日志，此时应打印出 t1 的全量三行 tail -f log/flink-*-taskexecutor-*.out # 6. 给运行中的任务打一个 Savepoint bin/flink savepoint <JobId> hdfs://node01:8020/flinkCDC/save # 7. 在 WebUI 里 cancel 任务，然后去 MySQL 里改几条数据 # UPDATE test.t1 SET name='zhaoliu' WHERE id='1001'; # DELETE FROM test.t1 WHERE id='1003'; # 8. 从 Savepoint 恢复 bin/flink run -s hdfs://node01:8020/flinkCDC/save/savepoint-xxxx \ -c com.example.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test-jar-with-dependencies.jar

判断恢复是否真的成功的标志，不是任务起来了，而是第 8 步之后的日志里出现了第 7 步改动的两条记录：一条 UPDATE、一条 DELETE，而且没有重新打印全量三行。如果日志里又是从头开始的三行 INSERT，说明位点没存进去，八成是检查点没开或者路径写错了。这一步我每次接新库都会强制走一遍，因为它是唯一能在出故障之前证明「这套配置能续上」的办法。

4. FlinkSQL 与 pipeline YAML：两条少写代码的落地路径

4.1 一条 DDL 把 MySQL 表变成动态表

同一个 MySQL 源，换成 SQL 方式后代码量掉了一个数量级。核心是一条带connector = 'mysql-cdc'的建表语句，声明完之后它就是一张会自己更新的动态表，后面接什么算子都行。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); tableEnv.executeSql( "CREATE TABLE t1 (\n" + " id STRING PRIMARY KEY NOT ENFORCED,\n" + // 主键不强制校验，但 CDC 需要它来做更新合并 " name STRING\n" + ") WITH (\n" + " 'connector' = 'mysql-cdc',\n" + " 'hostname' = 'node01',\n" + " 'port' = '3306',\n" + " 'username' = 'root',\n" + " 'password' = '******',\n" + " 'database-name' = 'test',\n" + " 'table-name' = 't1',\n" + " 'server-id' = '5401-5404',\n" + // 并行度 > 1 时必须显式给范围 " 'server-time-zone' = 'Asia/Shanghai'\n" + // 与 MySQL 时区对齐，否则时间字段差 8 小时 ")" ); // 简单过滤，只有 name 非空的行往下走 tableEnv.executeSql("SELECT id, name FROM t1 WHERE name IS NOT NULL") .print();

PRIMARY KEY NOT ENFORCED这半句必须写。NOT ENFORCED是 Flink 的语法要求，表示主键只用于优化和去重语义，不做运行期约束校验；而PRIMARY KEY本身对 CDC 至关重要——连接器靠它识别一行数据的唯一标识，从而把 INSERT 和 UPDATE 合并成 upsert 语义。少写主键，下游收到的会是带+I、-U、+U标记的 changelog 流，而不是最终状态。

server-time-zone是我吃过亏的地方。MySQL 里存的是DATETIME，不带时区信息，连接器用 JVM 默认时区去解释它，容器里的 JVM 时区常常是 UTC，结果同步到 Doris 之后所有时间字段整体偏移 8 小时。这个错误不会报异常，只会让你在报表上看到「凌晨三点的订单」，排查起来很费时间。

4.2 SQL 与 DataStream 两条路的取舍

两种写法都能跑通，选哪个看的是维护成本而不是能力上限。

经验上的分界线是：下游只需要一张表或简单映射，SQL 够用；一旦涉及自定义解析、复杂事件时间处理、多流合并，老老实实回到 DataStream。还有一个隐性成本——SQL 路线的报错信息经常被规划器包一层，栈里看不到原始异常，排查链路比 DataStream 长，这一点在赶工期的时候很致命。

4.3 pipeline 模式：MySQL 到 Doris 的三段式 YAML

3.0 版本最实用的变化是 pipeline 模式：不写一行 Java 代码，用一份 YAML 描述「从哪读、往哪写、怎么跑」，命令行提交即可。整库同步这种重复劳动的活，用它比手写 DataStream 省太多。

先把两个连接器插件放进 Flink CDC 安装目录的lib下，版本号必须与主程序一致：

tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ cd /opt/module/flink-cdc-3.0.0 # 把 MySQL 源连接器和 Doris 目标连接器都放进 lib cp flink-cdc-pipeline-connector-mysql-3.0.0.jar lib/ cp flink-cdc-pipeline-connector-doris-3.0.0.jar lib/ ls lib/

配置文件分三段，source、sink、pipeline各管一摊：

source: type: mysql hostname: node01 port: 3306 username: root password: "******" tables: test.\.* # 正则：test 库下所有表 server-id: 5400-5404 # 范围必须够并行度用 server-time-zone: UTC+8 sink: type: doris fenodes: node01:7030 # Doris FE 的 HTTP 端口，不是 MySQL 协议端口 username: root password: "******" # 源端加列时，Doris 侧做轻量 schema change，避免整表重建 table.create.properties.light_schema_change: true # 单 BE 环境副本数只能是 1，多 BE 建议 3 table.create.properties.replication_num: 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1

tables用的是正则，test.\.*里那个反斜杠是转义点号，表示「库名 test，表名任意」。正则写宽了会把整库卷进来，写窄了漏表，建议先在 MySQL 里SHOW TABLES对一遍。server-id给的是一个区间而不是单值，因为 pipeline 内部会按并行度切分快照，每个分片都要独立的从库身份，区间长度要大于等于parallelism。

sink段里fenodes最容易填错——它要的是 Doris FE 的 HTTP 查询端口，通常是 8030，资料里写的是 7030，实际部署时按自己环境fe.conf里的http_port填。填成 BE 地址或者 MySQL 协议端口，症状是「连接超时」而不是明确的配置错误提示。

4.4 路由变更：把多库多表映射到目标端

默认行为是同名入库同表，源端test.t1就落到目标端test.t1。当需要把两个源库合并到目标端一个库，或者给表名加前后缀时，就要用 route 段：

route: # 源端 test_route 库的 t1，落到目标端 doris_test_route 库的 t1 - source-table: test_route\.t1 sink-table: doris_test_route.t1 # 源端其余表按原样落到同名库 - source-table: test\.(.*) sink-table: test.$1

两侧都接受正则，sink-table里可以用$1引用捕获组。这里的关键是先跑通无路由的默认同步，再加上 route 段重跑一次，对比两次结果，才能确认路由规则真的按预期生成了目标表，而不是把所有表都写进了同一个地方。资料里那两张主键区间刻意错开的t2，就是给这一步准备的校验样本：如果test.t2和test_route.t2落进了同一张表，主键区间一合并立刻就能看出来。

4.5 启动顺序与首轮同步验证

pipeline 模式依赖 Flink 的检查点机制，所以启动前必须先让集群具备检查点条件：

# 1. Flink 侧开启检查点，写进配置文件后分发到所有节点 vim conf/flink-conf.yaml # 追加：execution.checkpointing.interval: 5000 bin/start-cluster.sh # 2. Doris 侧启动 FE 与 BE /opt/module/doris/fe/bin/start_fe.sh --daemon /opt/module/doris/be/bin/start_be.sh --daemon # 3. 目标库需要先建好（pipeline 只自动建表，不自动建库） mysql -uroot -p****** -P9030 -hnode01 \ -e "CREATE DATABASE IF NOT EXISTS test;" # 4. 提交同步任务 bin/flink-cdc.sh job/mysql-to-doris.yaml # 5. 到 Doris 侧验证 mysql -uroot -p****** -P9030 -hnode01 \ -e "SELECT * FROM test.t1 ORDER BY id;"

第 3 步的顺序不能省。pipeline 会自动建表，但库必须提前存在，否则任务启动阶段就报「database not found」，而且这个错在 Flink WebUI 上只显示成一句笼统的初始化失败。任务起来之后，去 MySQL 里对几张表做增、改、加列操作，再刷新 Doris 侧，三个动作都要验：新数据出现、修改生效、新列自动补上。

5. 避坑清单：五个把同步任务跑挂的现场记录

5.1 任务重启后从零开始，全量数据又灌了一遍

现象：任务因为集群重启中断，恢复之后日志里重新打印了全量快照，下游出现大量重复写入。

原因：检查点没开，或者开了但没开启外部化保留，cancel 时检查点目录被一起清掉。Flink CDC 的位点存在状态里，状态没了就等价于第一次启动。

解决：enableCheckpointing必须开，enableExternalizedCheckpoints(RETAIN_ON_CANCELLATION)必须开，检查点存储放到 HDFS 这类共享存储上而不是本地磁盘。恢复时用flink run -s <savepoint路径>而不是直接重新提交。

5.2 并行度调到 2 以上就报 server_id 冲突

现象：单并行度跑得好好的，把parallelism改成 3，任务启动阶段抛异常，提示从库 id 已被占用。

原因：连接器每个快照分片都要以一个独立从库身份连 MySQL，不显式指定server-id时会在默认区间随机取，取值碰撞就报错；同时跑多个 CDC 任务时碰撞概率更高。

解决：显式给server-id一个范围，长度不小于并行度，且不同任务之间相互避开。我在配置文件里按任务编号分段，比如同步任务 A 用 5400-5410、任务 B 用 5420-5430，从源头上不重叠。

5.3 目标端时间字段整体差 8 小时

现象：数据行数对得上，主键也对得上，但所有时间列在 Doris 里比 MySQL 里早或晚了 8 小时。

原因：MySQL 的DATETIME不带时区，连接器用 JVM 默认时区解释它。容器镜像里 JVM 默认时区常是 UTC，而业务数据是按东八区写进去的。

解决：源端配置加server-time-zone，pipeline 用server-time-zone: UTC+8，SQL 路线用'server-time-zone' = 'Asia/Shanghai'。改完之后必须重跑一次全量，历史数据不会自动纠正。

5.4 Doris 连接器类型识别不到，报找不到 sink

现象：bin/flink-cdc.sh提交后立刻失败，提示无法创建类型为 doris 的 sink，或者ClassNotFoundException。

原因：flink-cdc-pipeline-connector-doris-3.0.0.jar没放进lib目录，或者放进去的版本与主程序不一致。这个加载是插件式的，缺包不会在启动时给友好提示。

解决：lib下同时确认 mysql 和 doris 两个 pipeline 连接器 jar 存在且版本号一致，然后重启 Flink 集群让 classloader 重新扫描。加完包不重启是另一个常见疏漏。

5.5 源端加了一列，同步任务开始报 schema 不匹配

现象：MySQL 上执行ALTER TABLE ADD COLUMN之后，同步任务报错或者新列数据丢失。

原因：目标端的表结构是建表时按当时的源端结构生成的，源端加列后目标端没有对应字段，写入时字段数对不上。

解决：pipeline 的 sink 段开启table.create.properties.light_schema_change: true，让 Doris 走轻量 schema change；同时确认replication_num与 BE 数量匹配，副本数配大了建表会一直卡在等待副本。DDL 变更之后建议主动观察一轮日志，确认 schema change 真的执行了，而不是静默失败。

6. 收尾这一公里：Doris 端对账与三个值得调的参数

数据同步跑起来只是及格线，能证明数据对得上才算交付。我的习惯是在两端跑同一条聚合 SQL，比对行数和内容摘要，而不是肉眼看前几行。

-- MySQL 侧执行 SELECT COUNT(*) AS cnt, SUM(CRC32(CONCAT_WS('|', id, name))) AS digest FROM test.t1; -- Doris 侧执行同一口径 SELECT COUNT(*) AS cnt, SUM(CRC32(CONCAT_WS('|', id, name))) AS digest FROM test.t1;

两边的cnt和digest都要一致。CONCAT_WS里字段顺序必须两边完全一样，NULL值的处理也要提前统一，否则摘要一定对不上但数据其实没错。这个对账要跑三个时间点：首轮全量完成后、做一轮增删改之后、任务重启恢复之后。第三个时间点最能暴露位点问题，因为只有它会走到「从检查点续读」这条分支。

参数调优上，我只动了三个地方。第一是split-size，快照阶段按主键区间切分的块大小，默认值偏保守，大表首轮同步会切成很多小片、来回切库，调到几十 MB 一档能明显缩短首轮时间，但不能超过单分片内存承受范围。第二是检查点间隔，从 5 秒放宽到 30 秒后，TaskManager 的 IO 等待明显下降，恢复时需要补的数据量也在可接受范围。第三是 sink 端写入批次，Doris 是按批提交的，批次太小会放大导入次数，太大则单批失败重试代价高，通常按每秒数据量乘以 3 到 5 秒来估。

有一次我图省事，把检查点间隔调到 3 秒、split-size又调得很大，首轮快照期间检查点一直超时，任务反复重启，最后是靠把间隔改回 30 秒才稳住。从那以后我每接一个新库，都先把server-id范围、binlog 保留时长、检查点存储路径这三条写进上线检查单，再动手改 YAML。希望帮到你。

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