订单状态半夜被人改错，查日志发现是一次后台误操作，这类事早些年我靠定时任务扫表兜底，轮询间隔压到秒级之后数据库负载直接飙起来了。后来换成Flink CDC 监测 MySQL 数据库，事情才算理顺：数据变更不再靠猜，数据库把 binlog 里的每一条变更主动推过来。围绕这个场景，自定义反序列化是绕不开的一环，它决定了你下游拿到的是原始 SourceRecord 还是干净的业务消息；而落地上，DataStream API 和 Flink SQL 是两条完全不同的路，各有各的甜区和边界。这篇就把我这套跑在生产里的做法摊开讲：SourceRecord 长什么样、反序列化器怎么写、两条路怎么搭、哪些坑必须提前踩过。适合已经写过 Flink 作业、准备把 CDC 接进链路的中级读者，也适合刚接触 CDC、想知道"到底选 A 还是选 B"的同学。
1. 先把要做的事拆开：监测 MySQL 变更到底在做什么
很多人一上手就问"Flink CDC 怎么配"，其实先得把需求翻译成技术动作：数据库里某几张表的行数据发生变化时，我要在几秒内拿到"哪张表、什么操作、变更前是什么、变更后是什么"，并且这个链路要能扛住断点重启、不丢不重。拆开就是三件事：捕获变更事件、把事件翻译成业务能用的结构、把结果可靠地送出去。前两件事由 MySQL CDC Source 和反序列化器负责，第三件事由下游 Sink 和 checkpoint 机制兜底，三者缺一不可。
1.1 为什么不再用轮询扫表
轮询的本质是"主动去问数据库变没变"，代价是它必须全表或索引扫描，还得靠一个 update_time 字段做增量判断。问题在于：硬删除的数据永远扫不到；同一秒内的两次更新会被合并成一条；update_time 没有索引时扫描代价极大；为了降低延迟必须缩短间隔，数据库的连接数和 IOPS 就跟着涨。binlog 的思路完全反过来——数据库本来就要写 binlog（主从复制、数据恢复都依赖它），我们只是多开一个消费者去读这个日志流，对数据库几乎没有额外查询压力，删除、更新前后值、事务顺序全都天然带在事件里。
注意：MySQL 的 binlog 必须开成 ROW 模式，这是 CDC 能拿到行级前后值的前提。STATEMENT 模式下 binlog 里只有 SQL 语句文本，反序列化时根本拿不到结构化字段。
1.2 技术选型的取舍逻辑
同类方案里比较常见的有三种：自己解析 binlog（比如用开源的 binlog 客户端库）、独立的 CDC 中间件、以及 Flink CDC 直接内嵌到 Flink 作业里。前两种都要多维护一个中间组件，链路变成"数据库 → 中间件 → 消息队列 → 计算引擎"，节点越多，运维成本和排查难度越高。Flink CDC 把读取 binlog 的逻辑做成了 Flink 的 Source，作业一启动就直连数据库，变更事件直接进入算子链，省掉了中间的搬运环节。
更关键的一点是增量快照能力。老版本的 CDC 在全量阶段要加全局读锁，业务线上库根本不敢这么玩。现在主流的做法是先记录一个 binlog 位点，然后按主键区间切片并行拉取历史数据，全量阶段结束后无缝切换到 binlog 增量阶段，全程只在极短时间内持有一把表级 schema 锁。这个机制让"历史数据 + 实时变更"可以在同一个作业里一次搞定，不需要额外写一套全量导数据的脚本。
1.3 版本矩阵和依赖，别在这一步翻车
Flink CDC 的版本和 Flink 版本是强绑定的，装错组合会出现各种NoSuchMethodError，而且报错信息往往指向 Flink 内部类，极易误判。我一般是先定 Flink 版本，再去查对应 CDC 版本的兼容区间，不去赌"高版本应该兼容"。下面这套是我实际跑得比较稳的组合，可以当作基准。
| 组件 | 建议版本 | 说明 |
|---|---|---|
| Flink | 1.17.x | 1.14 之后 Source API 统一，兼容性最好 |
| flink-connector-mysql-cdc | 2.4.x | DataStream 方式使用，非 shaded 包 |
| flink-sql-connector-mysql-cdc | 2.4.x | SQL 方式使用，shaded 包，直接丢 lib |
| MySQL | 5.7 / 8.0 | 8.0 需注意认证插件与时区参数 |
| Debezium 内核 | 随 CDC 包内置 | 不要单独引 debezium-core，冲突概率极高 |
依赖上有两个死规矩：第一，flink-connector-mysql-cdc和flink-sql-connector-mysql-cdc不要同时放进 lib 目录，前者没做 shade，会和后者里的类打架；第二，自己项目里如果引了 jackson、guava、kafka-clients，一定要用 provided 或者与 Flink 内置版本对齐，我遇到过最隐蔽的一次是 guava 版本差了两个小版本，作业提交时报NoSuchMethodError: com.google.common.base.Preconditions.checkArgument，排查了大半天才发现是业务依赖带进来的。
2. 自定义反序列化：从 SourceRecord 到业务消息
反序列化器是这条链路里唯一一处"你能完全掌控数据形态"的地方。Flink CDC 给的默认实现能用，但基本只够你调试。真正上线，我见到的需求百分之百都会要求自定义：要么字段名要统一成下划线风格、要么时间字段要转成标准字符串、要么要把库名表名拼进消息体方便下游分流。
2.1 Debezium 的 SourceRecord 到底长什么样
Flink CDC 底层复用了 Debezium 的模型，SourceRecord 是个"键值对"，key 是主键信息，value 是变更详情，两者都是 Avro 风格的 Struct。真正有价值的信息全在 value 里，它的字段结构是这样的：
| 字段 | 类型 | 含义 |
|---|---|---|
| before | Struct | 变更前的行数据，insert 时为 null |
| after | Struct | 变更后的行数据，delete 时为 null |
| source | Struct | 元信息：库名、表名、binlog 位点、事件时间戳等 |
| op | String | 操作类型：r 快照读、c 新增、u 更新、d 删除 |
| ts_ms | Long | 事件产生时间（毫秒） |
source 这个 Struct 里的东西特别有用：db、table给你库表名，ts_ms给你数据库侧的事件时间，file和pos给你 binlog 位点。我习惯把 ts_ms 从 source 里取而不是从顶层取，因为不同 Debezium 版本顶层 ts_ms 的字段命名有过调整，source 里的相对稳定。
op的取值是反序列化的分叉点，一定不要漏判。慢着，这里还有个更隐蔽的坑：如果你在 MySqlSource 上开了includeSchemaChanges(true)（默认就是 true），DDL 变更也会以 SourceRecord 的形式流进反序列化器，而这种记录的 value 结构里根本没有op字段，直接强转取值会抛Field op not found in struct。我第一次遇到时百思不得其解，因为正常数据都好好的，只有跑 DDL 的时候炸。
2.2 完整的自定义反序列化器实现
我的写法是先判断 valueSchema 里有没有op，没有就直接忽略，然后按 op 分支取 before/after，最后统一组装一条带元信息的 JSON 消息。这样下游无论做分流还是做审计，都不需要再去关联原表。
public class CdcJsonDeserializer implements DebeziumDeserializationSchema<String> { private static final long serialVersionUID = 1L; private static final ObjectMapper MAPPER = new ObjectMapper(); @Override public void deserialize(SourceRecord record, Collector<String> out) throws Exception { Struct value = (Struct) record.value(); if (value == null) { return; } // DDL / schema change 事件没有 op 字段，直接跳过 Schema valueSchema = value.schema(); if (valueSchema.field("op") == null) { return; } String op = value.getString("op"); Struct source = value.getStruct("source"); Struct before = value.getStruct("before"); Struct after = value.getStruct("after"); Map<String, Object> msg = new LinkedHashMap<>(); msg.put("db", source.getString("db")); msg.put("table", source.getString("table")); msg.put("op", translateOp(op)); msg.put("op_ts", source.getInt64("ts_ms")); msg.put("binlog_file", source.getString("file")); msg.put("binlog_pos", source.getInt64("pos")); msg.put("before", structToMap(before)); msg.put("after", structToMap(after)); out.collect(MAPPER.writeValueAsString(msg)); } private String translateOp(String op) { switch (op) { case "r": return "SNAPSHOT"; case "c": return "INSERT"; case "u": return "UPDATE"; case "d": return "DELETE"; default: return op; } } private Map<String, Object> structToMap(Struct struct) { if (struct == null) { return null; } Map<String, Object> map = new LinkedHashMap<>(); for (Field field : struct.schema().fields()) { Object raw = struct.get(field); map.put(field.name(), normalize(raw, field.schema())); } return map; } private Object normalize(Object raw, Schema schema) { if (raw == null) { return null; } if (raw instanceof ByteBuffer) { byte[] bytes = new byte[((ByteBuffer) raw).remaining()]; ((ByteBuffer) raw).get(bytes); return Base64.getEncoder().encodeToString(bytes); } if (raw instanceof Struct) { return structToMap((Struct) raw); } if (raw instanceof Array) { List<Object> list = new ArrayList<>(); for (Object item : (Object[]) raw) { list.add(normalize(item, null)); } return list; } return raw; } @Override public TypeInformation<String> getProducedType() { return TypeInformation.of(String.class); } }几个关键点值得单独说。
第一，getProducedType 别乱写。这个方法决定了 Source 输出给下游的 TypeInformation。如果你写Types.STRING，输出必须是 String；如果你自己定义了 POJO，就必须返回对应的TypeInformation.of(YourPojo.class)，且 POJO 要满足 Flink 的序列化约定（无参构造、public 字段或标准 getter/setter）。我曾经贪图省事返回了泛型擦除后的 TypeInformation，作业跑起来反序列化正确，但一开 checkpoint 就报序列化异常，原因是下游算子拿不到真实类型信息。
第二，structToMap 一定要做类型归一化。Debezium 的原始 Struct 里，二进制字段是ByteBuffer，嵌套字段是Struct，数组是Array，直接塞进 Jackson 序列化会得到一堆奇怪的内部结构甚至抛异常。上面这段 normalize 是我调整过很多轮的版本，覆盖了实际遇到过的绝大部分字段类型。
第三，时间字段的处理要有统一策略。MySQL 的DATETIME、TIMESTAMP、DATE在 Debezium 里输出成什么样，取决于time.precision.mode这个参数。默认自适应模式下，TIMESTAMP会输出成长整型（微秒），DATETIME可能输出成java.time.LocalDateTime，DATE输出成java.time.LocalDate。这意味着你的反序列化器如果只做简单的toString()，时间字段会变成一串数字。我的做法是在 MySqlSource 上显式设置debezium.time.precision.mode=connect，让 Debezium 按 JDBC 标准类型输出，然后再自己按业务需要格式化成字符串。
Properties props = new Properties(); props.setProperty("time.precision.mode", "connect"); props.setProperty("decimal.handling.mode", "string"); props.setProperty("bigint.unsigned.handling.mode", "long");decimal.handling.mode=string这条对金额字段特别重要。DECIMAL 在 Debezium 里默认是BigDecimal或者带 scale 的字节数组，如果不转成 string，一旦字段精度变化，下游解析就会错位。转成 string 之后，金额的精度和尾零都原样保留，做对账类业务时这个细节能救命。
2.3 为什么不直接用 SimpleStringSchema 或者 JsonDebeziumDeserializationSchema
SimpleStringSchema理论上能编译通过，但它在运行时拿到的对象强转 String 会报 ClassCastException，因为它根本不认识 SourceRecord。JsonDebeziumDeserializationSchema是官方给的默认 JSON 实现，输出的是 Debezium 原生结构，字段名还是before/after/op/source那一套，而且默认会把before和after里所有字段都带上，包括一堆你不需要的元数据。
自定义的价值不只是改字段名，更重要的是在这里做一次数据清洗。比如敏感字段脱敏、金额单位换算、枚举值翻译、库表名统一加前缀。这些动作放在反序列化器里做，比在后面加一个 Map 算子更省一次序列化开销，也避免了"反序列化出来的字符串还要再解析一遍"的浪费。当然，如果你只需要最基础的字段，官方实现也是够用的，别为了自定义而自定义。
提示：反序列化器里不要做任何阻塞型的外部调用，比如查 Redis 补全信息。这是流处理的热路径，每个 binlog 事件都会走一次，一旦外部调用超时，整个 Source 就会被拖住，checkpoint 也会跟着超时。
3. DataStream API 落地：能精细控制的那条路
DataStream 方式是我在复杂场景下的首选，因为反序列化、并行度、算子链、旁路输出全都能自己掌控。搭起来的骨架并不复杂，难的是参数取舍。
3.1 环境初始化与 Source 构建
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(4); env.enableCheckpointing(10_000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5_000L); env.getCheckpointConfig().setCheckpointTimeout(60_000L); env.getCheckpointConfig().setExternalizedCheckpointCleanup( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); env.getCheckpointConfig().setCheckpointStorage("file:///data/flink/checkpoints"); Properties dbzProps = new Properties(); dbzProps.setProperty("decimal.handling.mode", "string"); dbzProps.setProperty("time.precision.mode", "connect"); dbzProps.setProperty("bigint.unsigned.handling.mode", "long"); MySqlSource<String> source = MySqlSource.<String>builder() .hostname("10.0.0.21") .port(3306) .databaseList("demo_db") .tableList("demo_db.t_order", "demo_db.t_order_item") .username("cdc_reader") .password("******") .serverId("5401-5404") .serverTimeZone("Asia/Shanghai") .startupOptions(StartupOptions.initial()) .deserializer(new CdcJsonDeserializer()) .debeziumProperties(dbzProps) .includeSchemaChanges(false) .build(); DataStreamSource<String> stream = env.fromSource( source, WatermarkStrategy.noWatermarks(), "mysql-cdc-source");serverId这里填的是一个区间，它决定了 Source 在读取 binlog 时向数据库注册的从库 ID。这个 ID 在数据库实例内必须唯一，如果你有多个 CDC 作业连同一个实例，区间千万别重叠，否则会出现经典的A slave with the same server_uuid/server_id as this slave has connected to the master报错。我一般约定：一个实例上给 CDC 分配 5400 到 5500 这一段，每个作业切三到五个，作业配置里写死区间，避免抢号。
includeSchemaChanges(false)这个参数我通常直接关掉，因为绝大多数场景不需要感知 DDL 事件，关掉之后反序列化器里那层 schema change 判断就成了双保险。如果你确实要做表结构变更的联动，比如字段新增后自动扩目标表的列，那就打开它，然后在反序列化器里单独处理historyRecord分支。
3.2 启动模式怎么选，一次说清
启动模式直接决定了作业第一次启动时从哪读数据，选错了要么重复消费，要么丢历史数据。
| 启动模式 | 行为 | 适用场景 |
|---|---|---|
| initial | 先全量快照，再从位点读增量 | 首次上线，需要完整数据 |
| earliest | 不做快照，从最早的 binlog 位点读 | 只关心变更、且 binlog 保留够久 |
| latest | 只读启动之后的增量 | 调试、临时排查 |
| specificOffset | 从指定 binlog 文件与位点读 | 故障恢复，精确补数 |
| timestamp | 从指定时间点开始读 | 补某段时间的变更 |
我踩过最深的一个坑是用了StartupOptions.latest()做压测，结果发现数据量对不上。原因是 latest 只读启动之后的变更，作业启动过程中那零点几秒的变更被跳过了。生产上除非明确只要"今后发生的事"，否则老老实实用 initial，全量补一遍不会很慢，还省了很多解释不清的数据差异。
还有一个细节：作业重启时会优先从 savepoint 或 checkpoint 里恢复位点，启动模式只在第一次启动或没有状态恢复时生效。所以改了启动模式不会让运行中的作业突然重读历史，必须清掉状态重新提交。这一点很多人会误解，觉得改个参数重启就重新导数据了，结果发现什么都没有变化。
3.3 下游写出与 checkpoint 配置的关系
反序列化之后的字符串流，下游接什么都行：写 Kafka、写数据湖、走自定义 Sink 做审计入库。这里我最想强调的只有一句：checkpoint 必须开，而且必须能成功。
Flink CDC 的增量快照机制依赖 checkpoint 来推进阶段切换。全量阶段每读一批 split 就做一次 checkpoint 记录进度，全部读完之后，只有在下一次 checkpoint 成功时，才会把 binlog 位点固化下来并切换到增量阶段。如果 checkpoint 一直失败，你会看到一个非常迷惑的现象：全量数据已经打出来了，但之后的更新一条都没有，作业也不报错，就静静地卡在那里。我第一次遇到时对着日志找了两个小时，最后才意识到是 checkpoint 目录权限问题。
env.getCheckpointConfig().setCheckpointTimeout(180_000L); env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3); env.setStateBackend(new HashMapStateBackend());超时时间也要给够。全量阶段读取的 split 可能会很大，如果 checkpoint 超时设得太短（比如默认的十分钟里还要做 split 快照），会频繁失败。我的经验值是：单表数据量在千万级以下，checkpoint 间隔 10 到 30 秒、超时 3 分钟足够；亿级大表建议把scan.incremental.snapshot.chunk.size调大（默认 8096，可以给到 16384 或 32768），减少 split 数量，加快全量阶段结束。
4. Flink SQL 落地：写几句 DDL 就能跑
如果你的需求是"把变更同步到另一张表"或者"做一次简单的过滤聚合后入库"，Flink SQL 的效率是 DataStream 的几倍。不用编译打包，不用管反序列化器，一个 DDL 加一个 INSERT 就完事。
4.1 建表语句与关键参数
CREATE TABLE mysql_order_source ( id BIGINT, order_no STRING, user_id BIGINT, amount DECIMAL(12, 2), status INT, update_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = '10.0.0.21', 'port' = '3306', 'username' = 'cdc_reader', 'password' = '******', 'database-name' = 'demo_db', 'table-name' = 't_order', 'server-id' = '5401-5404', 'server-time-zone' = 'Asia/Shanghai', 'scan.incremental.snapshot.enabled' = 'true', 'scan.incremental.snapshot.chunk.size' = '16384', 'scan.snapshot.fetch.size' = '1024', 'connect.timeout' = '30s', 'connect.max-retries' = '5' );三个参数必须盯紧。table-name支持正则，多表合并就写'demo_db\\.t_order.*'，注意在 SQL 字符串里反斜杠要转义。server-time-zone一定要和数据库实际时区一致，这是时间字段差八小时的元凶。PRIMARY KEY必须声明且标NOT ENFORCED，SQL 侧要靠它推导 changelog 类型，没有主键的表只能产生 INSERT 流，更新和删除会被降级处理。
建完之后，SQL 侧拿到的是一个标准 changelog 流：新增是 +I，更新是 -U 和 +U，删除是 -D。下游如果是kafkaconnector 配'format'='json'，你会丢掉 -U 这条消息，因为纯 JSON 格式表达不了撤回流。要保留完整语义，format 用debezium-json或changelog-json，或者下游用upsert-kafka按主键做幂等写。
4.2 SQL 侧想自定义输出，能走多远
这是被问得最多的问题：SQL 方式能不能用自己的反序列化器？答案是不能直接换，但有三条变通路子，按推荐顺序排。
第一条，用元数据列。CDC connector 支持把库名、表名、数据库侧事件时间作为虚拟列暴露出来，直接在主表 DDL 里声明即可。
CREATE TABLE mysql_order_source ( id BIGINT, order_no STRING, amount DECIMAL(12, 2), db_name STRING METADATA FROM 'database_name' VIRTUAL, tbl_name STRING METADATA FROM 'table_name' VIRTUAL, op_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL, PRIMARY KEY (id) NOT ENFORCED ) WITH ( ... );这样下游拿到的每一行都自带来源信息和事件时间，效果上已经覆盖了大部分"自定义输出"的诉求——尤其是需要按来源分流的场景。
第二条，用SQL 内置函数加工。字段重命名、格式化、脱敏、拼接都能在INSERT INTO ... SELECT里做，比如CONCAT(db_name, '.', tbl_name) AS source_table、DATE_FORMAT(update_time, 'yyyy-MM-dd HH:mm:ss')，复杂一点还能写 UDF。
第三条，混合链路。真正需要完全控制消息体的，就用 DataStream 读、自定义反序列化、再toChangelogStream()转成 Table 继续写 SQL。这条路兼顾了两边的优势，代价是多写一层胶水代码，并且要在同一个作业里管好两种 API 的 checkpoint 语义。
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); Table table = tEnv.fromChangelogStream(stream.map(...)); tEnv.createTemporaryView("t_order_rt", table); tEnv.executeSql("INSERT INTO sink_table SELECT ... FROM t_order_rt");4.3 两种方式怎么选，我的实际判断标准
| 维度 | DataStream API | Flink SQL |
|---|---|---|
| 反序列化控制 | 完全自定义 | 仅元数据列 + 内置函数 |
| 开发效率 | 需编译打包，迭代慢 | 改 SQL 即可，几分钟上线 |
| 复杂状态计算 | 灵活，可用 ProcessFunction | 受限于 SQL 算子 |
| 多表合并 | 手写 union，代码量大 | 表名正则一行搞定 |
| 调试难度 | 可打断点、可本地跑 | 依赖日志与 Web UI |
| 依赖风险 | 易与其他 jar 冲突 | shaded 包独立，风险低 |
我的判断很简单：链路里只要出现"反序列化要改字段语义"或者"要做跨事件的复杂判断"，直接上 DataStream；其余情况，尤其是需求方还在反复改字段的场景，SQL 方式能省下大量重启打包的时间。很多团队的实践是：先用 SQL 快速把链路跑通验证可行性，稳定之后再按需下沉到 DataStream 做精细化，这个顺序比反过来要省事。
5. 排查实录：那些真的会卡住你的问题
CDC 这类作业的特点是"要么很稳，要么一动不动"，报错信息有时候还特别具有误导性。下面这几类是我在实际环境里反复遇到的。
5.1 启动阶段最常撞的几堵墙
| 现象 | 根因 | 处理方式 |
|---|---|---|
| 连接直接拒绝 | 账号权限不足 | 补 SELECT、RELOAD、SHOW DATABASES、REPLICATION SLAVE、REPLICATION CLIENT 权限 |
| 报 binlog 格式错误 | 未开 ROW 模式 | 改binlog_format=ROW并重启实例 |
| 报 server id 冲突 | 多个作业用了同一区间 | 每个作业分配独立区间 |
| 全量阶段卡住不动 | checkpoint 未开启或持续失败 | 打开 checkpoint 并检查目录权限 |
| 报 GTID 相关错误 | GTID 模式与配置不匹配 | 单机与主从环境要分别确认配置 |
权限这一条特别容易被忽略。很多人只给了 SELECT，觉得读取数据够了，但 CDC 需要注册成从库去拉 binlog，REPLICATION SLAVE 和 REPLICATION CLIENT 是硬性要求，RELOAD 权限用于全量阶段拿一致性快照。少了任何一个，报错都指向很底层的位置。
还有一个隐蔽的坑：大小写敏感。数据库在 Linux 上表名默认区分大小写，而在本地开发机上往往不区分。本地跑得好好的作业，上线之后tableList写的表名大小写和实际不一致，作业不报错，就是收不到数据。我现在的习惯是建完表先SHOW TABLES把结果复制到配置里，不手打。
5.2 数据对不上：时间、精度与重复
时间差八小时，是 CDC 里排名第一的高频问题。它有三个来源，要逐个排查：数据库服务端时区设置、Source 的server-time-zone参数、Flink 运行环境的 JVM 时区。最常见的组合错误是数据库在东八区，但server-time-zone没配（默认 UTC），于是 Debezium 在解析时间字段时按 UTC 理解，输出就整体偏了八小时。我现在的标准做法是三处对齐：数据库配置、Source 参数、JVM 启动参数统一成Asia/Shanghai。
金额精度错位，通常和decimal.handling.mode有关。默认模式下 DECIMAL 会以字节数组形式输出，某些精度下序列化后是个看不懂的结构。设成string之后，反序列化器拿到的就是"12.30"这样的字符串，精度和尾零都在，下游建表时用 DECIMAL 或 VARCHAR 都能对上。
至于重复数据，多半不是 CDC 的问题，而是 Sink 的问题。CDC 本身靠 checkpoint 保证了 at-least-once 语义，故障恢复时可能重放少量事件。要拿到端到端的幂等，下游必须按主键做 upsert，或者写入时带唯一约束。这一点上upsert-kafka配合主键是最省心的组合。
5.3 并行度、吞吐与拆表策略
MySQL CDC Source 在增量阶段只能单并行度读 binlog，因为 binlog 本身是串行的。但全量阶段是可以并行的，并行度由 split 数量决定。这意味着：如果你的表都很小，把 Source 并行度设成 8 是浪费；如果是千万级大表，并行度 8 甚至 16 能显著缩短全量时间。
scan.incremental.snapshot.chunk.size这个参数直接决定 split 的粒度。默认 8096 行一个 chunk，大表会切出成千上万个 split，管理开销反而增大。我在处理一张六千万行的表时，把它调到 32768，split 数量从七千多降到两千以内，全量时间从四十多分钟缩到十二分钟。这不是线性关系，得看实际机器和网络，建议先用默认值跑一遍看监控，再决定要不要调。
还有一个只有实际跑过才知道的细节：多表合并时尽量让表结构相近。SQL 侧用正则匹配多张表，如果某些表缺字段，Flink 会按位置对齐，出现取值错位但不报错的情况。我现在要求所有参与合并的表必须显式列出相同字段集合，缺的字段在源库侧补齐或用CAST兜底，宁可多写几行 DDL，也不要赌运气。
6. 我个人的几条实操经验
跑了几十个 CDC 作业下来，有几条经验是文档里不太会写、但确实能省时间的。第一条是先小表跑通再上大表，用一张几万行的小表验证全链路（反序列化、下游写入、checkpoint 恢复），确认没问题再往百万千万级的表上推进。第二条是给作业留一个 Dry Run 模式，把反序列化后的消息只打进日志不落库，上线前跑十分钟看看数据形态对不对，比事后回滚便宜太多。
第三条是关于反序列化器的可测试性。这个东西的逻辑分支其实不少（insert、update、delete、快照、schema change），但很多人都是上线后用真实数据验证。我的做法是把deserialize里的核心逻辑抽成一个纯函数，输入是 value 的 Struct，输出是 Map，然后拿几条构造好的 Debezium 结构在单元测试里跑一遍。别小看这几行测试，我至少两次靠它在提交作业之前发现了 delete 分支取错 before 字段的问题。
最后分享一个排查小技巧：当你不确定反序列化器到底收到了什么，先临时把getProducedType改成Types.STRING，在 deserialize 里只输出record.value().toString()，把原始结构打出来看一眼。Debezium 的 Struct toString 会把字段名和值完整打印出来，这一下就能看清 op 是什么、before 和 after 哪个是空的、时间字段是什么类型。看完再改回正式逻辑，比在代码里盲猜快十倍。
