企业官网建设流程全解析

简介：面向具备一定 Flink 基础的大数据研发人员，这份由尚硅谷研究院编写的 docx 文档围绕 Flink CDC 3.0 展开，系统讲解从 MySQL 变更数据捕获到 Doris 实时同步的完整链路。内容先从 CDC 概念切入，对比基于查询与基于 Binlog 两类方案在执行模式、变化延迟和数据库压力上的差异，再引出 flink-cdc-connectors 组件直接读取全量与增量数据的能力。案例部分给出建库建表与造数的具体 SQL、DataStream 和 Flink SQL 两种接入写法，以及 MySQL 到 Doris 的 Streaming ETL 全流程，涵盖环境搭建、配置文件编写、任务启动与结果验证，并对开启 Binlog、设置检查点等容易出错的环节逐一提示。资源包共 1 个 docx 文件，约 145KB，章节按概念、实操、同步案例递进，篇幅集中便于查阅。已有 1789 人学习，适合需要独立完成同步链路部署、保障数据一致性的工程师参考。

1. 业务要的是秒级可见，而不是每小时跑一次的批量抽取

去年我接手一个数据看板的活，运营那边抱怨最多的不是报表做得丑，是数据太旧——凌晨跑一遍 DataX，全量抽一次，上午十点看到的还是昨天的库存和订单。当时团队的第一反应是「把批次从一天一次改成每小时一次」，我算了一下账：表体量最大的那张订单表，单次全量抽取要二十多分钟，压到每小时跑，MySQL 从库的 IO 直接顶到天花板，还未必能保证数据是连续的。所以我改了个方向，用 Flink CDC 3.0 从 MySQL 的 Binlog 做增量捕获，实时写进 Doris，整条链路从「定时批量」变成「变更即同步」。这套方案适合两类人：一类是要交大数据方向毕设或课程设计、需要一条能跑通、能演示、能说清原理的实时同步链路的同学；另一类是在公司里被安排做数仓实时入湖入仓、要拿出一份可以落地的技术方案的人。下面我按「怎么设计、怎么配、怎么跑、怎么验收」的顺序把这条链路拆开讲。

2. 链路总体设计：从 Binlog 到 Doris 的三段式结构

这条链路看着是「MySQL 到 Doris」两个点，实际落地时被我切成了三段：源端变更捕获、中间流处理、目标端写入。切三段的理由很实际——每一段出问题时的排查手段完全不同，混在一起做，出故障时你连日志该去哪个组件里翻都不知道。

2.1 为什么用三段式，而不是一把梭

先明确一个常识性的分界。CDC 的实现方式主要分两类：基于查询的和基于 Binlog 的。前者代表是 Sqoop、DataX 这类工具，本质是定时发SELECT，执行模式是批处理，抓不到删除动作，延迟取决于调度周期，而且每次查询都会给源库增加压力。后者代表是 Canal、Maxwell、Debezium，走的是数据库自己的日志流，执行模式是流式，增删改都能捕获，延迟在秒级，对源库几乎没有额外查询压力。

Flink CDC 站在第二类这一边，它把 Debezium 的采集能力封装成 Flink 的 Source 组件，直接从 MySQL、PostgreSQL 这类库读全量快照和增量变更。三段式的划分是这样的：

把这三段分开之后，很多设计决策就变得有依据了。比如「要不要在中间加一层 Kafka」，如果下游只有 Doris 一个消费者，且业务能接受 Flink 作业重启期间的短暂中断，那就不加，链路短、延迟低、少一个组件少一个故障点；如果有三个以上的下游系统要消费同一份变更，那就必须加 Kafka 做分发，让 Flink 只负责采集，各下游各自消费。我在这个项目里选了不加，因为需求只有一个目标库。

另一条设计原则是「先全量后增量」的切换必须由 Checkpoint 兜住。Flink CDC 首次启动时会对监控的表做一次全量快照，快照完成后无缝衔接到 Binlog 的当前位置继续读。这个「无缝」不是凭空来的，它依赖 Flink 的检查点机制把读取位点以状态的形式存下来。所以 Checkpoint 不是可选项，是这条链路能不能断点续传的前提。我一般会在方案评审阶段就把这条写进去，避免后面有人为了「跑得快一点」把 Checkpoint 关掉，重启之后整库重扫一遍，那是真翻车。

