在实时数仓和业务库异构同步这两个场景里做了几年绕不开的一个组件就是 Flink CDC。它做的事情说起来朴素——把数据库的变更增、删、改变成一条条流式记录源源不断地送进计算引擎但真要在生产跑起来里面涉及的全量快照算法、位点与状态的绑定、并发分片、DDL 演进这些细节足以把人按在地上摩擦好几轮。我把这几年踩过的坑、调过的参数、翻过的源码整理出来讲清楚Flink CDC到底是怎么把一条 MySQL 的 update 变成 Flink 里的一条流记录以及一条从业务库到下游的同步链路该怎么落地。这篇内容适合三类人刚接触实时同步、想知道全量加增量到底怎么衔接的入门同学已经在用 CDC 但被延迟、重复、断点续传问题困扰的中间层工程师以及需要给团队做技术选型和容量评估的负责人。下面的内容会从设计思路讲起再到原理拆解、代码实操、问题排查代码可以直接抄走改改就用。1. 先想清楚为什么要在流式管道里捕获变更1.1 从定时抽数到日志级捕获中间隔了多少坑早些年做数据同步主流做法是定时任务跑批量 SQL比如每小时拉一次全表或者拉update_time 上次时间的增量。这套方案在小数据量下能跑但只要数据量上去、业务表结构稍微复杂一点问题就集中爆发。最典型的是物理删除——delete from t where id 100你按更新时间拉增量这条删除是拉不到的下游表里就永远留着一条幽灵数据。其次是更新覆盖同一行在一小时内被改了三次批量抽数只能看到最后一次的终态中间过程全丢了做实时风控或者审计根本没法用。再往后演进是时间戳 触发器的妥协方案在业务表上挂 trigger 写一张变更日志表。这个方案对业务库的侵入性太强DBA 一般不会同意而且 trigger 在高并发写入下本身就是性能瓶颈。真正让这件事变得优雅的是数据库自己写的 binlog——MySQL 的 row 模式 binlog 记录了每一行数据的完整前后镜像它本来就是数据库为了主从复制和崩溃恢复而存在的我们只是顺带读一下对业务链路几乎零侵入。这里有一个认知很关键CDC 不是我们去查数据变化而是去听数据库自己在说什么。前者是轮询后者是订阅两者的延迟量级和资源模型完全不是一个东西。Flink CDC 就是站在这个基础上把 binlog 的读取、解析、快照衔接、断点续传这一整套脏活累活封装成了 Flink 的 Source 算子。它的价值不在于能读 binlog——Canal、Maxwell 早就能读了——而在于它把变更捕获做成了一个有状态、可容错、可并行的流处理源。1.2 三种主流方案的横向对比为什么最后选了 Flink CDC做选型的时候我一般会把候选方案摊开按几个维度打分运维复杂度、端到端延迟、全量能力、断点续传可靠性、是否依赖中间件。维度定时批量抽数Canal/Maxwell 消息队列Flink CDC端到端延迟分钟到小时级秒级秒级甚至亚秒级删除捕获做不到除非软删支持支持全量初始化单独脚本需要自己拼内置与增量自动衔接断点续传靠时间戳字段依赖 MQ 位点依赖 Flink Checkpoint额外组件调度系统必须部署采集端 MQ只需 Flink 集群对源库压力定时尖峰持续低压力全量阶段可控增量低压力运维成本低但功能弱中多一套组件中但链路短表格里最后一列是我最终倾向 Flink CDC 的核心理由链路短。Canal 方案里采集端、消息队列、消费端是三套独立组件任何一套挂了都要单独排查位点对不上还要人工介入。Flink CDC 把采集和计算合并在一个作业里位点信息随 Checkpoint 一起持久化作业从 savepoint 恢复时binlog 位点自动跟着恢复不存在消息队列里的数据算不算消费过这种边界问题。代价当然也有。第一CDC Source 的并发模型受源库限制MySQL 单表在早期版本里只能单并发读快照大表初始化会很慢第二作业的重启成本比无状态任务高状态越大恢复越慢第三所有的容错压力都压在 Flink 的 Checkpoint 机制上一旦 Checkpoint 持续失败位点不推进作业就等于卡死。这三点在后面的章节里会逐一展开。1.3 版本演进从Debezium 套壳到独立框架理解 Flink CDC 的架构最好先理清它的版本脉络因为网上大量教程还停留在 1.x 时代照着抄很容易踩到架构层面的坑。1.x 版本的实现非常直接把 Debezium 的EmbeddedEngine塞进 Flink 的RichSourceFunction里全量阶段就是一条SELECT * FROM t增量阶段切到 binlog 流。这个版本最大的问题是全量阶段要锁表默认snapshot.modeinitial会加全局读锁来保证快照一致性一张几千万行的表锁上十几分钟业务直接告警。另外它是单并发的Source 并行度设成 4 也只有 1 个 subtask 在干活。2.x 是真正意义上的分水岭引入了Incremental Snapshot Framework增量快照框架。核心思路是把表按主键切成若干 chunk多个 subtask 并行读不同的 chunk全程不加锁同时后台并行消费 binlog最后靠水位线算法把两边数据合并去重。这一步直接把大表初始化的耗时从小时级压到了分钟级。3.x 又往前走了一步做了架构级的解耦把 Source 拆成 Framework 层和 Connector 层引入了 YAML 描述的数据管道Pipeline原生支持 schema evolution、表结构变更同步、路由route和字段转换transform。同时包名从com.ververica.cdc迁移到了org.apache.flink.cdc这个变化意味着很多老代码升级时第一件事就是改 import。我在实际项目里的建议是新项目直接上 2.4 之后的稳定版本或 3.x别再碰 1.x。如果你现在还在 1.x 上跑升级前务必确认两件事——下游消费逻辑能否处理UPDATE_BEFORE事件以及状态是否兼容增量快照框架的状态结构与旧版完全不同不能直接从 savepoint 恢复。2. 原理拆解一条 update 是怎么流进来的2.1 全量与增量的无缝衔接是整个组件最难的部分先说结论全量快照和增量日志之间天然存在一个时间缝隙。你在 T1 时刻开始读全量读到 T2 结束这中间数据库又发生了大量变更。如果只读全量T1 到 T2 之间的变更就丢了如果只读增量T2 之前的存量数据又没有。Flink CDC 的解法是先记位点再读快照再回放位点之后的日志。具体来说作业启动时先向 MySQL 发起一次SHOW MASTER STATUS拿到当前的 binlog 文件名和 position这就是起始位点然后再开始读全量数据。全量读完之后从刚才记下的位点开始消费 binlog。这样 T1 到 T2 之间的所有变更都会在增量阶段被重新应用一遍。那么问题来了如果一行数据在 T1 之前就存在全量阶段读到了它的最新值在 T1 到 T2 之间它又被改了一次那么增量阶段回放这条 update 时会再应用一次。这不会出错因为应用的是最终值覆盖幂等。但如果这行数据在 T1 到 T2 之间被删除了呢全量阶段读到了它因为读的时候它还在增量阶段回放 delete最终结果正确。反过来T1 之后新增的行全量读不到增量回放 insert也对。真正棘手的是并发快照场景。2.x 把表切成 chunk 并行读每个 chunk 读到的时间点不一样。假设 chunk A 在 10:00 读完chunk B 在 10:05 才读完那么 chunk B 里的数据比 chunk A 新。如果 chunk A 里某行在 10:02 被修改了增量阶段回放这条 updatechunk A 的数据会被正确覆盖成新值——这没问题。但如果某行在 10:02 被删除而这条 delete 在 binlog 里的位置早于某些 chunk 的读取时间去重逻辑就必须能识别出来。Flink CDC 用的是**低水位 / 高水位low watermark / high watermark**机制。每个 chunk 读取时会记录两个位点开始读之前记 low watermark读完记 high watermark。binlog 事件在低水位之前的部分说明对应的数据在快照读取时已经是最新的直接丢弃快照数据、保留 binlog位于低水位和高水位之间的事件需要和快照数据做去重按主键判断哪边更新高水位之后的事件直接应用。这个算法用一个更直白的类比你在抄一份正在被修改的名单。你抄到第 3 页的时候有人把第 1 页的一个名字改了。你在开抄前拍了张照片low watermark抄完后也拍一张high watermark。回看照片发现这个名字在两张照片之间被改过那你就以修改记录为准把你抄的内容覆盖掉。逻辑一致。2.2 Debezium 在整条链路里的角色翻译官不是搬运工很多人以为 Flink CDC 是 Flink 自己解析 binlog 的其实不是。真正跟 MySQL 建立复制连接、解析 binlog 二进制协议、还原成结构化变更事件的是Debezium。Flink CDC 在 2.x 里仍然内置了 Debezium 引擎3.x 通过pipeline也还是复用它只是把它包在了增量快照框架里。分工大致是这样Debezium 负责跟数据库说话把 binlog event 解析成SourceRecord里面的关键字段是before变更前镜像、after变更后镜像、op操作类型ccreate、uupdate、ddelete、rread 快照读。Flink CDC 的DebeziumDeserializationSchema再把它翻译成 Flink 的RowData或自定义对象往下游发。op字段的取值直接决定了下游怎么处理op 值含义触发场景下游建议处理r快照读全量阶段读取存量数据当作 INSERT 处理c插入新增一行INSERT / upsertu更新修改一行拆成 UPDATE_BEFORE UPDATE_AFTERd删除删除一行DELETE这里有个很容易被忽略的细节Debezium 默认会把一条 update 拆成两条事件——UPDATE_BEFORE带变更前的 before 镜像after 为 null和UPDATE_AFTER带变更后的 after 镜像。如果你用的是 Flink SQL 的 CDC connector 且声明了主键框架会自动处理成 changelog 流你感知不到但如果你用 DataStream API 且用了JsonDebeziumDeserializationSchema发到下游 Kafka 的会是两条消息消费端必须自己处理这种半条记录的情况。我踩过一次坑下游是个只处理after字段的消费程序遇到UPDATE_BEFORE事件时after是 null直接 NPE 把整个消费线程打挂了。后来统一在中间加了一层过滤只保留UPDATE_AFTER或者干脆改用 upsert 语义的下游表。2.3 位点持久化Checkpoint 到底存了什么这是容错的核心。Flink CDC Source 在每次 Checkpoint 时会把当前已经处理完的 binlog 位点文件名 position或者 GTID、当前快照的分片进度、已经完成的 chunk 列表一并写进状态后端。作业失败重启时从最近一次成功的 Checkpoint 恢复Source 会从保存的位点继续消费而不是从头再来。这里有个非常重要的前提MySQL 的 binlog 保留时间必须大于作业可能的最长恢复时间。假设 binlog 只保留 3 天而作业因为某种原因挂了 4 天还没恢复那恢复时需要的位点已经被 MySQL 清理掉了作业会直接报错只能从最新的位点重新开始也就是丢数据。我一般会在参数里显式加上debezium.database.history.instance.name并且确保同一张表在同一个 Flink 作业里只用一个实例名。如果两个作业用了同一个 instance name 去读同一个库Debezium 的 offset store 会互相打架表现出来就是位点莫名其妙跳变。另外MySQL 的expire_logs_days或者 8.0 的binlog_expire_logs_seconds建议设置成至少 7 天重要的库设 14 天。这个空间成本跟业务中断的代价比起来完全值得。2.4 并行度与分片为什么大表初始化也能跑起来2.x 的增量快照框架让 Source 并行度真正生效了。分片逻辑是这样先查表的 min 和 max 主键值按scan.incremental.snapshot.chunk.size默认 8096 行切成 N 个区间每个区间就是一个 chunk分给不同的 subtask 并行读。举个例子一张表主键从 1 到 1000 万chunk size 设 8096理论上会切出 1200 多个 chunk。Source 并行度设 4就是 4 个 subtask 轮流领 chunk 去读。这个吞吐提升是线性的实测下来 4 并发比单并发快将近 4 倍。但这里有两个约束必须清楚。第一切分列必须是主键或唯一索引列且类型最好是数值或字符串等可比较类型。如果表没有主键切分退化成单 chunk又回到单并发。第二每个 subtask 在连接 MySQL 时需要一个独立的server-id配置方式是server-id5400-5403范围写法如果范围比并行度小作业会起不来。还有一个隐性成本并发读快照时每个 subtask 都在跑SELECT ... WHERE pk BETWEEN ? AND ?这些查询会走主键索引压力可控但如果表上有大量二级索引或者查询条件命中了非主键索引对 IO 的冲击会明显起来。我的做法是把快照分片放到业务低峰期启动或者临时调小scan.snapshot.fetch.size默认 1024来降低单次查询的数据量。3. 实操从零搭一条 MySQL 到下游的实时链路3.1 环境与依赖版本矩阵必须先对齐版本不匹配是这个组件最高频的玄学问题来源。Flink 版本、Flink CDC 版本、Debezium 版本三者之间有严格的对应关系装错了可能编译通过但运行时报NoSuchMethodError。我目前验证过比较稳的组合是这样的Flink 版本Flink CDC 版本说明1.14 ~ 1.162.3.x / 2.4.x较老但生态稳定很多存量项目在用1.17 ~ 1.182.4.x主流组合SQL connector 支持完善1.19 ~ 1.203.0.x / 3.1.x新架构包名变更Pipeline 能力完整依赖引入方式有两种。如果用 Maven 管理DataStream API 加这个dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-mysql-cdc/artifactId version3.1.0/version /dependency如果是 SQL 方式或者用 Flink 发行包直接提交作业把对应版本的 connector jar 丢进$FLINK_HOME/lib就行。注意 3.x 之后 artifactId 有过调整2.x 时代是flink-sql-connector-mysql-cdc别搞混。一个很实用的小技巧用mvn dependency:tree看一下有没有 Debezium 的多个版本被间接引入。我就遇到过一次项目里另一个依赖带了debezium-core:1.9和 CDC 内置的debezium-core:2.4冲突表现是启动时报ClassCastException排查了两个小时才发现是依赖树的问题。用maven-shade-plugin做 relocations 能彻底解决。3.2 MySQL 侧的准备这几项没配好后面全是白干源库侧要改的东西不多但每一项都是硬性要求漏一项作业起不来。第一步确认 binlog 开启且格式为 ROW-- 检查 binlog 状态 SHOW VARIABLES LIKE log_bin; -- 必须是 ON SHOW VARIABLES LIKE binlog_format; -- 必须是 ROW SHOW VARIABLES LIKE binlog_row_image; -- 必须是 FULL SHOW VARIABLES LIKE binlog_expire_logs_seconds; -- 至少 6048007天binlog_format必须是 ROW因为只有 ROW 格式才记录行级的完整镜像。binlog_row_image必须是 FULL如果是 MINIMALupdate 事件里只带被修改的列Debezium 拿不到完整的 before 镜像下游做整行覆盖时会出问题。这两项在 MySQL 8.0 上默认值就是对的但 5.7 上binlog_row_image有些镜像被人改过一定要确认。第二步建一个专用的同步账号只给必要权限CREATE USER cdc_reader% IDENTIFIED BY YourStrongPassw0rd; GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO cdc_reader%; FLUSH PRIVILEGES;REPLICATION SLAVE和REPLICATION CLIENT是用来建立复制连接和查位点的SELECT是全量阶段读数据用的RELOAD在某些快照模式下会用到FLUSH TABLES WITH READ LOCK。别图省事直接给ALL PRIVILEGES同步账号泄露的风险比想象中大。第三步也是最容易被忽略的一步确认server-id不冲突。CDC 客户端在 MySQL 眼里就是一个从库必须有一个唯一的server-id。如果同一个库上已经有真实的主从复制在跑或者有其他 CDC 任务分配的server-id范围一定要避开。-- 查看当前 server-id SHOW VARIABLES LIKE server_id;我在参数里一般写成server-id5400-5403配合并行度 4。如果多个作业读同一个库就给不同的范围比如第二个作业用5404-5407。3.3 DataStream API 写法适合需要自定义处理的场景如果你的下游逻辑比较复杂比如要按事件类型做分支、要做字段级别的转换、要写到多个目标端DataStream API 更灵活。下面这段是完整可跑的骨架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.serialization.SimpleStringSchema; import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema; import org.apache.flink.connector.kafka.sink.KafkaSink; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class CdcSyncJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); // Checkpoint 是位点持久化的前提间隔太大会放大丢数据窗口太小会增加源库压力 env.enableCheckpointing(3000L); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L); env.getCheckpointConfig().setCheckpointTimeout(120000L); MySqlSourceString source MySqlSource.Stringbuilder() .hostname(10.0.0.11) .port(3306) .databaseList(biz_order) // 支持正则如 biz_.* .tableList(biz_order.t_order, biz_order.t_order_item) .username(cdc_reader) .password(YourStrongPassw0rd) .serverId(5400-5403) .serverTimeZone(Asia/Shanghai) .startupOptions(StartupOptions.initial()) .deserializer(new JsonDebeziumDeserializationSchema()) .includeSchemaChanges(true) // 订阅 DDL 变更 .splitSize(4096) // 每个 chunk 的行数 .fetchSize(1024) // 每次网络往返拉多少行 .build(); DataStreamSourceString stream env.fromSource( source, WatermarkStrategy.noWatermarks(), mysql-cdc-source) .setParallelism(4); KafkaSinkString sink KafkaSink.Stringbuilder() .setBootstrapServers(10.0.0.21:9092) .setRecordSerializer(KafkaRecordSerializationSchema.builder() .setTopic(cdc_order_binlog) .setValueSerializationSchema(new SimpleStringSchema()) .build()) .build(); stream.sinkTo(sink).setParallelism(4); env.execute(mysql-cdc-to-kafka); } }几个参数值得单独说。startupOptions(StartupOptions.initial())表示首次启动做全量快照、之后接着读增量这是最常用的模式如果下游已经有全量数据只想接增量用latest()如果想从某个特定时间点开始补用timestamp(毫秒值)。includeSchemaChanges(true)会额外产出一类特殊的 schema change 事件下游能感知到表结构变更。这个开关在 2.x 里是老版本的替代方案老版本通过debezium.skipped.operations控制3.x 才真正把 schema evolution 做完整。splitSize和fetchSize是两个直接影响性能的参数后面调优部分会算。3.4 Flink SQL 写法十分钟出一条链路如果只是标准的两端同步没有复杂的转换逻辑SQL 方式效率高得多不用编译打包纯配置。-- 1. 定义源表 CREATE TABLE t_order ( id BIGINT, order_no STRING, user_id BIGINT, amount DECIMAL(18, 2), status INT, create_time TIMESTAMP(3), update_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname 10.0.0.11, port 3306, username cdc_reader, password YourStrongPassw0rd, database-name biz_order, table-name t_order, server-id 5400-5403, server-time-zone Asia/Shanghai, scan.startup.mode initial, scan.incremental.snapshot.enabled true, scan.incremental.snapshot.chunk.size 4096, scan.snapshot.fetch.size 1024 ); -- 2. 定义目标端示例用 print实际换成 jdbc/kafka/doris 等 CREATE TABLE t_order_sink ( id BIGINT, order_no STRING, user_id BIGINT, amount DECIMAL(18, 2), status INT, create_time TIMESTAMP(3), update_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector print ); -- 3. 提交同步 INSERT INTO t_order_sink SELECT * FROM t_order;PRIMARY KEY (id) NOT ENFORCED这一行非常关键。声明了主键Flink SQL 才会把源表当作 changelog 处理下游才能收到带I、-U、U、-D语义的变更。如果忘了写作业虽然能跑但下游只能看到 insertupdate 和 delete 全丢了——这个问题我在别人的项目里见过不止一次。如果目标端是支持 upsert 的表比如 JDBC 到 MySQL、Doris、StarRocksINSERT INTO会自动编译成按主键的 upsert不会产生-U事件效率更高。3.5 参数调优这几个数字背后都是有算法的调参之前要先搞清楚瓶颈在哪儿。全量阶段和增量阶段的瓶颈完全不同全量阶段是 IO 密集看重 chunk 切分和并发增量阶段是网络和解析看重 fetch size 和下游反压。scan.incremental.snapshot.chunk.size默认 8096决定每个分片多大。设太大单个 chunk 读得久并发度上不去设太小chunk 数量爆炸元数据开销上来。我的经验公式是chunk size ≈ 期望单 chunk 耗时(秒) × 单秒扫描行数。一般业务表设 2048 到 8192 之间比较合适超大宽表单行 10KB 以上建议降到 1024 到 2048。scan.snapshot.fetch.size默认 1024每次网络往返拉取的行数。这个值直接影响源库内存和网络开销。宽表建议 256 到 512窄表可以到 2048 甚至 4096。connection.pool.size默认 20连接池大小。并发高、表多的时候要相应调大经验值是并行度 × 表的数量 × 1.5但不要超过 MySQL 的max_connections的 20%。Checkpoint 间隔这是容错窗口的直接决定因素。间隔 3 秒意味着最坏情况下重复消费 3 秒的数据但如果下游不幂等这 3 秒的重复就会变成脏数据。如果下游写入很重每次都做 Checkpoint 会明显拖慢吞吐。我的做法是下游幂等比如写入带主键的表就用 3 到 5 秒下游不幂等或者写外部 API就缩到 1 到 2 秒同时把minPauseBetweenCheckpoints设成间隔的一半避免连续 Checkpoint 把作业压死。server-id与并行度的关系范围的大小必须 ≥ 并行度。设并行度 4范围写5400-5403包含 4 个值。写5400-5402只有 3 个值作业启动时直接抛The required server id range is not enough。4. 常见问题与排查实录4.1 报错速查表这几条覆盖了八成故障报错关键词根因处理方式binlog_format must be ROW源库 binlog 格式不对改binlog_formatROW并重启 MySQLThe required server id range is not enoughserver-id 范围小于并行度扩大范围或降低并行度Could not find first log file name in binary log index file位点对应的 binlog 已被清理提高 binlog 保留时间从最新位点重启Communications link failure网络中断或连接被 kill加大connect.timeout检查中间网络设备空闲断连Encountered change event for table ... whose schema isnt known全量阶段表结构变更重启作业重新快照或开启 schema change 订阅java.lang.OutOfMemoryError: Java heap spacechunk 太大或 fetch size 太大调小 chunk size 和 fetch sizeThe server time zone value xxx is unrecognized时区配置不匹配显式设置server-time-zoneDatabase not found/ 表读不到权限或库名大小写问题检查lower_case_table_names与授权范围4.2 数据延迟正常但数值不对几个隐蔽的陷阱比报错更麻烦的是作业跑得好好的数据就是不对。这类问题我遇到过几次原因都挺隐蔽。时区错位。MySQL 里create_time是2024-01-15 10:00:00同步到下游变成了02:00:00。这是server-time-zone和 Flink 的table.local-time-zone不一致导致的。Debezium 会按配置的时区把 datetime 转成时间戳如果中间任何一个环节时区设错了就会出现固定偏移。解决办法是在 connector 参数里显式指定server-time-zone Asia/Shanghai同时在 Flink 的flink-conf.yaml里设env.java.opts: -Duser.timezoneAsia/Shanghai。大小写敏感。源库在 Linux 上lower_case_table_names0表名是T_Order而建 Flink 表的时候写的是t_order直接读不到。这个问题在不同环境的 MySQL 迁移时特别常见。无主键表。表没有主键Flink SQL 里又不能声明主键那么下游收到的 changelog 里 update 和 delete 无法定位到具体行容易出现重复插入。无主键表做 CDC 同步基本等于给自己埋雷我一般要求业务方要么加主键要么用整行所有字段做联合唯一键。字段类型映射。TINYINT(1)在 MySQL 里常被当作 boolean 用Debezium 可能把它解析成Boolean而不是数字下游类型对不上就报错。DECIMAL的精度在源表是DECIMAL(10,2)但 Flink 表写的是DECIMAL(18,2)某些情况下会出现精度丢失。这类问题只能靠严格的 Schema 对齐来防。update 事件的 before/after。前面提过如果用了JsonDebeziumDeserializationSchema但没处理UPDATE_BEFORE会收到afternull的半条记录。稳妥做法是在 Source 之后加一个 flatMap把u类型的事件拆成两条独立记录或者直接丢掉 before 那条。4.3 全量阶段把源库拖垮了怎么办这是运维层面最头疼的场景。一张 5000 万行的表Source 并行度 8狂读 10 分钟源库的 QPS 监控直接飙红业务接口超时。事后复盘发现是并发渡过高的锅。现在我的标准做法是三个动作。第一限制单次查询的数据量把scan.snapshot.fetch.size从 1024 降到 256scan.incremental.snapshot.chunk.size从 8096 降到 2048。这两个一降单次扫描的内存占用和锁持有时间都下来了。第二控制并发Source 并行度不要超过 4如果源库本身负载就高甚至先用 2 跑完全量再提上去。第三错峰执行把首次全量放在凌晨 2 点到 5 点之间启动startupOptions用initial()全量结束之后自动切增量之后就不再有大压力了。还有一招是从只读从库读全量。MySQL 主从架构下全量快照的读压力可以导到从库上但要注意 binlog 位点必须从主库拿或者确保从库的位点和主库一致Seconds_Behind_Master为 0。混用主从做 CDC 容易出问题我更倾向于直接在从库上做完整同步主库只负责业务写入。一个反直觉的经验并行度不是越高越好。我试过把并行度从 4 提到 16全量耗时只从 12 分钟降到 9 分钟但源库 CPU 从 30% 冲到 85%。因为 MySQL 的 IO 能力是有上限的并发再高也只是把请求排队。找到耗时曲线开始变平的那个点就足够了。4.4 DDL 变更这条链路最容易断的地方业务方加了个字段ALTER TABLE t_order ADD COLUMN remark VARCHAR(255)然后 CDC 作业就挂了。原因是 Flink SQL 里定义的表结构没有这个字段解析到未知列时直接异常。处理方式分两种。如果用的是 SQL 方式且没有开启 schema evolution只能停作业、改表定义、从 savepoint 恢复。这个操作在 Flink 1.17 之后可以通过table.dynamic-table-options.enabled配合动态参数调整但过程还是挺麻烦。3.x 之后有了原生的 schema evolution配scan.newly-added-table.enabled true和 pipeline 的 schema change 相关配置能自动把新加的列同步过去。不过说实话我在生产环境里对自动 DDL 同步还是比较谨慎——增列一般没问题但改列类型、删列这种操作自动同步到下游很可能把下游表搞坏。我的做法是作业里只订阅 schema change 事件并告警不自动应用人工确认后再走变更流程。这样虽然土一点但出了事能追溯。5. 生产落地容量估算和监控指标5.1 并行度和资源怎么算给一张表配多少资源我一般按这个顺序推先估增量速率再定并行度最后定内存。增量速率好估看 MySQL 的 binlog 生成速度就行。执行SHOW MASTER STATUS间隔 10 秒取两次用Position差值除以 10 得到每秒字节数。假设是 2MB/sFlik CDC 单并行度的处理能力大约在 3 到 5MB/s跟反序列化复杂度有关那么 1 个并行度就够。考虑到峰值配 2 到 4 比较稳妥。全量阶段则看表大小和主键分布。总行数除以 chunk size 得到 chunk 数量chunk 数量除以并行度得到每个 subtask 要处理的 chunk 数再乘以单 chunk 平均耗时就是全量总耗时。内存这一块TaskManager 的堆内存建议至少是并行度 × fetch size × 单行字节数 × 4。举个具体的并行度 4fetch size 1024单行平均 2KB那么 4 × 1024 × 2KB × 4 ≈ 32MB只是缓冲区的基础开销实际还要加上状态后端、反序列化对象、下游算子缓冲所以单个 TM 配 4GB 堆是比较合理的起点实测 8GB 更宽裕。5.2 必须盯住的几个指标作业跑起来不等于没事下面这几个指标我每天都会看。currentFetchEventTimeLag和currentEmitEventTimeLag是最重要的两个延迟指标。前者是数据进入 Source 到被处理的延迟后者是处理完到发出的延迟。这两个数突然往上走说明要么源库写入暴涨要么下游反压了。正常情况应该稳定在几百毫秒以内。sourceIdleTime反映 Source 空闲时间。如果一直很高说明没有新数据进来这时候延迟指标高是正常的不用报警。反之如果sourceIdleTime是 0 但延迟在涨那才是真的有问题。numRecordsInPerSecond用来判断吞吐是否正常如果突然掉到 0基本就是作业卡住了。Checkpoint 相关指标里lastCheckpointDuration如果持续超过间隔的一半说明状态太大或者下游太慢需要调大间隔或者优化状态。lastCheckpointSize持续增长而不回落通常意味着算子状态没清理干净。最后再补一个实战经验上线前一定要做故障演练。手动 kill 掉一个 TaskManager看作业能不能从 Checkpoint 自动恢复恢复后数据有没有重复或丢失。我做过的项目里至少有三分之一的CDC 可靠性问题是在演练时才暴露出来的——比如下游没做幂等、binlog 保留时间不够、Checkpoint 目录权限不对。这些问题在平稳运行时完全看不出来一旦真出事就是数据事故。演练完记得校验一遍源库和下游各取 1000 行按主键比对确认终态一致。这套东西搭起来之后日常运维工作量其实很小真正花时间的是一次性的参数对齐和上线前的那几轮压测。我自己的习惯是给每个 CDC 作业写一份配置清单把版本组合、源库参数、并行度、chunk size、Checkpoint 间隔这些固定下来新项目直接复用能省掉大量重复踩坑的时间。