Flink CDC 3.0 整库实时同步 MySQL 到 Doris 实战避坑

简介：面向具备一定 Flink 基础的大数据研发与实时同步工程师，这份由尚硅谷研究院编写的教程文档围绕 Flink CDC 3.0 展开，重点解决 MySQL 变更数据实时捕获并稳定同步至 Doris 的落地问题。压缩包仅含 1 个 docx 文件，大小约 145KB，内容从 CDC 概念与基于查询、基于 Binlog 两种方案对比讲起，延伸到 flink-cdc-connectors 组件的使用，并给出数据准备、DataStream 与 Flink SQL 两套实操路径。文档还完整覆盖 MySQL 到 Doris 的 Streaming ETL 流程，包括环境搭建、配置文件编写、任务启动与测试验证，对 Binlog 开启、检查点设置等关键配置也有说明。已有 1789 人学习，适合希望快速上手项目部署、理解各配置项作用并按业务调整参数的中高级读者。

1. 从一张对不上的看板说起：Flink CDC 3.0 值不值得花两天搭一遍

某公司的订单库在 MySQL，看板和报表查的是 Doris，中间靠一个每小时跑一次的批处理搬运。上线第三周出过一次事：运营改了一笔订单金额，看板上还是旧值，客服拿着两个数字来回问。后来换成 Flink CDC 3.0 做 MySQL 到 Doris 的实时同步，常驻作业跑着，binlog 一有变更，毫秒级就能在 Doris 里查到。这份资料讲的正是这条链路：CDC 到底怎么选、Flink CDC 3.0 的 DataStream 和 FlinkSQL 两种写法、以及 3.0 新增的 YAML Pipeline 整库同步到 Doris 怎么配。适合已经会写 Flink 作业、但没在生产上跑过 CDC 的数据开发，也适合被批处理延迟折磨过的同学。

2. CDC 选型：Binlog 路线为什么比批量抽数稳

2.1 基于查询和基于 Binlog，差的不是延迟而是语义

很多团队第一次做同步，第一反应是写个定时任务，select * from t where update_time > 上次时间，跑完记个水位线。这条路简单、无依赖、出问题好查，所以它的边界也特别清楚。把它和 Binlog 路线放在一张表里看，差距不在"快多少"，而在于能不能拿到完整的变更语义。

物理删除这一条最致命。批处理想感知删除，只剩一条路：把源表主键集合拉回来，和目标表做差集。百万级还能忍，千万级就是每次同步都全表比对，跑一次顶一次。而 row 格式的 binlog 里，DELETE 事件带着完整的行镜像，主键清清楚楚，直接下发就行。

update_time这类字段看着能用，实际用起来玄学得很。应用服务器时钟漂移、事务提交晚于字段赋值、批量更新时同一秒内几百行、软删除忘记加过滤条件，任何一条都能让数据漏掉或者重复。我见过最典型的翻车是：业务方为了"减少写压力"，把update_time设成ON UPDATE CURRENT_TIMESTAMP，结果批量修数脚本用INSERT ... ON DUPLICATE KEY UPDATE覆盖时没有真正改字段值，MySQL 不更新这一列，下游就永远不知道这行变过。

2.2 Flink CDC 3.0 在架构里的位置：从 Source 组件到 Pipeline

Flink CDC 早期是个 source 组件，flink-cdc-connectors提供 MySQL、PostgreSQL 等库的 CDC Source，读出来是 DataStream，后面接什么 sink、做什么加工全凭自己写。它的好处是灵活：可以在中间做双流 join、打宽、过滤，甚至用 FlinkSQL 把 CDC 流和维表关联。代价是每加一张表就要改代码、重新打包、重新提交作业，表多的时候维护成本很吓人。

3.0 引入 Pipeline 连接器之后，形态变了。MySQL source、Doris sink、库表路由、表结构变更传播被打包进一个 YAML 文件，用bin/flink-cdc.sh job/xxx.yaml提交，框架在内部把这份声明翻译成一个 Flink 作业。加表只改一行正则，加列不用改代码，schema evolution 会顺着链路传到 Doris 侧。