2.2 源端表结构与 Binlog 参数的定稿

源端的准备工作分两件事：建库建表，和打开 Binlog。先看表结构，演示环境我建了两个库，一个叫demo，一个叫demo_route，后者专门用来验证路由（把源库的表写到目标库的不同库名下）这个能力。每个库里放三张表：

-- 源库 demo：三张表，t1/t2 结构一致，t3 多一个性别字段 CREATE DATABASE IF NOT EXISTS demo; USE demo; 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.\..* 这类正则 CREATE DATABASE IF NOT EXISTS demo_route; USE demo_route; CREATE TABLE t1 ( `id` VARCHAR(255) PRIMARY KEY, `name` VARCHAR(255) );

插入一批主键固定的测试数据，主键统一用1001、1002、1003这种，好处是后面做更新和删除验证时，你能精确知道哪一行应该变成什么。这一步别图省事用自增主键，否则目标端核对数据时你连比对基准都没有。

然后是 Binlog。MySQL 8 默认是开启 Binlog 的，但格式和过滤范围不一定符合 CDC 的要求，必须显式改配置文件并重启：

# 编辑 MySQL 配置文件（路径随安装方式不同，一般在 /etc/my.cnf） sudo vim /etc/my.cnf # 在 [mysqld] 段落下追加 server-id = 1 log-bin = mysql-bin binlog_format = row # CDC 必须为 row，statement/mixed 会丢字段变更 binlog-do-db = demo # 只对需要同步的库记日志，减少体积 binlog-do-db = demo_route # 改完重启 sudo systemctl restart mysqld

三个参数各自的作用要说清楚，因为它们是最容易配错的地方。server-id是这台 MySQL 在复制拓扑里的唯一标识，CDC 连接器本质上是以一个「伪从库」的身份去订阅 Binlog，所以这个值必须存在且全局唯一；binlog_format必须是row，因为只有行模式才会把每一行变更前后的完整字段值记下来，语句模式只记 SQL 文本，你拿不到变更后的列值；binlog-do-db是白名单，不配就是全库记录，Binlog 体积涨得很快，生产上一般都会收敛到实际要同步的几个库。

配完用SHOW VARIABLES LIKE 'binlog_format';和SHOW MASTER STATUS;确认一下，后者能返回文件名和位点，说明日志已经在写了。

2.3 版本矩阵的锁定

这条链路涉及四个组件，版本对不上会出现各种奇怪的类找不到和协议不兼容。项目里我锁的是这一组：Flink 1.18.0、flink-connector-mysql-cdc 3.0.0、MySQL 8.0.31、Doris 1.2.4（FE 的 MySQL 协议端口 9030）。JDK 用 8，Maven 编译的 source/target 都设 8。Doris 这边至少一个 FE 加一个 BE，BE 的数量会直接影响后面建表时副本数的设置，这个坑我在第 5 章会单独说。Flink 集群用 Standalone 模式起就够验证了，start-cluster.sh一把起完，Web UI 默认在 8081。

3. DataStream 实现：Checkpoint 与启动模式是两个决定成败的参数

三种落地方式（DataStream、Flink SQL、3.0 的 Pipeline YAML）里，DataStream 是最「原始」的一种，代码量最大，但也是你能把每个参数看明白的一种。毕设里如果只贴 YAML，答辩时被问「全量转增量是怎么做到的」很容易卡壳，所以我建议至少把 DataStream 版本自己敲一遍。

3.1 依赖清单与打包方式

Maven 里除了 Flink 的三件套（flink-java、flink-streaming-java、flink-clients），还要 Table 相关的三个包（flink-table-planner_2.12、flink-table-runtime、flink-table-api-java-bridge），以及 CDC 连接器和 MySQL 驱动。CDC 连接器在 3.0 版本换了 groupId，用的是com.ververica：

