企业官网建设流程全解析

简介：这是一份面向有一定Flink基础的大数据研发人员的Flink CDC 3.0技术文档，由尚硅谷研究院编写，系统讲解CDC概念、基于查询与基于Binlog两种方式的差异，并重点演示flink-cdc-connectors组件如何从MySQL等数据库读取全量与增量变更数据。文档包含完整案例实操，覆盖DataStream与Flink SQL两种应用方式，逐步指导完成MySQL到Doris的Streaming ETL同步，涉及环境搭建、配置编写、任务启动与测试验证，同时强调开启Binlog、设置检查点等关键细节。资源为单个docx文档，大小145KB，内容结构清晰，配有代码示例和操作指南，适合用于理解原理、上手实践及排查同步问题。该文档已有1788人浏览学习，适合需要快速掌握实时数据同步能力的工程师参考。

1. Flink CDC 3.0：MySQL 到 Doris 实时同步的全流程拆解

以前做实时数仓，MySQL 到 Doris 的数据同步是最折磨人的环节：定时任务轮询延迟高，DataX 抽全量又给源库加压力，业务侧一问数据怎么还没到，只能干瞪眼。这套 Flink CDC 3.0 全流程讲义，把「MySQL 数据实时捕获与同步至 Doris」这件事从原理到落地讲得很透——基于 Binlog 的变更捕获机制、3.0 版本主推的 YAML 管线式 StreamingETL、DataStream 和 FlinkSQL 两种接入方式，一条命令就可以把整库变更流进 Doris，还支持路由改名和断点续传。适合有 Flink 基础、正在搭实时数仓或做数据同步的工程师，照着步骤能复现，也能把底层逻辑看明白。

2. CDC 原理与选型：基于查询和基于 Binlog 的取舍

2.1 两种 CDC 模式的本质差异

CDC（Change Data Capture）的核心思想不复杂：监测并捕获数据库的变动——包括数据行的插入、更新、删除，也包括表结构的变化——按发生顺序完整记录下来，写入消息中间件供其他服务订阅和消费。真正拉开差距的是「怎么捕获」这件事，业界分成两条完全不同的路线。

基于查询的 CDC 是轮询派：周期性执行 SQL 查询，把变化的数据捞出来。开源产品里 Sqoop、DataX 属于这一类。它的优点是实现简单，对数据库版本没有要求；缺点是只能查出「当前状态」，中间被修改过的数据直接丢了，每次全表扫描还会给源库造成不小的压力，延迟也高。

基于 Binlog 的 CDC 是日志派：直接读 MySQL 的二进制日志，依靠 binlog_format=row 记录每一行变更前后的完整镜像。开源产品有 Canal、Maxwell、Debezium。这类方式能拿到完整的历史变更序列，延迟低，也不增加数据库查询压力。下面这张对照表把差异列得很清楚，我做选型时基本就看这一张表：

选型结论也很直接：对延迟敏感、要拿全部增删改记录的场景，直接走 Binlog 路线；临时导数据、离线批处理，用查询型工具反而更省事，毕竟不用动数据库配置。

2.2 flink-cdc-connectors：Flink 生态里的 CDC 组件

Flink 社区在 CDC 这条路上做了一个很关键的组件：flink-cdc-connectors。它是一个 source 组件，可以直接从 MySQL、PostgreSQL 等数据库读取数据，而且天然支持「全量 + 增量」一起读——第一次启动做全量快照，快照完成后无缝切换到 Binlog 增量读取，用户侧完全不需要关心快照和增量之间的衔接问题。

这个设计对使用者非常友好。以前做数据同步，全量和增量是两个任务，中间还有数据窗口期；用 Flink CDC，一个 source 就把两段串起来了。再配合 Flink 的 Checkpoint 机制，Binlog 的读取位置会作为状态保存下来，任务挂了可以从检查点续跑，这就是后面要展开讲的断点续传。组件本身是开源的，Maven 坐标是 com.ververica:flink-connector-mysql-cdc，版本跟 Flink 版本有对应关系，第 3 章会给出具体搭配。

2.3 开启 Binlog：让 MySQL 先「开口说话」

Binlog 在 MySQL 8.0 里默认是开启的，但很多自建实例或云上实例出于性能考虑会关掉。Flink CDC 要读变更，第一步就是确认目标库的 Binlog 配置。讲义里的做法是在 /etc/my.cnf 里加一段配置：

