企业官网建设流程全解析

在实时数仓和业务库异构同步这两个场景里做了几年，绕不开的一个组件就是 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

做选型的时候我一般会把候选方案摊开，按几个维度打分：运维复杂度、端到端延迟、全量能力、断点续传可靠性、是否依赖中间件。

表格里最后一列是我最终倾向 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.mode=initial会加全局读锁来保证快照一致性），一张几千万行的表锁上十几分钟，业务直接告警。另外它是单并发的，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 被修改了，增量阶段回放这条 update，chunk 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（操作类型：c=create、u=update、d=delete、r=read 快照读）。Flink CDC 的DebeziumDeserializationSchema再把它翻译成 Flink 的RowData或自定义对象，往下游发。

op字段的取值直接决定了下游怎么处理：

这里有个很容易被忽略的细节：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-id=5400-5403（范围写法），如果范围比并行度小，作业会起不来。

还有一个隐性成本：并发读快照时，每个 subtask 都在跑SELECT ... WHERE pk BETWEEN ? AND ?，这些查询会走主键索引，压力可控，但如果表上有大量二级索引，或者查询条件命中了非主键索引，对 IO 的冲击会明显起来。我的做法是把快照分片放到业务低峰期启动，或者临时调小scan.snapshot.fetch.size（默认 1024）来降低单次查询的数据量。

3. 实操：从零搭一条 MySQL 到下游的实时链路

3.1 环境与依赖：版本矩阵必须先对齐

版本不匹配是这个组件最高频的"玄学问题"来源。Flink 版本、Flink CDC 版本、Debezium 版本三者之间有严格的对应关系，装错了可能编译通过但运行时报NoSuchMethodError。

我目前验证过比较稳的组合是这样的：

依赖引入方式有两种。如果用 Maven 管理，DataStream API 加这个：

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>3.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'; -- 至少 604800（7天）

binlog_format必须是 ROW，因为只有 ROW 格式才记录行级的完整镜像。binlog_row_image必须是 FULL，如果是 MINIMAL，update 事件里只带被修改的列，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-id=5400-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); MySqlSource<String> source = MySqlSource.<String>builder() .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(); DataStreamSource<String> stream = env.fromSource( source, WatermarkStrategy.noWatermarks(), "mysql-cdc-source") .setParallelism(4); KafkaSink<String> sink = KafkaSink.<String>builder() .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语义的变更。如果忘了写，作业虽然能跑，但下游只能看到 insert，update 和 delete 全丢了——这个问题我在别人的项目里见过不止一次。

如果目标端是支持 upsert 的表（比如 JDBC 到 MySQL、Doris、StarRocks），INSERT 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 报错速查表：这几条覆盖了八成故障

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.timezone=Asia/Shanghai。

大小写敏感。源库在 Linux 上lower_case_table_names=0，表名是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，会收到after=null的半条记录。稳妥做法是在 Source 之后加一个 flatMap，把u类型的事件拆成两条独立记录，或者直接丢掉 before 那条。

4.3 全量阶段把源库拖垮了怎么办

这是运维层面最头疼的场景。一张 5000 万行的表，Source 并行度 8，狂读 10 分钟，源库的 QPS 监控直接飙红，业务接口超时。事后复盘发现是并发渡过高的锅。

现在我的标准做法是三个动作。第一，限制单次查询的数据量，把scan.snapshot.fetch.size从 1024 降到 256，scan.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/s，Flik CDC 单并行度的处理能力大约在 3 到 5MB/s（跟反序列化复杂度有关），那么 1 个并行度就够。考虑到峰值，配 2 到 4 比较稳妥。

全量阶段则看表大小和主键分布。总行数除以 chunk size 得到 chunk 数量，chunk 数量除以并行度得到每个 subtask 要处理的 chunk 数，再乘以单 chunk 平均耗时，就是全量总耗时。

内存这一块，TaskManager 的堆内存建议至少是并行度 × fetch size × 单行字节数 × 4。举个具体的：并行度 4，fetch 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 间隔这些固定下来，新项目直接复用，能省掉大量重复踩坑的时间。