两条路怎么选，我一般这么判断：表在十张以内、中间需要加工逻辑（比如多表合并、维度补齐、字段脱敏），写 DataStream 或 FlinkSQL，可控性最高；整库搬迁、几十上百张表、逻辑就是"原样同步"，用 Pipeline，省下的是大量重复代码和上线流程。别小看后者——给五十张表各写一遍 sink 建表语句，出一次字段类型对不上的事故就够喝一壶。

2.3 什么情况下别硬上 CDC

CDC 不是万能药，有几个场景我会直接劝退。

源表没有主键。Flink CDC 的增量快照要靠主键切分 chunk，Doris 的 Unique Key 模型也要主键来做 upsert。没有主键的表接进来，要么退化成单并行度全表扫，要么下游建不了合适的模型，更新变成重复插入。遇到这种表，先推动业务加主键，比在链路里打补丁划算得多。

源库根本拿不到 binlog。托管实例有的只给 binlog 文件不给订阅权限，有的只保留很短时间（binlog_expire_logs_seconds默认 30 天，有的被调成几小时）。如果全量初始化要跑五个小时，而 binlog 只留两小时，中途一旦失败重来，位点就断了。上线前一定确认保留时长大于最坏情况下的初始化耗时。

只需要 T+1 的报表。链路越实时，运维成本越高：checkpoint 要配、位点要监控、故障要能续跑。如果业务方看的就是昨天的汇总数，跑批足够，别为了技术先进性给自己找活干。

需要跨多库多表做重口径加工。CDC 的定位是"搬运"，不是"计算"。复杂的关联聚合，老老实实落到数仓分层里做，让 CDC 只负责把 ODS 层喂满。

3. 把 MySQL 侧的地基打好：Binlog 四个参数与初始化数据

3.1 开 Binlog：四个参数决定后面所有事

CDC 读不到数据，九成问题出在 MySQL 这一侧，而且基本都是配置问题。需要改的是my.cnf（或者my.ini），改完必须重启，SET GLOBAL只能临时生效，重启就回去了。

# /etc/my.cnf [mysqld] # 实例唯一标识，CDC 连接时会伪装成一个从库挂上去，不能和已有的从库冲突 server-id = 1 # 开启 binlog，值作为日志文件前缀，实际文件形如 mysql-bin.000001 log-bin = mysql-bin # 必须是 ROW，STATEMENT/MIXED 下拿不到行级前后镜像 binlog_format = ROW # 白名单：只有这里列出的库产生的变更才会写进 binlog binlog-do-db = test binlog-do-db = test_route # 保留时长，按最坏情况下的全量初始化耗时来定，默认 30 天，别调到小时级 binlog_expire_logs_seconds = 604800

四个参数里，server-id和binlog-do-db是最容易出事的。前者在于多个 CDC 任务如果共用同一个 server-id，MySQL 侧会认为是从库重复接入，第二个任务直接连不上或者被踢掉；后者在于它是"按当前默认库"判定的白名单，一个事务里USE a; UPDATE b.t ...这种跨库写法，判定用的是a，b的变更就漏了。

改完重启，用两条命令确认。

-- 确认关键变量 SHOW VARIABLES LIKE 'log_bin'; -- 期望 ON SHOW VARIABLES LIKE 'binlog_format'; -- 期望 ROW SHOW VARIABLES LIKE 'binlog_do_db'; -- 期望看到 test,test_route -- 确认 binlog 文件已经生成 SHOW MASTER STATUS\G

SHOW MASTER STATUS在 8.0 里被SHOW BINARY LOG STATUS取代了，老版本用前者。输出里的File和Position两列是 CDC 的起点，initial模式下会先做全量快照再从这个位置往后读。

账号权限也要提前给足。CDC 需要的权限比想象中多：SELECT（读表结构）、RELOAD（执行FLUSH TABLES之类）、SHOW DATABASES、REPLICATION SLAVE（挂从库）、REPLICATION CLIENT（读位点信息）。生产上别直接用 root，单独建一个账号，密码里避免:和@这类会和连接串打架的字符。