<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 运行时与流处理 API --> <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> <!-- Table API，Flink SQL 方式会用到 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_2.12</artifactId> <version>${flink-version}</version> </dependency> <!-- MySQL CDC Source，注意 groupId 是 com.ververica --> <dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>3.0.0</version> </dependency> <!-- MySQL 驱动，版本要与源库大版本对齐 --> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>8.0.31</version> </dependency> </dependencies> <build> <plugins> <!-- 打 fat jar，否则集群上跑会报 ClassNotFound --> <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-connector-mysql-cdc这个包本身已经把 Debezium 的依赖打进去了，但 Flink 自己的flink-streaming-java这类包在集群上是由 Flink 的 lib 目录提供的，所以打 fat jar 时理论上有冲突风险。实际做法是本地开发不设 scope，打出来的包直接flink run也能跑；如果集群上报「类加载冲突」，就把 Flink 相关依赖改成provided再打一次。这个坑我在第三个项目上才彻底记住。

3.2 MySqlSource 的构建与五个启动模式

核心代码不长，但每一个 builder 方法都对应一个必须在方案里说清楚的设计点：

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(5000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务被 cancel 时保留最后一次 CK，方便用 savepoint 恢复 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 失败率重启：1 天内最多 3 次，每次间隔至少 1 分钟 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); // 状态后端与 CK 存储位置 env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://node-01:8020/flink-cdc/ck"); System.setProperty("HADOOP_USER_NAME", "flink"); // 3. 构建 MySQL CDC Source MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("node-01") .port(3306) .databaseList("demo") // 捕获的库 .tableList("demo.t1") // 捕获的表，注意要带库名前缀 .username("cdc_user") .password("Your@Passw0rd") .deserializer(new JsonDebeziumDeserializationSchema()) // SourceRecord 转 JSON .startupOptions(StartupOptions.initial()) // 先全量后增量 .build(); // 4. 接入数据流并打印 DataStreamSource<String> mysqlDS = env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), "MySQLSource"); mysqlDS.print(); env.execute("mysql-cdc-datastream-demo"); } }

startupOptions是这个类里最需要讲明白的参数，它有五种取值，选错了行为完全不同：

演示和毕设场景我一般用initial()，因为它一次启动就能把「全量 + 增量」两条路径都演示出来，答辩时有话可讲。earliest()要慎用，它只有在 Binlog 完整保留了库的全部历史时才正确，日志被清理过就会漏数据。

deserializer用JsonDebeziumDeserializationSchema，把 Debezium 的SourceRecord转成 JSON 字符串，打印出来你能直接看到before、after、op三个关键字段，op取值c是新增、u是更新、d是删除、r是快照读。生产上通常会自定义反序列化器，只取after里的字段拼成目标表结构，因为 Doris 的 Sink 一般不需要 before 值。

3.3 Checkpoint 配置决定了断点续传能不能用

上面那段代码里，Checkpoint 相关的有七行，每一行都不是摆设。enableCheckpointing(5000L, EXACTLY_ONCE)设了 5 秒一次的间隔和精确一次语义，间隔太大会导致故障时重复消费的窗口变大，太小会给 HDFS 造成大量小文件。setCheckpointTimeout和setMinPauseBetweenCheckpoints是防止 Checkpoint 堆积的保护。enableExternalizedCheckpoints(RETAIN_ON_CANCELLATION)是最容易被忽略的一条——如果不开，作业被 cancel 之后状态就没了，下次启动只能从头再来。

验证方式我通常走这一套：作业跑起来之后先看 TaskManager 日志确认全量快照开始读；然后在 Web UI 里给作业打一个 Savepoint 到 HDFS；接着在 MySQL 里对demo.t1做一次 insert、一次 update、一次 delete；最后用-s参数从 Savepoint 重启作业。如果日志显示它从快照位点接着读，而不是重新扫全表，说明这条链路的续传能力是真的通了。这个演示动作在答辩时非常加分，因为它证明的不是「能跑」，而是「断了能接上」。

# 触发 savepoint bin/flink savepoint <JobId> hdfs://node-01:8020/flink-cdc/save # 从 savepoint 恢复 bin/flink run -s hdfs://node-01:8020/flink-cdc/save/savepoint-xxxx \ -c com.demo.cdc.FlinkCDCDataStreamTest ./flink-cdc-demo.jar

