简介：面向具备一定 Flink 基础的大数据研发人员，这份 docx 文档围绕 Flink CDC 3.0 展开，聚焦 MySQL 数据实时捕获并同步至 Doris 的完整链路，帮助读者理解变更数据捕获原理并落地 Streaming ETL 场景。内容从 CDC 概念切入，对比基于查询与基于 Binlog 两种方式在延迟、全量捕获和数据库压力上的差异，并结合 flink-cdc-connectors 说明 MySQL、PostgreSQL 等数据源的读取方式。包内共 1 个 docx 文件，压缩包约 145KB，文档给出数据准备、DataStream 与 Flink SQL 应用、MySQL 到 Doris 的环境搭建、配置编写、任务启动和测试步骤，并提示开启 Binlog、设置检查点等关键细节。已有 1789 人学习，适合需要掌握部署流程、保证同步一致性与稳定性的工程师参考实践。

1. 从一张凌晨还在跑的对账表说起

对账报表凌晨两点还在跑批量任务，业务方早上八点就要看昨天下午的订单明细——这种需求下，用 Flink CDC 3.0 把 MySQL 的变更实时同步到 Doris 做 OLAP 查询，基本绕不过去。它直接读 MySQL 的 binlog，把 insert/update/delete 按发生顺序变成一条流，经过 Flink 加工后写进 Doris，能做到秒级可见。适合谁？手里有 Flink 基础、要搭一条 MySQL 到 Doris 实时同步链路的后端或大数据开发。这份资料从 CDC 概念讲到 DataStream、FlinkSQL 两种写法，再落到 3.0 的 Pipeline yaml 配置，属于能照着敲完就跑起来的那种。

2. 架构与选型：为什么是 Binlog + Flink + Doris 这条链路

动手写代码前，得先想清楚为什么不直接用 DataX 定时抽一把。选型错了，后面调参数全是白费力气。

2.1 基于查询和基于 Binlog，差的不是一点延迟

CDC 这个概念本身不复杂，核心是监测数据库的变动，把插入、更新、删除按发生顺序完整记下来，写到消息中间件或直接投给下游。难的是“怎么拿到这些变动”，业内只有两条路。

一条是基于查询：定时跑一条select * from t where update_time > ?，靠时间戳或自增 ID 捞增量。Sqoop、DataX 走的就是这条路，执行模式是 Batch，也就是一批一批来。它的毛病很直接：物理删除捞不到，一条记录在两次查询之间改了三次，你只能看到最后一次，中间态全丢了；为了不漏数据，查询间隔不敢拉太长，库的压力就上来了。

另一条是基于 Binlog：订阅 MySQL 的二进制日志，数据库写什么你就收到什么。Canal、Maxwell、Debezium 都是这个路子，执行模式是 Streaming，延迟低，删除操作天然带op = d标记，不会丢。代价是要开 binlog、要处理位点、要占一个 server-id。

Flink CDC 属于后者。Flink 社区把 Debezium 嵌进了 Source 组件里，flink-cdc-connectors可以直接从 MySQL、PostgreSQL 读全量数据和增量变更，全量阶段按主键分片并行扫，扫完自动切到 binlog 阶段接着读，这个“无锁快照 + 增量衔接”是整个组件最值钱的地方。

提示：全量阶段默认会对表加轻量级锁来拿一致性位点，大表上要留意snapshot.select.statement.overrides和锁等待超时，别在业务高峰期第一次起任务。

2.2 Flink CDC 3.0 的三条落地路径

同一套 connector，Flink CDC 提供了三种写法，很多人一开始就选错，写到一半发现路走不通再返工。

第一种是DataStream API。手工MySqlSource.builder()构造 Source，env.fromSource()接进流里，后面能接任意算子做过滤、打宽、聚合。自由度最高，但整库同步几十张表的时候，你得自己写循环、自己拼 sink，代码量上去了。

第二种是FlinkSQL。一句CREATE TABLE ... WITH ('connector' = 'mysql-cdc', ...)就把 Source 声明出来了，配合tableEnv.sqlQuery或者直接INSERT INTO写目标端。适合字段映射规整、逻辑能用 SQL 表达的场景，改起来最快。

