简介面向具备一定 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 这类库读全量快照和增量变更。三段式的划分是这样的段位承担职责对应组件关键产出源端捕获读全量快照 持续追 BinlogMySQL flink-connector-mysql-cdc带 before/after 的变更事件流中间处理反序列化、字段映射、路由、限流Flink 作业DataStream / SQL / Pipeline结构化、可写入的行级变更目标端写入批量攒批、幂等写入、建表与改表Doris Sink ConnectorDoris 中的明细表把这三段分开之后很多设计决策就变得有依据了。比如「要不要在中间加一层 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 必须为 rowstatement/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.4FE 的 MySQL 协议端口 9030。JDK 用 8Maven 编译的 source/target 都设 8。Doris 这边至少一个 FE 加一个 BEBE 的数量会直接影响后面建表时副本数的设置这个坑我在第 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.ververicaproperties maven.compiler.source8/maven.compiler.source maven.compiler.target8/maven.compiler.target flink-version1.18.0/flink-version project.build.sourceEncodingUTF-8/project.build.sourceEncoding /properties dependencies !-- Flink 运行时与流处理 API -- dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink-version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink-version}/version /dependency !-- Table APIFlink SQL 方式会用到 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-table-planner_2.12/artifactId version${flink-version}/version /dependency !-- MySQL CDC Source注意 groupId 是 com.ververica -- dependency groupIdcom.ververica/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.0.0/version /dependency !-- MySQL 驱动版本要与源库大版本对齐 -- dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId version8.0.31/version /dependency /dependencies build plugins !-- 打 fat jar否则集群上跑会报 ClassNotFound -- plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-assembly-plugin/artifactId version3.0.0/version configuration descriptorRefs descriptorRefjar-with-dependencies/descriptorRef /descriptorRefs /configuration executions execution idmake-assembly/id phasepackage/phase goalsgoalsingle/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 MySqlSourceString mySqlSource MySqlSource.Stringbuilder() .hostname(node-01) .port(3306) .databaseList(demo) // 捕获的库 .tableList(demo.t1) // 捕获的表注意要带库名前缀 .username(cdc_user) .password(YourPassw0rd) .deserializer(new JsonDebeziumDeserializationSchema()) // SourceRecord 转 JSON .startupOptions(StartupOptions.initial()) // 先全量后增量 .build(); // 4. 接入数据流并打印 DataStreamSourceString mysqlDS env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), MySQLSource); mysqlDS.print(); env.execute(mysql-cdc-datastream-demo); } }startupOptions是这个类里最需要讲明白的参数它有五种取值选错了行为完全不同取值首次启动行为适用场景initial()先做全量快照再从快照位点接 Binlog默认选择历史数据也要同步earliest()不做快照从 Binlog 最早位置读源库日志完整保留了全部历史且表很大不想扫全表latest()不做快照只读启动之后的变更目标端已经有全量数据只补增量specificOffset()从指定的 binlog 文件名 位点开始手工修复某段数据timestamp()从指定时间戳开始追补某一天的数据演示和毕设场景我一般用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.jar4. Flink SQL 与 3.0 Pipeline从写代码到写 YAML 的演进DataStream 方式把逻辑写死在 Java 里改一张表就要重新编译打包。Flink SQL 方式把源表定义抽成 DDL3.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 YourPassw0rd, 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: YourPassw0rd tables: demo.\.* # 正则匹配整库demo 下所有表都同步 server-id: 5400-5404 # 给伪从库分配一个 id 区间避免与已有从库冲突 server-time-zone: UTC8 # 时区必须与源库一致否则时间字段会整体偏移 sink: type: doris fenodes: node-01:7030 # FE 的 HTTP 端口 username: root password: YourPassw0rd 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 侧打开 CheckpointPipeline 依赖它做位点管理 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: UTC8DataStream 方式则在 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. 验收与答辩用一次「删改加列」把链路证明清楚这条链路的验收不能只看「数据过去了」要证明的是「变更被完整、有序、可恢复地搬过去了」。我一般用一页验收清单走完六个动作每个动作都有明确的预期结果直接拿去当答辩演示脚本也行。序号操作预期结果证明的能力1首次启动作业目标库自动建表历史数据全量写入全量快照2MySQL 中 INSERT 一行秒级内目标端出现该行增量捕获3MySQL 中 UPDATE 同一行目标端该行值被覆盖不新增行upsert 语义4MySQL 中 DELETE 一行目标端该行消失删除同步5给源表加一列并写值目标端表结构同步新增该列表结构变更6打 Savepoint 后 cancel再恢复作业从断点继续不重扫全表断点续传第 3 条和第 6 条是最能体现方案深度的两个点。第 3 条证明你理解 upsert 流和追加流的区别第 6 条证明你理解 Checkpoint 在这条链路里的位置。答辩时如果被追问「为什么不能只用 Flink SQL」可以答SQL 方式适合单表快速验证Pipeline 方式适合整库批量同步且支持 schema 变更DataStream 方式适合需要精细控制反序列化和限流的场景三种不是替代关系是不同交付粒度。演示脚本我建议写成 shell一条命令跑完一轮变更比手动敲 MySQL 命令稳也方便重复#!/bin/bash # 演示用的变更脚本依次触发增、改、删每步间隔 5 秒观察同步效果 set -e MYSQLmysql -h node-01 -ucdc_user -pYourPassw0rd demo $MYSQL -e INSERT INTO t1 VALUES(2001,test_insert); sleep 5 $MYSQL -e UPDATE t1 SET nametest_update WHERE id2001; sleep 5 $MYSQL -e DELETE FROM t1 WHERE id2001; # 加列验证注意这一步在部分 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 恢复」的组合验证哪怕甲方只要求「能同步就行」——因为这条链路的失败模式几乎都是静默的数据少了、时间偏了、重启重扫了界面上都看不出异常只有主动去触发这些边界动作才能把黑匣子打开。希望这套流程能帮到你。本文还有配套的精品资源点击获取