4. Flink SQL 与 3.0 Pipeline：从写代码到写 YAML 的演进

DataStream 方式把逻辑写死在 Java 里，改一张表就要重新编译打包。Flink SQL 方式把源表定义抽成 DDL，3.0 的 Pipeline 方式更进一步，整条链路用一个 YAML 描述，连 Flink 作业代码都不用写。三种方式的能力是递进的，选择哪种取决于你要交付的是「一个能改表的工具」还是「一个能讲清原理的代码」。

4.1 Flink SQL 建 MySQL CDC 源表

SQL 方式的核心就是一段带WITH参数的 DDL，把连接信息写进表属性里：

-- 在 Flink SQL 里注册一张映射到 MySQL 的源表 CREATE TABLE t1 ( id STRING, name STRING, PRIMARY KEY (id) NOT ENFORCED -- 主键必须声明，否则 upsert 语义无从谈起 ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'node-01', 'port' = '3306', 'username' = 'cdc_user', 'password' = 'Your@Passw0rd', 'database-name' = 'demo', 'table-name' = 't1' ); -- 直接查，结果会持续更新 SELECT * FROM t1;

这段 DDL 里唯一需要强调的约束是PRIMARY KEY ... NOT ENFORCED。它告诉 Flink 这张表的主键是什么，从而决定下游收到的是 insert 流还是 upsert 流。如果不声明主键，Flink 会把它当追加流处理，同一行更新多次就会在下游产生多条重复记录，这个坑非常隐蔽——表面上任务不报错，数据也在涨，但你去 Doris 里 count 一下就会发现行数对不上。我在第一次做这个演示时就栽在这上面，查了半天以为是 Sink 的幂等问题。

SQL 方式跑起来和 DataStream 一样：本地table.execute().print()可以直接看结果，打成 jar 上传集群用flink run提交，行为和 DataStream 版本一致。

4.2 3.0 的 Pipeline：一条 YAML 描述整条链路

3.0 版本引入的 Pipeline 是这个版本最大的变化，它把「源 → 目标」的同步做成声明式配置，你只需要提供一个 YAML，用flink-cdc.sh提交，不需要写一行 Java。部署上要先解压flink-cdc-3.0.0-bin.tar.gz，然后把两个连接器 jar 丢进 lib 目录：flink-cdc-pipeline-connector-mysql-3.0.0.jar和flink-cdc-pipeline-connector-doris-3.0.0.jar。少放一个，启动时会直接告诉你找不到对应的 connector 类型。

配置文件长这样：

# job/mysql-to-doris.yaml source: type: mysql hostname: node-01 port: 3306 username: cdc_user password: "Your@Passw0rd" tables: demo.\.* # 正则匹配整库，demo 下所有表都同步 server-id: 5400-5404 # 给伪从库分配一个 id 区间，避免与已有从库冲突 server-time-zone: UTC+8 # 时区必须与源库一致，否则时间字段会整体偏移 sink: type: doris fenodes: node-01:7030 # FE 的 HTTP 端口 username: root password: "Your@Passw0rd" table.create.properties.light_schema_change: true # 支持加列这类轻量改表 table.create.properties.replication_num: 1 # 副本数不能超过 BE 数量 pipeline: name: Sync MySQL Database to Doris parallelism: 1

三个段落各有讲究。source段的tables用的是 Java 正则，demo.\.*表示 demo 库下所有表，注意那个反斜杠转义，写成demo.*有的版本能识别有的不能，稳妥起见按正则写。server-id给的是一个区间，Pipeline 会从这个区间里给每个并行读取任务分配一个 id，如果和现有主从复制的 id 撞了，MySQL 会拒绝连接，日志里的表现是「A slave with the same server_uuid/server_id as this slave has connected to the master」。server-time-zone是个血泪项，源库是东八区而连接器默认 UTC 时，所有datetime字段进 Doris 之后会差 8 小时，这种错误不会报错，只会让你在核对数据时怀疑人生。