# 数据库实例唯一 id，复制拓扑中不能重复 server-id = 1 # 开启 binlog，该参数值会作为 binlog 文件名前缀 log-bin=mysql-bin # binlog 格式必须为 row，CDC 工具需要行级镜像 binlog_format=row # 只记录指定库的变更，需根据实际情况修改 binlog-do-db=test binlog-do-db=test_route

四个参数各有用处：server-id 是实例在复制拓扑里的唯一标识，同一套环境里不能重复；log-bin 开启后 MySQL 会在数据目录里滚动生成 mysql-bin.000001 这类文件；binlog_format=row 最关键，它决定 Binlog 记录的是每行数据变更前后的完整镜像而不是 SQL 语句，CDC 工具只有拿到 row 格式才能还原出 update 和 delete 的具体内容；binlog-do-db 做库级过滤，只记录 test 和 test_route 两个库的变更，省磁盘也省同步开销。

改完配置文件必须重启 MySQL 才生效。验证方式也不复杂，登录 MySQL 执行 SHOW VARIABLES LIKE 'log_bin';，看到 ON 说明已开启；再执行 SHOW VARIABLES LIKE 'binlog_format';，确认是 ROW。

注意：binlog-do-db 是个容易埋雷的参数，以后新加库进来必须改配置再重启，否则新库的变更不会进 Binlog。避坑章节我会再展开。

3. DataStream 方式实操：Checkpoint 与 Savepoint 断点续传

3.1 Maven 依赖与打包配置

DataStream 方式是最贴近 Flink 编程模型的写法：把 CDC Source 当成一个普通的 DataStream Source，用 env.fromSource 接进来，后面接任意算子和 sink。讲义里的依赖配置如下：

<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 DataStream 核心依赖 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink-version}</version> </dependency> <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 相关依赖，FlinkSQL 方式会复用同一套 --> <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> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-base</artifactId> <version>${flink-version}</version> </dependency> <!-- CDC 核心：MySQL Source --> <dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>3.0.0</version> </dependency> <!-- MySQL JDBC 驱动 --> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>8.0.31</version> </dependency> </dependencies> <build> <plugins> <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 的版本要跟 Flink 版本匹配，讲义里 Flink 用 1.18.0、CDC connector 用 3.0.0，这组搭配是验证过的。mysql-connector-java 用 8.0.31，对应 MySQL 8.0 服务端。打包用 maven-assembly-plugin 生成 jar-with-dependencies 的 fat jar——任务要提交到 Flink 集群上跑，集群节点不一定有这些依赖，打成一个大包是最省心的做法。

3.2 核心代码：MySqlSource 构建与 Checkpoint 配置

讲义里的 DataStream 示例代码有两大重点：Checkpoint 配置和 MySqlSource 的构建参数。完整代码长这样：