提示：云上托管实例如果只提供内网地址，注意 Flink TaskManager 所在机器能不能通到这个地址，别在本地 IDE 里跑通了就以为集群也行。

3.2 建库建表与初始化数据：为后面的路由实验铺路

资料里准备了两个库，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) ); 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) );

两个库的表名完全一样，这是刻意设计的。后面验证 route 段的时候，如果两个库都同步到 Doris 的同一个库，表名会撞车；只有配上路由规则，test_route的表才能落到另一个库里。数据也造两份，注意test_route.t2用的是 1004 到 1006，和test.t2的主键不重复，方便观察是不是真的同步过来了。

USE test; 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; 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');

建完先别急着跑 Flink，用SELECT COUNT(*)对一下两边的行数，记下来。后面每验证一步，都拿这个数字做基线，比看日志猜要快得多。

3.3 依赖坐标与打包：别让 classpath 毁掉一上午

工程用 Maven 管依赖，Flink 版本 1.18.0，CDC 连接器 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 核心，提交到集群时可以改 provided，本地调试要用 compile --> <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 走这三个，版本必须和 flink-version 完全一致 --> <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 连接器，资料里用的是这个坐标 --> <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，把 CDC 连接器和 MySQL 驱动一起塞进 fat jar。这里有个常被忽略的点：Flink 集群的lib目录里通常没有 CDC 连接器，如果打包时把它设成provided，提交上去就是ClassNotFoundException: com.ververica.cdc.connectors.mysql.source.MySqlSource，日志在 TaskManager 上，看半天找不到。

如果拉不到com.ververica坐标，换org.apache.flink:flink-connector-mysql-cdc:3.0.0试一下，功能等价，只是 groupId 迁移过。这类坐标问题不值得死磕，能编译过、能跑通就是对的。

3.4 DataStream 版 MysqlSource：检查点配不对，断点续传就是空话

DataStream 写法把每一步都摆在明面上，适合需要精细控制的场景。整个类分六块：环境、检查点、Source、读取、输出、执行。中间检查点那一段最长，也最容易被复制粘贴时漏掉。