sink段的replication_num必须小于等于 BE 的节点数。只有一个 BE 却写 3，建表语句直接失败，Pipeline 启动后卡在建表阶段。

启动流程是有顺序的，顺序错了会报连接不上：

# 1. Flink 侧打开 Checkpoint，Pipeline 依赖它做位点管理 vim conf/flink-conf.yaml # 加入：execution.checkpointing.interval: 5000 # 分发到其他节点后启动集群 bin/start-cluster.sh # 2. 起 Doris 的 FE 和 BE bin/start_fe.sh bin/start_be.sh # 3. 在 Doris 里建好目标库（库名要与源库对应，或通过 route 指定） mysql -uroot -p -P9030 -hnode-01 -e "CREATE DATABASE IF NOT EXISTS demo;" # 4. 提交整条同步任务 bin/flink-cdc.sh job/mysql-to-doris.yaml

提交之后去 Doris 里刷新demo库，应该能看到三张表连同数据一起被创建出来。接着在 MySQL 里对这几张表做新增、修改、加列操作，再回 Doris 刷新，能同步过来就说明链路通了。加列这一步是重点，它验证的是light_schema_change这个参数生效没有。

4.3 路由配置：把源库的表写到目标端的另一个库里

路由（route）解决的是「源库名和目标库名不一致」的问题。比如源端有个demo_route库，你想把它同步到 Doris 的doris_demo_route库下面，源端的配置不用动，只需要在 YAML 里加一段 route 规则。实际效果是源端demo_route.t1的数据会被写进doris_demo_route.t1，字段结构保持不变。

这个能力在多环境隔离时很有用：测试库和生产库表结构一样，但目标端要分开，只用一份 Pipeline 配置加不同的 route 规则就能切换目标。我通常会在方案里把路由规则单独列一张表，写清「源库.源表 → 目标库.目标表」的映射，评审时比看 YAML 直观得多。路由规则本身也是正则匹配，多个规则按顺序匹配，第一条命中就生效，所以规则的顺序不能随意调。

5. 排查手册：这条链路上最容易翻车的五个点

这条链路我在不同环境里搭过四次，前三次踩的坑基本都不重样。下面这五条是复现率最高的，每条我都按「现象、原因、解决」写，方便你直接对照日志定位。

5.1 源端与连接器类故障

第一条：任务起来了，但只同步了全量，后面的增量一条都不来。

现象是启动时能看到快照阶段把历史数据读完了，之后日志安静下来，MySQL 里再怎么改数据，目标端也没反应。原因基本是两个：一是 MySQL 的binlog_format不是row，语句模式下 CDC 拿不到行级变更；二是binlog-do-db白名单里没写当前要同步的库，日志里压根没有这个库的变更记录。排查方法很直接，在 MySQL 上执行SHOW VARIABLES LIKE 'binlog_format';和SHOW VARIABLES LIKE 'binlog_do_db';，两个值对不上就改配置重启。改完重启 MySQL 之后，Flink 作业也要重启，因为连接器缓存的位点信息已经失效了。

第二条：作业反复重启，日志里报 server-id 冲突或者连接被踢。

现象是作业跑几十秒就挂，重启策略触发，再跑再挂，TaskManager 日志里能看到「connection reset」或者「slave with the same server_id」。原因是同一个 MySQL 上跑了多个 CDC 任务，各自的server-id撞了，MySQL 认为这是同一个从库在重复连接，会把前一个踢掉。解决办法是给每个任务分配不重叠的 id 区间，比如任务 A 用5400-5404，任务 B 用5405-5409，写在 YAML 的server-id里或者 DataStream 的 builder 上。这个坑在多任务并行同步不同库时特别容易出现，配置里看不出问题，只有跑起来互相踢才发现。

第三条：时间字段全都差了 8 小时。

现象是数据能同步过来，结构也对，但所有datetime、timestamp类型的字段值都比源库小 8 小时（或者大 8 小时）。原因是连接器的时区默认是 UTC，而源库在东八区，读取时按 UTC 解释了一遍。解决就是在源端配置里显式写上server-time-zone: UTC+8，DataStream 方式则在 builder 上调用对应方法设置时区。这个错误最难受的地方是它不报错，数据看着是「正常」的，只有你拿一条已知时间的数据去比对才会发现。