第三种是 3.0 才正式成型的Pipeline。不写 Java 代码，写一份 yaml，描述 source、sink、pipeline 三段，用bin/flink-cdc.sh提交。整库同步、表路由、加列同步这些原本要写代码的事，配置里两三行搞定。这也是这份资料里 MySQL 到 Doris 那一节用的方式。

我一般的判断标准是：只做搬运，选 Pipeline；搬运途中要洗数据，选 FlinkSQL；要接自定义 UDF、要和多条流 join，老老实实写 DataStream。

2.3 端到端链路里每个组件在干什么

把链路拆开看，一共四段。MySQL 侧负责产生 binlog，binlog_format必须是row，否则 Debezium 解析不出变更前后的完整行；Flink CDC Source 负责把 binlog event 反序列化成 Flink 的内部 RowData，全量阶段和增量阶段在同一个 Source 里完成切换，位点信息以状态形式存进 Checkpoint；中间的 Flink 算子做清洗和路由；最后 Doris Sink 把数据以 Stream Load 的方式批量灌进去。

Doris 这一端有个前提容易被忽略：目标表必须是唯一键模型（Unique Key），才能把op = u的更新和op = d的删除变成 upsert 和 delete。如果建成明细模型，同一条记录改十次你就得到十行，业务方一对账就发现数字翻倍。建表时UNIQUE KEY(id)加上DISTRIBUTED BY HASH(id) BUCKETS 1，小表这么写就够了，大表按数据量调 bucket 数。

还有一层是状态存储。Flink CDC 靠 Checkpoint 记住“读到哪个 binlog 文件的哪个位置”，所以必须配一个可靠的 Checkpoint 存储，HDFS 或对象存储都行。这份资料里用的是 HDFS 路径，恢复的时候从 Savepoint 起，能接着上次的位置读，不重不漏。

3. 环境与数据准备：Binlog、依赖与工程骨架

环境这一步最容易翻车，因为报错信息往往和真实原因隔了十万八千里。按顺序来，一步别跳。

3.1 建库建表与造数

先准备两个库，test做基础同步验证，test_route专门用来验证表路由。每个库里三张表，t1、t2结构一致，t3故意换个字段名，这样在路由和字段映射出问题时能第一时间发现。

-- 源库：基础同步验证 CREATE DATABASE IF NOT EXISTS test; USE test; CREATE TABLE t1 ( `id` VARCHAR(255) PRIMARY KEY, `name` VARCHAR(255) ); CREATE TABLE t2 ( `id` VARCHAR(255) PRIMARY KEY, `name` VARCHAR(255) ); -- 故意用不同字段，验证 schema 是否被正确识别 CREATE TABLE t3 ( `id` VARCHAR(255) PRIMARY KEY, `sex` VARCHAR(255) ); 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');

路由验证库同理，建一个test_route，表结构照抄，但t2里塞几条 ID 不同的数据（比如 1004、1005、1006）。这么做的目的是：同步跑起来后，你看 Doris 里test.t2和doris_test_route.t2的行数不一样，就能确认路由规则真的生效了，而不是所有表都被灌进了同一个库。

注意：主键用 VARCHAR 而不是自增 BIGINT，是为了模拟业务里常见的字符串主键场景。Flink CDC 全量阶段按主键切分 chunk，字符串主键也能切，但排序和切分效率不如整型，大表上建议还是用整型或雪花 ID。

3.2 开启 Binlog 的三个必改参数

MySQL 8 默认binlog_format是 ROW，但log-bin和server-id不一定开。改配置文件，加完重启。

# /etc/my.cnf server-id = 1 log-bin = mysql-bin binlog_format = row # 只对指定库产生 binlog，降低 binlog 体积 binlog-do-db = test binlog-do-db = test_route

server-id是给复制协议用的标识，CDC 客户端本质上就是一个伪装成 Slave 的消费者，没有它连不上。log-bin决定 binlog 文件前缀和是否启用。binlog_format = row是硬性要求，mixed 和 statement 都会让 Debezium 拿不到变更前后的完整行镜像。binlog-do-db是可选优化，不写就是全库记录。