import com.ververica.cdc.connectors.mysql.source.MySqlSource; import com.ververica.cdc.connectors.mysql.table.StartupOptions; import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.restartstrategy.RestartStrategies; import org.apache.flink.api.common.time.Time; import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; import org.apache.flink.streaming.api.CheckpointingMode; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.CheckpointConfig; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class FlinkCDCDataStreamTest { public static void main(String[] args) throws Exception { // 1. 流处理环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. 开启 Checkpoint：Flink-CDC 把 binlog 读取位置存进状态， // 想断点续传必须从 Checkpoint 或 Savepoint 启动 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); // 任务取消时保留最后一次 CK 数据，方便手动恢复 env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 失败重启策略：1 天内最多重启 3 次，间隔 1 分钟 env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); // 状态后端：HashMap + HDFS 存储 env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC"); // 访问 HDFS 的用户名 System.setProperty("HADOOP_USER_NAME", "atguigu"); // 3. 构建 MySqlSource MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("hadoop103") .port(3306) .databaseList("test") // 捕获的数据库 .tableList("test.t1") // 捕获的表 .username("root") .password("000000") .deserializer(new JsonDebeziumDeserializationSchema()) // SourceRecord 转 JSON String .startupOptions(StartupOptions.initial()) .build(); // 4. 接入 env，打印输出 DataStreamSource<String> mysqlDS = env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), "MysqlSource"); mysqlDS.print(); env.execute(); } }

代码里 Checkpoint 部分的注释值得细看。CDC Source 的断点续传依赖状态：Flink 会把当前读到的 binlog 文件名和 offset 存进 Checkpoint，任务失败后从最近一次成功的 Checkpoint 恢复，就能接着上次的位置继续读，不丢也不重。所以 env.enableCheckpointing(3000L, EXACTLY_ONCE) 不是锦上添花，而是 CDC 任务的刚需；setCheckpointTimeout 控制单次 Checkpoint 超时时间，minPauseBetweenCheckpoints 防止 Checkpoint 过于频繁互相挤占；enableExternalizedCheckpoints 设置取消时保留最后一次 CK，是手动恢复的前提。

MySqlSource 的构建参数里，databaseList 和 tableList 支持库表过滤，tableList 的格式是「库名.表名」；deserializer 用了 JsonDebeziumDeserializationSchema，把 Debezium 的 SourceRecord 转成 JSON 字符串，print 出来就能看到完整的变更事件。startupOptions 决定任务启动时的读取位置，五种模式差别很大：

提示：initial 是最省心的选择，兼顾存量数据和增量数据；latest 适合只关心启动后变化的场景，但它不会帮你补历史数据，很容易误用。

3.3 提交任务与 Savepoint 恢复验证

讲义里的测试流程是一条完整链路，值得照着走一遍。先把项目打成 fat jar 传到集群节点，然后依次启动 HDFS 和 Flink 集群：

# 启动 HDFS start-dfs.sh # 启动 Flink 集群 bin/start-cluster.sh # 提交 CDC 任务到集群 bin/flink run -m hadoop102:8081 -c com.atguigu.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar

任务跑起来后观察 TaskManager 日志，能看到它从 MySQL 的 test.t1 表把全量数据读出来。这时候做断点续传验证：先给任务创建 Savepoint，然后在 WebUI 上取消任务，接着在 MySQL 里修改或删除数据，最后从 Savepoint 重启：

# 创建 Savepoint bin/flink savepoint <JobId> hdfs://hadoop102:8020/flinkCDC/save # 从 Savepoint 恢复重启 bin/flink run -s hdfs://hadoop102:8020/flinkCDC/save/savepoint-xxxx \ -c com.atguigu.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar

恢复后观察日志，会发现程序从上次保存的 binlog 位置继续读，取消期间产生的变更也能补上——这就是 CDC 任务最核心的可靠性保证。实战里我还会多做一步：取消前记下 MySQL 当前数据行数，恢复后对比读出来的行数和 binlog 事件数，确认不丢才算验证通过。

4. FlinkSQL 与 Pipeline 方式：从 SQL DDL 到 YAML 配置

4.1 FlinkSQL 方式：mysql-cdc Connector 的建表 DDL

DataStream 方式灵活，但代码量大。如果只是想把 MySQL 表映射成 Flink 表做查询或同步，FlinkSQL 方式显然更轻——不需要写 Source 构建逻辑，一张 CREATE TABLE 的 DDL 就搞定了：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; public class FlinkSQLTest { public static void main(String[] args) { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); tableEnv.executeSql( "CREATE TABLE t1 (" + " id string PRIMARY KEY NOT ENFORCED," + " name string" + ") WITH (" + " 'connector' = 'mysql-cdc'," + " 'hostname' = 'hadoop103'," + " 'port' = '3306'," + " 'username' = 'root'," + " 'password' = '000000'," + " 'database-name' = 'test'," + " 'table-name' = 't1'" + ")"); Table table = tableEnv.sqlQuery("select * from t1"); table.execute().print(); } }

这段代码里有两个容易忽略的细节。第一个是 PRIMARY KEY NOT ENFORCED，Flink Table 不会强制主键约束，但这个声明对 CDC 很重要——update 和 delete 事件要靠它定位要更新的行，Source 内部会用主键做变更数据的合并。第二个是 WITH 里的 connector = 'mysql-cdc'，FlinkSQL 通过这个标识自动找到 flink-connector-mysql-cdc 插件，不需要像 DataStream 那样手动 new Source。hostname、port、database-name、table-name 这些参数跟 DataStream 的 builder 一一对应，只是换成了 SQL 风格。

FlinkSQL 方式的测试也更简单，本地 IDE 直接跑 main 方法就能看到结果；提交到集群的方式跟 DataStream 完全一样，打成 fat jar 用 flink run 提交即可。

4.2 MySQL 到 Doris 的 Streaming ETL：YAML 配置拆解

3.0 版本最大的变化是引入 Pipeline（管线）模式：不用写 Java 代码，一份 YAML 配置文件描述 source、sink、route、pipeline，然后用 flink-cdc.sh 直接启动。讲义里 MySQL 到 Doris 的同步变更案例，配置如下：

source: type: mysql hostname: node01 port: 3306 username: root password: "123" tables: test.\.* # 正则匹配 test 库下所有表 server-id: 5400-5404 # 一组 server-id，供并行读取 binlog server-time-zone: UTC+8 sink: type: doris fenodes: node01:7030 # Doris FE 地址 username: root password: "000000" table.create.properties.light_schema_change: true # 开启轻量 Schema 变更 table.create.properties.replication_num: 1 # 副本数，测试环境设 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1

每个字段都不是随便写的。tables 用 test..* 这种正则表达式，表示捕获 test 库下所有表，比 DataStream 里写死 test.t1 灵活得多，这也是 Pipeline 模式的主打能力——整库同步。server-id 配的是区间 5400-5404 而不是单个值，因为 Pipeline 默认会有多个并行子任务去读 binlog，每个子任务需要独立的 server-id，配区间避免冲突。sink 部分两个 table.create.properties 是给 Doris 建表用的：light_schema_change 让 Doris 支持轻量级 schema 变更（加列、改列不需要重写数据文件），replication_num 控制副本数，测试环境设 1 可以省存储。

启动流程讲义写得很细。先要在 Flink 的 conf/flink-conf.yaml 里打开 Checkpoint：

# 1. 修改 Flink 配置，每隔 5 秒做一次 Checkpoint vim conf/flink-conf.yaml # 在文件中加入：execution.checkpointing.interval: 5000 # 2. 分发配置到所有 Flink 节点后启动集群 bin/start-cluster.sh # 3. 启动 Doris FE 和 BE bin/start_fe.sh bin/start_be.sh # 4. 连接 Doris 并创建目标库 mysql -uroot -p000000 -P9030 -hhadoop102 create database test; # 5. 启动 Flink CDC 同步任务 bin/flink-cdc.sh job/mysql-to-doris.yaml

注意：flink-cdc.sh 会把任务提交到 Flink 集群，启动前先确认 Flink 的 Checkpoint 配置已生效，否则任务挂了没法自动恢复。

任务起来后刷新 Doris 的 test 库，能看到 MySQL test 库下的 t1、t2、t3 三张表已经在 Doris 里建好，数据也同步过来了。这时候在 MySQL 侧做新增、修改、甚至新增列的操作，再刷新 Doris，会发现数据变化和表结构变化都跟着过来了。

4.3 同步变更与路由变更：两种管线的区别

Pipeline 模式里「同步变更」和「路由变更」是两种常见场景，讲义各给了一个案例。同步变更对应上面的配置：源库 test 同步到 Doris 后库名不变、表名不变，适合库名一致的场景。路由变更则要解决「库名或表名不一样」的问题，比如 MySQL 的 test_route 库要落到 Doris 的 doris_test_route 库：

source: type: mysql hostname: hadoop103 port: 3306 username: root password: "000000" tables: test_route.\.* server-id: 5400-5404 server-time-zone: UTC+8 sink: type: doris fenodes: node01:7030 username: root password: "000000" # 路由规则：源表映射到目标表，实现库名/表名改名 route: - source-table: test_route.t1 sink-table: doris_test_route.t1 - source-table: test_route.t2 sink-table: doris_test_route.t2 - source-table: test_route.t3 sink-table: doris_test_route.t3 pipeline: name: Route MySQL Database to Doris parallelism: 1

route 段就是改名规则，source-table 和 sink-table 一一对应。这种场景在真实业务里非常常见：源库是业务库、表名混乱，目标库要按数仓规范重新命名；或者多个源库要汇聚到一个目标库，路由规则可以解决同名冲突。验证方式跟同步变更一样，在 MySQL 的 test_route 库里改数据、加列，然后刷新 Doris 的 doris_test_route 库观察变化。

5. 避坑与排查：Binlog、Checkpoint、Doris 侧的常见坑

这套讲义我前后跑了两遍，第一遍照着敲完，第二遍故意制造故障验证恢复能力。中间踩了不少坑，挑五个最典型的记在这里，按「现象 → 原因 → 解决」的顺序写，遇到同类问题可以直接对照排查。

5.1 Binlog 环节：配置不生效与 server-id 冲突

第一个坑：改完 my.cnf 重启 MySQL，Flink CDC 任务还是读不到增量变更。现象是任务能启动、全量数据也正常打印，但在 MySQL 里 insert 数据后，TaskManager 日志没有任何反应。原因排查下来有两个方向：最常见的是 binlog_format 没生效，MySQL 8.0 默认格式虽然是 ROW，但很多从 5.7 升上来的实例还停在 STATEMENT，CDC 工具拿不到行镜像；另一个是 binlog-do-db 只配了 test 和 test_route，而实际操作的表不在这些库里。解决方法是登录 MySQL 执行 SHOW VARIABLES LIKE 'binlog_format'，先确认是 ROW，再确认 binlog-do-db 覆盖了目标库，最后重启 mysqld 让它真正加载新配置。这三步按顺序做完，九成问题能解。

第二个坑：多个 CDC 任务连同一个 MySQL，跑一会儿某个任务就报错断开，日志里出现 Got error reading packet from server。原因是这些任务用了相同的 server-id。MySQL 的 binlog dump 协议要求每个消费端有唯一 server-id，重复 id 会被后来的连接挤掉，表现就是先启动的任务突然断开。解决方法是给每个 CDC 任务分配独立的 server-id，或者像讲义里那样配一个区间 5400-5404。注意区间大小要大于任务的实际并行度，否则并行子任务之间还是会撞。

5.2 Checkpoint 环节：HDFS 权限与恢复路径

第三个坑：任务启动后报 HDFS 权限错误，Checkpoint 写不进去，任务反复重启。现象是日志里出现 Permission denied，指向 hdfs://hadoop102:8020/flinkCDC 这个路径。原因是提交任务的用户和 HDFS 上的目录属主不一致。解决方法是代码里加 System.setProperty("HADOOP_USER_NAME", "atguigu")，并且保证这个路径在 HDFS 上已创建。我一般把这行放在 main 方法最前面，避免任务跑起来才报错；另外，如果不用 HDFS 而是本地文件系统，要确认所有 TaskManager 节点都能访问同一个路径，否则状态后端不一致也会翻车。

第四个坑：从 Savepoint 恢复后读取位置不对，要么重复消费一段，要么丢了取消期间的变更。现象是恢复后日志里打印的 binlog offset 和取消前对不上。原因多半是 -s 参数指定错了目录——savepoint 目录下会有多个 savepoint-xxxx 子目录，每个对应一次手动 savepoint，命令行里手敲路径很容易指错。解决方法是创建 savepoint 后，把命令输出的完整路径复制下来，恢复时原样粘贴，不要自己拼。另外，恢复用的 jar 包必须和创建 savepoint 时是同一份代码，类结构变了状态恢复会直接失败。

5.3 Doris 环节：表结构变更不生效

第五个坑：MySQL 侧 ALTER TABLE 加了一列，Doris 里的表结构没跟着变，新列的数据也没进来。现象是 Pipeline 任务运行正常，Doris 查询却看不到新列。原因是 sink 配置里没开启 light_schema_change，Doris 默认的 schema change 流程是重写数据文件，在 CDC 实时 DDL 场景下往往不会自动触发。解决方法是确认 sink 配置里 table.create.properties.light_schema_change: true，同时确认 Doris 版本支持轻量 schema 变更。如果配置没问题还是不生效，先手动在 Doris 执行 ALTER TABLE 补上列，让后续增量能正常写入，再来排查 Pipeline 侧 DDL 事件是不是被过滤掉了。

6. 二次开发：从单表同步到多库多表动态路由

讲义里的案例停留在单表捕获、固定路由，实际业务里需求会往前走好几步。我基于这套代码做过一轮改造，几个方向可以直接复用。

第一，把 DataStream 方式里写死的 databaseList("test") 改成可配置。做法是把库表列表、用户名密码、启动模式全部抽到 properties 文件或启动参数里，main 方法只做组装。这样同一个 jar 包，换一套参数就能跑不同数据源，不用每次改代码重新打包。

第二，自定义 DeserializationSchema。JsonDebeziumDeserializationSchema 输出的 JSON 里带 schema 描述，下游 Kafka 或 Doris sink 不一定需要。常见做法是实现 DebeziumDeserializationSchema 接口，把 SourceRecord 解析成业务字段的 POJO，顺便把 op 字段（c/u/d 对应增删改）单独拎出来，方便下游做分流处理。

第三，Pipeline 模式下把正则表匹配和路由规则做成动态的。tables 用 test..* 已经能覆盖整库，但路由规则如果新增表，YAML 里要手动加一条。我改造时写了个脚本，扫描 MySQL 里的表清单自动生成 route 段，再调用 flink-cdc.sh 重启任务，整个过程不用手改配置。

验证方法也固定成了一套：先核对 Binlog 参数，再起任务看全量，然后手动做一次 insert、一次 update、一次 delete、一次 add column，最后从 Savepoint 恢复一次确认位点正确。这套验证做完，任务上线基本不会翻车。从那以后我每次上线 CDC 任务都强制走一遍这套流程，哪怕只是改一个表名。这套讲义把从 CDC 原理到 Pipeline 落地的完整链路都覆盖了，代码和 YAML 配置都可以直接抄，希望帮到你。

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