public class FlinkCDCDataStreamTest { public static void main(String[] args) throws Exception { // 1. 流处理环境，并行度先给 1，方便对日志 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. 检查点：CDC 的 binlog 位点存在状态里，不开 CK 就没有断点续传 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 次就彻底失败，避免无限重启刷日志 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", "hadoop"); // 3. CDC Source MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("node01") .port(3306) .databaseList("test") .tableList("test.t1") .username("cdc_user") .password("Cdc@2024") .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); // 4. 读取并打印 DataStreamSource<String> mysqlDS = env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), "MySQLSource"); mysqlDS.print(); // 5. 执行 env.execute("flink-cdc-datastream-test"); } }

参数这一块值得逐条过一遍，因为每个都对应一类线上事故。

databaseList和tableList是两层过滤，前者决定连哪些库，后者精确到表，格式是库名.表名，支持正则如test\.t.*。只写databaseList("test")不写tableList，会把整个库的表全抓下来，表多的时候初始化能跑几个小时。

startupOptions有五种取值，行为差别很大。initial()是首次启动先做全量快照再接着读增量，最常用；earliest()不做快照，从 binlog 最早的位置读，前提是 binlog 没被清理过，否则直接报错；latest()不做快照，只读连接建立之后的新变更，适合做增量对齐；specificOffset()和timestamp()用于从指定位点或时间点恢复，做数据修复时用得上。

deserializer决定输出格式。JsonDebeziumDeserializationSchema输出的是 Debezium 风格 JSON，里面op字段标了操作类型：r是快照读、c是插入、u是更新、d是删除。before和after分别是变更前后的行镜像，删除操作只有before，插入只有after。第一次看这个 JSON 会觉得字段多得离谱，其实有用的就这几个。

检查点里的setCheckpointStorage是最容易在测试环境被写成本地路径的一项。本地文件系统在单机 Standalone 上能跑，一旦换成 Kubernetes 或者会话集群，JobManager 重启换了机器，状态目录就找不到了，作业只能从空状态起，于是重复消费一遍全量。测试环境图省事用本地路径，上线前必须换成 HDFS 或对象存储。

并行度这里给 1，是为了输出顺序稳定、日志好对。真上生产，CDC Source 的并行度取决于表的数量和主键分布，initial快照阶段会按主键切 chunk 并行读，增量阶段每个并行子任务各挂一个 binlog 连接，并行度开太大反而会给 MySQL 造成连接压力。

打包上传提交：

# 打包，产物在 target 下 mvn clean package -DskipTests # 提交到会话集群，-s 指定 savepoint 路径即可从位点恢复 bin/flink run \ -m node01:8081 \ -c com.example.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test.jar

提交后在 TaskManager 日志里能看到全量快照阶段的输出，op全是r。等快照跑完，往test.t1里插一条，日志里立刻出现op=c的记录。到这一步，说明 binlog 通了、权限够了、反序列化也正常。

4. 从 FlinkSQL 到 CDC 3.0 Pipeline：把 MySQL 整库搬到 Doris

4.1 FlinkSQL 建表：最省事的入门口子

不想写 Java 的话，FlinkSQL 的 CDC 连接器是最快的验证方式。核心就是一条CREATE TABLE，把 MySQL 表映射成 Flink 表，之后当普通表查。

CREATE TABLE t1 ( id STRING, name STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'node01', 'port' = '3306', 'username' = 'cdc_user', 'password' = 'Cdc@2024', 'database-name' = 'test', 'table-name' = 't1' );

PRIMARY KEY ... NOT ENFORCED这句必须写，但 Flink 不会真的去校验唯一性，它只是告诉优化器"这列可以做 upsert 的 key"。漏写的话，下游 JDBC sink 或者 Doris sink 会退化成 append 模式，更新就变成重复插入，Doris 里同一个 id 出现好几行。这个坑我在早期项目里踩过，排查了半天以为是 CDC 重复消费，其实是建表语句少了一行。

字段类型映射也别大意。MySQL 的TINYINT(1)在 Flink 里映射成BOOLEAN，到了 Doris 又变成BOOLEAN，如果下游按数值处理就会出问题；DECIMAL的精度要显式声明，不写默认是DECIMAL(10,0)，小数部分直接被截断；DATETIME和TIMESTAMP在 Flink 里是两种类型，前者无时区、后者带时区，跨时区同步时对不上八小时往往就出在这里。

FlinkSQL 跑起来最直接：

# 交互式客户端里直接粘贴上面的 DDL，再查一下 SELECT * FROM t1;

它会先做一遍全量，然后把增量变更持续打出来。适合快速验证，也适合在 FlinkSQL 任务里再接一层聚合、双流 join。生产上表一多，DDL 写起来就烦了，这时候该上 Pipeline。

4.2 CDC 3.0 Pipeline：连接器放置与目录约定

Pipeline 模式不需要写一行 Java，但要准备好安装包和连接器 jar。把flink-cdc-3.0.0-bin.tar.gz解到/opt/module/，然后确认两件事：MySQL 和 Doris 的 pipeline 连接器 jar 在lib目录下，作业描述文件放在任意目录（习惯上建个job/）。

tar -zxvf flink-cdc-3.0.0-bin.tar.gz -C /opt/module/ cd /opt/module/flink-cdc-3.0.0 # 两个连接器 jar 必须在 lib 下，缺一个启动就报 connector not found ls lib/ | grep pipeline-connector # flink-cdc-pipeline-connector-doris-3.0.0.jar # flink-cdc-pipeline-connector-mysql-3.0.0.jar mkdir -p job

Flink 集群侧还要开检查点，Pipeline 依赖它做位点恢复。改conf/flink-conf.yaml，改完分发到所有节点再启动集群。

# conf/flink-conf.yaml execution.checkpointing.interval: 5000 state.backend: hashmap state.checkpoints.dir: hdfs://node01:8020/flink-cdc/ck execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION

execution.checkpointing.interval不给的话，作业能跑起来，也能同步数据，但一旦重启就从最开始重来，全量快照再跑一遍。这属于"能跑但不对"的那类配置，测试时不容易暴露，上线后宕一次机才发现。

4.3 mysql-to-doris.yaml 逐项拆解

整份配置就三段：source、sink、pipeline。每段里的字段不多，但每个都影响行为。

source: type: mysql hostname: node01 port: 3306 username: cdc_user password: "Cdc@2024" tables: test.\.* # 正则，test 库下所有表 server-id: 5400-5404 # 区间，框架自动分配，避免多任务冲突 server-time-zone: UTC+8 sink: type: doris fenodes: node01:7030 # FE 的 HTTP 端口 username: root password: "000000" table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1

tables用的是正则而不是精确表名，test.\.*表示test库下全部表。注意 YAML 里反斜杠在无引号标量中会原样保留，写成"test\\.\\*"反而会报错，这点和 Java 字符串不一样。

server-id写成区间，是 Pipeline 模式相对 DataStream 的一个改进。Flink 内部会按并行度从区间里取不同的 id 给每个读取子任务，避免一个作业内部自己跟自己冲突。但也意味着区间要留够，parallelism是 4 就得至少 4 个 id。

server-time-zone: UTC+8决定了 Debezium 解析TIMESTAMP时按哪个时区解释。不配的话默认 UTC，同步到 Doris 的DATETIME字段会整体差八小时，而且这个偏差很隐蔽——全量阶段可能看不出来，增量阶段一对比才发现。

sink 侧的fenodes是 Doris FE 的 HTTP 端口，默认 8030，资料里改成 7030。注意不是 9030，9030 是 MySQL 协议端口，填错了连接会超时，报错信息也不直观。

table.create.properties.replication_num: 1对单 BE 的测试环境是必须的。Doris 建表默认副本数是 3，只有一个 BE 的时候建表直接失败，报Failed to find enough host with storage medium and tag。这个错看着像磁盘问题，实际是副本数配多了。

light_schema_change: true打开轻量级 schema change，这是自动建表时带上的属性。它的作用是让后续加列这类变更不用重刷全量数据，配合 CDC 的 schema evolution 才能真正做到"MySQL 加一列，Doris 自动跟上"。

4.4 route 段：库表改名和整库同步

默认情况下，源库表名和目标库表名一致，落到 Doris 的test库。但test_route库也有同名表，两边直接同步就会撞。这时加 route 段做改名。

source: type: mysql hostname: node01 port: 3306 username: cdc_user password: "Cdc@2024" tables: test_route.\.* server-id: 5400-5404 server-time-zone: UTC+8 sink: type: doris fenodes: node01:7030 username: root password: "000000" table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 route: - source-table: test_route\.t1 sink-table: test_route.t1_route - source-table: test_route\.t2 sink-table: test_route.t2_route pipeline: name: Sync test_route to Doris parallelism: 1

source-table写源端的库表正则，sink-table写目标端的库表名，两边都可以用正则捕获组。表多的时候用一条规则批量改：source-table: test_route\.(.*)配sink-table: test_route.dwd_$1，把所有表统一加前缀。这个写法在做库级隔离时很省事，不用一张一张列。

路由规则是从上往下匹配，第一条命中就停。规则写多了要注意顺序，宽泛的正则放后面，否则后面的精确规则永远不生效。

4.5 启动、验证与增量观察

Doris 侧先把 FE 和 BE 拉起来，用 MySQL 客户端连 9030 端口建库。

# 启动 Doris bin/start_fe.sh --daemon bin/start_be.sh --daemon # 用 MySQL 协议连进去，注意端口是 9030 mysql -uroot -p000000 -P9030 -hnode01

CREATE DATABASE test; CREATE DATABASE test_route; SHOW DATABASES;

然后提交 Pipeline 作业。

cd /opt/module/flink-cdc-3.0.0 bin/flink-cdc.sh job/mysql-to-doris.yaml

命令会打印作业 ID 和 Flink WebUI 地址。到 WebUI 上看，作业拓扑是一个 source 接一个 sink，并行度取自 YAML 里的parallelism。全量阶段能看到 source 端有快照读取的记录数，跑完后切到增量。

验证分三步，一步都别省。

第一步，全量对账。在 Doris 里SELECT COUNT(*) FROM test.t1，和 MySQL 侧的基线数字比。不一致的话先看是不是表里有大字段或者特殊字符导致解析失败，sink 侧的脏数据记录一般会打 warn 日志。

第二步，增量插入。往test.t1插一条新数据，等两三秒，Doris 里查一下。能查到说明整条链路通了。

第三步，更新和删除。UPDATE test.t1 SET name='xxx' WHERE id='1001'，看 Doris 里是不是真的改了而不是多出一行；DELETE FROM test.t1 WHERE id='1003'，看 Doris 里是不是也没了。删除这一条最能暴露表模型问题——如果 Doris 侧建的是 Duplicate 模型，删除就变成了插入一条带删除标记的记录，查询时还能看见。

第四步，加列。ALTER TABLE test.t1 ADD COLUMN age INT，然后往新列写值。开了 schema evolution 的话，Doris 侧会自动加列并同步数据。这一步是 3.0 相对 2.x 最实用的改进，也是资料里特意提到要观察的部分。

5. 避坑与排查：五条血泪记录和一张上线自检表

5.1 五条踩过的坑，按现象、原因、解决来记

现象一：作业启动就报The MySQL server is not configured to use a ROW binlog_format，或者能连上但收不到任何更新。

原因基本锁定在binlog_format不是ROW，或者log-bin压根没开，再或者binlog-do-db没把目标库列进去。还有一种隐蔽情况：配置改在[mysqld]之外的小节里，比如放在[mysqld_safe]下，MySQL 读到了但不生效。

解决顺序是先SHOW VARIABLES LIKE 'binlog%'确认运行时值，再回头查配置文件的位置和段落。8.0 支持SET GLOBAL binlog_format = 'ROW'临时切，但已经写出去的 STATEMENT 格式日志没法补，测试环境可以这么应急，生产必须改配置文件重启。

现象二：第二个 CDC 任务提交后，第一个任务断开重连，日志里反复出现A slave with the same server_uuid/server_id as this slave has connected。

原因是两个作业用了同一个server-id。DataStream 写法里这个 id 由 Debezium 自动生成或者从连接参数里取，Pipeline 写法里从 YAML 的区间分配。多个人在同一套库上各跑各的任务，最容易撞。

解决是规划好 id 段，比如测试环境用 5400-5499，预发用 5500-5599，每个作业分一个小区间。Pipeline 的server-id写区间而不是固定值，就是为了让框架自己错开。

现象三：作业重启后数据重复，Doris 里同一个主键出现多行，或者整库数据被重放了一遍。

没开检查点，或者检查点目录配成了本地路径。CDC 的位点是存在 Flink 状态里的，状态丢了就等于失忆，只能按startupOptions重新来一遍。initial()会重跑全量，下游如果没有幂等去重，直接翻倍。

解决是把 checkpoint 间隔压到 5 秒以内，状态后端和存储目录放到共享存储，任务取消时RETAIN_ON_CANCELLATION。另外 Doris 侧建表一定用 Unique Key 模型，主键相同自动覆盖，这是最后一道后悔药。

现象四：Doris 建表失败，报Failed to find enough host with storage medium and tag或者replication num should be less than the number of available BE。

单 BE 环境下replication_num默认 3，凑不齐副本。这个错和磁盘、标签都没关系，纯粹是副本数配多了。

解决是在 sink 段加table.create.properties.replication_num: 1，或者再起两个 BE。生产环境三副本没问题，测试环境别照抄生产配置。

现象五：时间字段整体差八小时，或者DATETIME值变成1970-01-01。

差八小时是server-time-zone没配或者配错，Debezium 默认按 UTC 解析。变成 1970 一般是源表里的零值日期（0000-00-00 00:00:00）在转换时溢出，或者字段类型映射成了DATE而实际存的是DATETIME。

解决是 source 段固定写server-time-zone: UTC+8，源表尽量避免零值日期；类型映射上，MySQL 的DATETIME对应 Flink 的TIMESTAMP(3)，对应 Doris 的DATETIME，别图省事全写成STRING，后面做时间聚合会哭。

5.2 上线前照着过一遍的自检表

这张表我一般会贴在值班文档里，每次上新链路前对着勾一遍，比事后翻日志省事。

5.3 出问题先看哪几个地方

Flink WebUI 是第一个要看的地方，不是日志。先看作业是不是 RUNNING，再看 checkpoint 那一栏的最近一次成功时间和大小。checkpoint 一直失败的话，后面所有数据问题都别分析了，先把状态存储的问题解决掉。

第二个看 source 端的指标，numRecordsOut是否持续增长，currentFetchEventTimeLag这类延迟指标是不是在合理范围。延迟持续上涨说明下游 sink 扛不住，Doris 的导入任务在排队。

第三个才是 TaskManager 日志，按ERROR和WARN过滤，重点看debezium和doris关键字。CDC 的日志噪音很大，全量阶段满屏都是快照进度，真正有用的报错往往夹在中间。

Doris 侧看SHOW LOAD WHERE LABEL LIKE '%cdc%'，导入失败的话这里能看到具体原因，比如字段类型不匹配、分区不存在、内存超限。很多"数据没同步过来"的问题，最后发现是 Doris 导入任务失败，Flink 侧正常。

6. 进阶：Savepoint 换版本和一次能跑的对账 SQL

作业写完能跑只是开始，真正难的是怎么在不丢数据的前提下换版本、改逻辑、加表。CDC 作业有个天然优势：位点存在状态里，只要 Savepoint 在，就能从断点续上，不用重跑全量。

换版本的标准动作是这样：先在 WebUI 里触发 Savepoint，等它成功（别用 cancel 代替，cancel 可能不触发保存）；确认 HDFS 上出现了savepoint-xxxx目录；用新 jar 从 Savepoint 启动；观察新作业追平延迟；最后再停旧的。中间这几分钟两边会同时消费同一段 binlog，所以下游必须是幂等的——Doris 的 Unique Key 模型天然满足这一点，同主键后写的覆盖先写的，不会出现双份。

# 触发 savepoint，记下返回的路径 bin/flink savepoint <jobId> hdfs://node01:8020/flink-cdc/savepoint # 从 savepoint 恢复，换成新打出来的 jar bin/flink run \ -s hdfs://node01:8020/flink-cdc/savepoint/savepoint-xxxx \ -c com.example.cdc.FlinkCDCDataStreamTest \ ./flink-cdc-test-v2.jar

Pipeline 模式也一样，bin/flink-cdc.sh支持从 savepoint 恢复，只是参数位置在文档里不太显眼，第一次用容易漏。

另一个常被问的问题是"怎么确认两边真的一致"。别只比行数，行数对得上但内容错了的情况太多了。我一般跑这个对账 SQL：在 MySQL 和 Doris 上分别算一遍主键的哈希聚合，两边比对。

-- 源端和目标端各跑一次，结果应完全一致 SELECT COUNT(*) AS cnt, SUM(CRC32(CONCAT_WS('|', id, IFNULL(name,'')))) AS sig FROM test.t1;

CRC32够快，碰撞概率在这个量级可以接受；字段多的话换成MD5或者SHA2，慢一点但更稳。这个查询我从两边各跑一次，cnt和sig都对上才算过关。差异定位时按主键区间二分，比逐行 diff 快得多。

还有个小技巧：给 CDC 作业单独配一个告警，盯currentFetchEventTimeLag这个指标，超过 60 秒就报。延迟涨上去往往不是 CDC 的问题，而是 Doris 那边导入排队或者 BE 磁盘满了，早发现比等业务方来问强。这条是我踩过一次坑之后加的——那次 Doris 的 BE 挂了两个，CDC 作业表面上还在 RUNNING，数据全堵在内存里，谁也没发现，直到第二天报表全空。

从那以后我每次上新链路，都强制走一遍「插一条、改一条、删一条、加一列、重启一次」这五步，少一步都不放心。希望帮到你。

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