改完重启，登进去确认一下：

mysql -uroot -p -e "show variables like 'binlog_format'; show variables like 'log_bin';"

两个值分别是ROW和ON才算过关。另外账号要有REPLICATION SLAVE、REPLICATION CLIENT、SELECT三个权限，缺一个都会在连接阶段被拒。

3.3 Maven 依赖与打包插件

工程用 Flink 1.18.0 打底，CDC connector 用 3.0.0，JDK 8。依赖分三组：Flink 运行时、Table API、CDC connector，外加 MySQL 驱动。

<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> <!-- 提交到集群时提供 Client 能力 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink-version}</version> </dependency> <!-- FlinkSQL 路线需要这一组，缺一个都会在 createTable 时报 NoMatchingTableFactory --> <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> <!-- 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，把所有依赖打进一个 fat jar。集群提交时只传这一个文件，省得在lib/目录里到处塞包。

<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>

这里有个血泪经验：scope千万别给 Flink 那几个包加provided。本地 IDE 跑的时候不会报错，打成 fat jar 上集群就抛ClassNotFoundException，因为集群自带的版本和你编译时用的可能不一致。用 assembly 全打进去，虽然 jar 大几十兆，但能省掉一整类排查。

提示：3.0 之后官方发行版逐步把包名迁到了org.apache.flink.cdc.*命名空间。这份资料的代码用的是com.ververica.cdc.*，如果你手上下的是 Apache 命名空间的包，import 要整体换掉，两套混着引会在运行时报方法签名不匹配，编译器还查不出来。

4. 核心链路拆解：DataStream、FlinkSQL 与 3.0 Pipeline

三种写法逐个过一遍，重点是搞清每个参数改了什么、不改会怎样。

4.1 DataStream 主程序：从环境到 Source

这段代码是整个链路的地基，Checkpoint 和 Source 两块的配置尤其不能省。