5.2 目标端与作业运行类故障

第四条：Doris 建表失败，任务卡在启动阶段不动。

现象是flink-cdc.sh提交之后，作业状态一直是初始化，日志里能看到建表的 SQL 报错。原因通常是replication_num设成了 3 而实际只有一个 BE 节点，Doris 要求副本数不能超过可用 BE 数。另一个可能原因是light_schema_change没开，源端一张带特殊类型的表转成 Doris 的建表语句时失败了。解决是把table.create.properties.replication_num调成 1（单 BE 环境），并确认light_schema_change: true已经写上。如果还不行，就在 Doris 里手工把目标库建好，让 Sink 只负责写入不负责建表，先跑通再说。

第五条：源端加了一列，Doris 那边没跟着变，后续写入开始报错。

现象是业务给 MySQL 的表加了个字段，增量数据继续同步，但 Doris 里的表结构没变，接着写入就开始报列数不匹配。原因是表结构变更（schema evolution）需要 Sink 端显式支持，Doris 的轻量改表能力要靠light_schema_change打开。解决是在 YAML 的 sink 段加上table.create.properties.light_schema_change: true，并且确认 Doris 版本支持这个特性。加完之后重新提交任务，源端加列会被同步过去。这里有个前提：只有加列这类轻量变更能被自动处理，改字段类型、删列这类操作在多数版本上仍然需要人工介入，方案里要把这个边界写清楚，别让业务方以为「开了这个参数就什么都能自动改」。

6. 验收与答辩：用一次「删改加列」把链路证明清楚

这条链路的验收不能只看「数据过去了」，要证明的是「变更被完整、有序、可恢复地搬过去了」。我一般用一页验收清单走完，六个动作，每个动作都有明确的预期结果，直接拿去当答辩演示脚本也行。

第 3 条和第 6 条是最能体现方案深度的两个点。第 3 条证明你理解 upsert 流和追加流的区别，第 6 条证明你理解 Checkpoint 在这条链路里的位置。答辩时如果被追问「为什么不能只用 Flink SQL」，可以答：SQL 方式适合单表快速验证，Pipeline 方式适合整库批量同步且支持 schema 变更，DataStream 方式适合需要精细控制反序列化和限流的场景，三种不是替代关系，是不同交付粒度。

演示脚本我建议写成 shell，一条命令跑完一轮变更，比手动敲 MySQL 命令稳，也方便重复：

#!/bin/bash # 演示用的变更脚本，依次触发增、改、删，每步间隔 5 秒观察同步效果 set -e MYSQL="mysql -h node-01 -ucdc_user -pYour@Passw0rd demo" $MYSQL -e "INSERT INTO t1 VALUES('2001','test_insert');" sleep 5 $MYSQL -e "UPDATE t1 SET name='test_update' WHERE id='2001';" sleep 5 $MYSQL -e "DELETE FROM t1 WHERE id='2001';" # 加列验证，注意这一步在部分 Doris 版本上需要 light_schema_change 支持 $MYSQL -e "ALTER TABLE t2 ADD COLUMN age INT DEFAULT 0;" $MYSQL -e "INSERT INTO t2 VALUES('2002','test_col',18);"

跑完这一轮，去 Doris 里依次SELECT三张表，对照上表的预期结果逐条打勾。如果第 5 条不通，说明 schema 变更没生效，回第 5 章第四条排查；如果第 6 条恢复后作业重新扫了全表，说明 Checkpoint 没存住位点，回头检查setCheckpointStorage的 HDFS 路径是否可写、enableExternalizedCheckpoints是否开了。

从那以后我每次交付这类实时同步链路，都会强制走一遍这套「删改加列 + Savepoint 恢复」的组合验证，哪怕甲方只要求「能同步就行」——因为这条链路的失败模式几乎都是静默的，数据少了、时间偏了、重启重扫了，界面上都看不出异常，只有主动去触发这些边界动作，才能把黑匣子打开。希望这套流程能帮到你。

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