public class FlinkCDCDataStreamTest { public static void main(String[] args) throws Exception { // 1. 流处理环境，并行度先给 1，便于观察日志顺序 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2. Checkpoint 七件套，缺一项都可能影响断点续传 env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L); env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, Time.days(1L), Time.minutes(1L))); env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://node01:8020/flinkCDC"); // 写 HDFS 需要用户名，否则 Permission denied System.setProperty("HADOOP_USER_NAME", "flink"); // 3. 构造 MySQL CDC Source MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("node03") .port(3306) .databaseList("test") .tableList("test.t1") .username("root") .password("000000") .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); // 4. 接入流并打印 DataStreamSource<String> mysqlDS = env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), "MySQLSource"); mysqlDS.print(); env.execute(); } }

逐项说。enableCheckpointing(3000L, EXACTLY_ONCE)里的 3000 是间隔毫秒数，意思是每 3 秒做一次快照，生产上一般给 30 秒到 1 分钟，本地验证给 3 秒是为了快速看到效果。EXACTLY_ONCE保证恢复时不重复处理，代价是两阶段提交，Doris 侧要支持事务性写入，用 Stream Load 的话靠 label 去重。setCheckpointTimeout给 60 秒，大状态快照慢的时候不至于被误判超时。RETAIN_ON_CANCELLATION很关键，任务手动 cancel 时保留最后一次 Checkpoint，否则你点一下停止，位点就没了，再起来只能从头全量扫。

databaseList和tableList支持正则，写test表示整个库，写test.t1表示精确到表。想一次抓多张表可以写test.t1|test.t2，或者直接test\..*。JsonDebeziumDeserializationSchema把 SourceRecord 转成 JSON 字符串，输出里带before、after、op、ts_ms四个关键字段，调试阶段直接 print 最直观。

4.2 启动模式的五个取值，选错就是全量重扫

StartupOptions是 Flink CDC 里最容易被忽略、又最容易出事的一个参数。

// initial：首次启动先做全量快照，然后接着读最新 binlog（最常用） .startupOptions(StartupOptions.initial()) // earliest：不做快照，直接从 binlog 最早的位置读，要求 binlog 完整保留 // .startupOptions(StartupOptions.earliest()) // latest：不做快照，只读任务启动之后的变更，历史数据完全没有 // .startupOptions(StartupOptions.latest()) // specificOffset：从指定的 binlog 文件+位点开始 // .startupOptions(StartupOptions.specificOffset("mysql-bin.000003", 154)) // timestamp：从指定时间戳开始，会从头遍历 binlog 跳过更早的事件 // .startupOptions(StartupOptions.timestamp(1690000000000L))

initial是绝大多数场景的答案：第一次跑把存量数据全捞过来，捞完自动切增量。latest只在“目标端已经有存量数据、只想要增量”时用，用错了会发现 Doris 里空空如也。earliest最危险，它假设 binlog 里有完整历史，而 MySQL 的 binlog 默认只保留几天，过期就被 purge 了，这时候你读到的是残缺数据，而且不会报错。timestamp看着好用，实际会从头扫 binlog，binlog 体积大的时候启动要等很久。

4.3 Checkpoint 与 Savepoint：断点续传怎么验证

光配了没用，得真的验证一遍。这份资料的验证路径很标准，值得抄下来当 checklist。

# 1. 提交任务 bin/flink run -m node01:8081 -c com.example.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar # 2. 观察日志，应该先打出一批全量数据的 JSON，然后安静下来等 binlog # 3. 手动触发 Savepoint bin/flink savepoint <JobId> hdfs://node01:8020/flinkCDC/save # 4. WebUI 里 Cancel 任务 # 5. 去 MySQL 里对 t1 做几条 insert / update / delete # 6. 从 Savepoint 恢复 bin/flink run -s hdfs://node01:8020/flinkCDC/save/savepoint-xxxxx \ -c com.example.cdc.FlinkCDCDataStreamTest ./flink-cdc-test.jar

恢复后看日志：如果只打出了第 5 步那几条变更，说明位点恢复成功；如果又把 1001 到 1003 全量打了一遍，说明 Checkpoint 没生效或者状态后端不可写。第二种情况我遇到过，原因是 HDFS 路径的写权限没给对，HADOOP_USER_NAME设了个不存在的用户，Checkpoint 静默失败，任务照跑，日志里只有一行 warning。

注意：enableExternalizedCheckpoints不做设置的话，任务 cancel 时 Checkpoint 会被清掉。这个坑我在测试环境踩过一次，改完参数忘了重启任务，直接 cancel，位点没了，只能从头再扫一遍全量。

4.4 FlinkSQL 写法：一行 DDL 换来什么

SQL 路线把 Source 声明成一张表，后面接标准 SQL。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); // 声明 CDC Source 表，字段名要和 MySQL 对上 tableEnv.executeSql( "CREATE TABLE t1 (\n" + " id STRING PRIMARY KEY NOT ENFORCED,\n" + " name STRING\n" + ") WITH (\n" + " 'connector' = 'mysql-cdc',\n" + " 'hostname' = 'node03',\n" + " 'port' = '3306',\n" + " 'username' = 'root',\n" + " 'password' = '000000',\n" + " 'database-name' = 'test',\n" + " 'table-name' = 't1'\n" + ")"); // 查询并打印，结果表会持续输出 changelog tableEnv.sqlQuery("SELECT * FROM t1").execute().print();

PRIMARY KEY NOT ENFORCED是必须的，FlinkSQL 靠它判断这是 upsert 流而不是 append 流，下游做聚合时行为完全不一样。connector = 'mysql-cdc'是工厂标识，Flink 靠它找到对应的 TableFactory。和 DataStream 相比，SQL 版本少了 Checkpoint 配置，本地跑没问题，上集群一定要在外层补上，否则一样没法续传。

两种写法的取舍很实际：字段要做CASE WHEN、要做时间和字符串转换、要和维表 join，SQL 写着舒服；要接自定义反序列化、要把一条变更拆成多条下游消息、要动态决定写哪张目标表，只能上 DataStream。

4.5 3.0 Pipeline：一份 yaml 搞定整库同步

3.0 最大的变化在这儿。装好flink-cdc-3.0.0-bin之后，把flink-cdc-pipeline-connector-mysql-3.0.0.jar和flink-cdc-pipeline-connector-doris-3.0.0.jar扔进lib/目录，然后写配置。

source: type: mysql hostname: node01 port: 3306 username: root password: "123456" tables: test.\.* # 正则匹配整库，注意转义点号 server-id: 5400-5404 # 给个区间，多个任务不会撞车 server-time-zone: UTC+8 # 不写会出现时间差 8 小时 sink: type: doris fenodes: node01:7030 # 按实际 FE 的 http_port 填 username: root password: "000000" table.create.properties.light_schema_change: true table.create.properties.replication_num: 1 # 单 BE 测试环境必须给 1 pipeline: name: Sync MySQL Database to Doris parallelism: 1

source段的tables支持正则，test.\.*表示test库下所有表。server-id写成区间，Flink 会从中分配，多个同步任务并存时不至于互相踢下线。server-time-zone不配的后果是 datetime 字段整体偏移，这个坑很隐蔽，因为数据量看着对得上，只有对账时才发现时间戳全错。sink段的light_schema_change开启后，源端加列能自动同步到 Doris，不用手工改表；replication_num在单节点环境必须写 1，默认是 3，副本数不够会建表失败。

提交和原生 Flink 任务不一样，用的是 CDC 自带的脚本：

# Flink 侧先开 Checkpoint，3.0 的 Pipeline 依赖它做位点恢复 # conf/flink-conf.yaml 里加一行 # execution.checkpointing.interval: 5000 bin/flink-cdc.sh job/mysql-to-doris.yaml

跑起来之后去 Doris 里show tables，应该能看到和 MySQL 同名的库表。然后回 MySQL 对几张表做增删改，再刷新 Doris，数据应该同步过来。这一步验证通过，说明整条链路是通的。

路由变更的配置思路一样，只是加一段route：

route: - source-table: test_route\.t1 sink-table: doris_test_route.t1

source-table写库名加表名的正则，sink-table指定落到哪个库哪张表。整库改名的常见写法是sink-table: doris_route.<>，<>承接源表名，这样几十张表一条规则就迁完了，不用一张一张列。

5. 避坑与排查：那几条让我加班到凌晨的记录

下面五条都是我或者身边同事真实撞过的，按现象、原因、解决三段写，遇到了直接对号入座。

5.1 只跑全量不进增量，日志停在 snapshot 阶段

现象：任务启动后哗哗打出一批全量数据，然后就没动静了。去 MySQL 里改数据，Flink 日志一条新的都没有，但任务状态是 RUNNING。

原因：九成是 binlog 没配对。要么binlog_format还是MIXED，要么log-bin没开，要么账号缺REPLICATION SLAVE权限。还有一种情况是tableList写的是库名而不是表名，全量能扫到，增量订阅时匹配不上。

解决：先show variables like 'binlog_format'和show variables like 'log_bin'确认，再show grants for 'root'@'%'看权限。都正常的话，把日志级别调到 DEBUG，看有没有BinlogSplitReader相关的报错。我遇到过最离谱的一次是 MySQL 重启后 binlog 文件被轮转，客户端还拿着旧位点连，这时候清掉 Checkpoint 重跑最快。

5.2 从 Savepoint 恢复后数据重复或直接重扫全量

现象：任务 cancel 再恢复，Doris 里出现大量重复行，或者日志显示又从第一条数据开始扫。

原因：Checkpoint 没真正落盘。可能是状态后端配置和存储路径不匹配，可能是 HDFS 权限不足导致快照静默失败，也可能是 cancel 时忘了配RETAIN_ON_CANCELLATION，快照被清理掉了。

解决：先在 WebUI 的 Checkpoints 页签确认最近一次快照的 state 是 COMPLETED，再去 HDFS 上看目录里有没有chk-xxx文件夹。恢复时用-s参数指定具体路径，别用-s :latest，那个在 1.18 上偶尔会解析到已失效的目录。数据重复的话，Doris 侧如果是唯一键模型，重新灌一遍会自动去重，算是兜底。

5.3 源端加列之后 Doris 侧建表失败

现象：Pipeline 跑得好好的，开发在 MySQL 上ALTER TABLE ADD COLUMN，过一会儿 Doris 侧报 schema change 相关的错，同步中断。

原因：Doris 的 light schema change 默认关闭，加列需要走重量级的 schema change，耗时长而且和写入冲突。另外如果目标表不是唯一键模型，加列行为也不一样。

解决：配置文件里必须显式写table.create.properties.light_schema_change: true，这个参数只对新建表生效，已经建好的表要手工改。加列操作建议放在低峰期，改完观察 Doris 的SHOW ALTER TABLE COLUMN状态直到 FINISHED。

5.4 多个同步任务互相踢，日志报 server-id 冲突

现象：第二个同步任务起来之后，第一个任务突然断开重连，两个任务反复抢。

原因：每个 CDC 客户端在 MySQL 眼里都是一个 Slave，server-id就是身份标识。两个任务用了同一个 ID，MySQL 会认为后连的把前一个顶掉。

解决：配置里用区间，比如server-id: 5400-5404，Flink 会从中分配不同的值。任务多了就把区间放大，别偷懒写死一个数字。这个问题在单任务测试时永远不会暴露，一上生产就炸。

5.5 时间字段整体差 8 小时

现象：数据都同步过来了，行数也对，但 Doris 里的create_time比 MySQL 里晚了 8 小时。

原因：MySQL 的datetime类型本身不带时区，binlog 里存的是服务器本地时间，客户端默认按 UTC 解释，一进一出就差 8 小时。这个错误不报异常，最容易被忽略。

解决：配置里加server-time-zone: UTC+8。已经同步过去的数据改配置也不会自动修正，要重新灌一遍。上线前我一般会挑一条带时间戳的记录，在两边各查一次做比对，比看日志靠谱。

6. 二次开发改造：从能跑到敢放进生产

链路跑通只是第一步。真正上生产，有几个地方我基本都会动。

换掉 JSON 反序列化。JsonDebeziumDeserializationSchema调试好用，生产环境序列化开销不小，而且下游还得再解析一遍。常见的做法是自己实现DebeziumDeserializationSchema<T>，直接输出自定义 POJO 或者 RowData，把before、after、op打平成业务需要的结构。

public class FlatRowDeserializer implements DebeziumDeserializationSchema<RowData> { @Override public void deserialize(SourceRecord record, Collector<RowData> out) { // 从 record 里取出 Struct，按 op 类型分别处理 Struct value = (Struct) record.value(); String op = value.getString("op"); // c=insert, u=update, d=delete, r=snapshot Struct after = value.getStruct("after"); // 只保留业务需要的字段，按目标端顺序组装成 RowData out.collect(RowData.create(...)); } @Override public TypeInformation<RowData> getProducedType() { return ...; // 和下游算子声明的类型必须严格一致 } }

这里的重点是op的四个取值：c是插入，u是更新，d是删除，r是全量快照阶段读出来的行。删除事件的after是 null，字段要从before里取，这点如果写漏了，删除操作会直接 NPE。

加一层打宽。原始 binlog 是一条一条变更，业务侧要的是宽表。常见做法是拿 CDC 流去 join 一张维表，用 FlinkSQL 的 temporal join，维表用 JDBC connector 或者 HBase 缓存。改造完之后 Pipeline 就用不了了，得回到 DataStream 或 SQL 路线。

监控指标。Pipeline 模式下能看到的只有 Flink 的吞吐和延迟，业务侧感知不到。我一般会在 Sink 前后各挂一个Counter，记录处理条数和失败条数，暴露到 Prometheus。验证方法是：先看总条数是否和源端变更次数对得上，再看端到端延迟的 P99 有没有突然抬升。

改完别急着发生产，先在测试库跑一轮完整的“全量 + 增删改 + 加列 + 恢复”组合拳。我踩过最贵的一次坑是：只测了插入，没测删除，上线第一周业务方发现删掉的订单在报表里还活着，排查了半天才发现是op = d分支写错了字段来源。从那以后我每次改反序列化逻辑，都会强制走一遍包含删除和加列的回归清单，哪怕只改了一行。希望帮到你。

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