1. 为什么 CDC 场景下 MySQL 读写是 Flink 的必修课

做实时数仓的朋友大概率都绕不开一个场景：业务库的数据要实时同步到下游，可能是另一张宽表、可能是消息队列、也可能是另一个存储系统。而业务库十有八九是 MySQL。这时候 Flink 就成了天然的中间层——它既能从 MySQL 拉数据，又能把处理结果写回 MySQL 或者写到别的地方。

我最早接触这个组合是在一个订单实时看板的需求里。业务方要求订单状态一变，看板上的数字就得跟着动，延迟不能超过几秒。当时第一反应是用定时任务轮询，但订单表数据量大、更新频繁，轮询既浪费资源又保证不了实时性。后来换成 Flink 的 MySQL CDC source，配合 JDBC sink 写结果表，整个链路才稳定下来。

这篇文章就围绕Flink 从 MySQL 读取数据（source）和写入数据（sink）这条主线展开。我会把两种 source 方案（CDC 和 JDBC 轮询）的取舍逻辑讲清楚，把 sink 端的几种写入模式掰开揉碎，再补上实际跑任务时踩过的坑和调优经验。不管你是刚接触 Flink 连接器的新手，还是已经跑过几个任务但总觉得哪里不对劲的老手，应该都能从里面找到对你有用的东西。

需要提前说明的是，Flink 的版本迭代比较快，连接器 API 在不同版本间有差异。我下面讲的内容主要基于 Flink 1.13 到 1.17 这个区间，这也是目前生产环境里用得比较多的版本段。如果你用的是更早或更晚的版本，个别配置项名称可能有出入，但核心思路是通的。

2. MySQL Source 的两条路：CDC 与 JDBC 轮询怎么选

2.1 先搞清楚两种 source 的本质区别

Flink 从 MySQL 读数据，主流就两种方式：一种是基于CDC（Change Data Capture）的增量读取，另一种是基于JDBC 的周期性轮询。很多人一开始分不清该用哪个，其实判断标准很简单——你要的是"变更通知"还是"定时快照"。

CDC 的思路是伪装成 MySQL 的从库，订阅 binlog，数据库里每发生一次 insert、update、delete，Flink 这边就能收到一条对应的事件。它是真正的实时，延迟通常在毫秒到秒级。而 JDBC 轮询是每隔一段时间执行一条 select 语句，把满足条件的数据捞出来。它只能感知到"当前状态"，感知不到"中间变化"，而且轮询间隔决定了延迟下限。

我见过不少团队一上来就用 JDBC 轮询，理由是"配置简单、不用开 binlog"。结果跑了一段时间发现，订单从"待支付"变成"已支付"再变成"已发货"，中间的状态跳变全丢了，下游只看到最终态。这种场景就必须上 CDC。

反过来，如果只是每天凌晨同步一次维表数据，或者数据量很小、变更不频繁，那 JDBC 轮询完全够用，还省去了维护 binlog 解析链路的成本。

2.2 CDC source 的配置要点与 binlog 前置条件

用 CDC source 之前，MySQL 服务端必须满足几个硬性条件，这些条件不满足的话任务根本起不来，而且报错信息往往不够直观，新手容易卡在这里。

第一，binlog 必须开启，且格式必须是ROW模式。STATEMENT模式记录的是 SQL 语句，CDC 解析不出具体的数据行变化；MIXED模式在某些语句下会退化成 STATEMENT，同样不可靠。检查命令是SHOW VARIABLES LIKE 'binlog_format'，返回必须是ROW。

第二，binlog_row_image 建议设为 FULL。这个参数决定了 binlog 里记录的是整行数据还是只记录变化列。设为FULL时，update 事件会带上变更前后的完整行数据，CDC 解析起来最省事。如果设成MINIMAL，只有主键和变化列，某些连接器版本会解析异常。

第三，需要给同步账号授予特定权限。除了常规的 SELECT 权限，还需要REPLICATION SLAVE和REPLICATION CLIENT权限，因为 CDC 本质上是模拟从库。授权语句大概是这样：

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'flink_user'@'%'; FLUSH PRIVILEGES;

配置层面，CDC source 的核心参数有这么几个。hostname、port、username、password是连接信息；database-name和table-name指定要同步的库表；server-id必须全局唯一，如果多个 Flink 任务连同一个 MySQL，每个任务的 server-id 不能重复，否则会互相踢下线；server-time-zone建议显式设置成Asia/Shanghai，不然时间字段可能出现时区偏移。

还有一个容易忽略的点：启动模式（startup mode）。默认是initial，意思是先做一次全量快照，再从当前 binlog 位置开始增量。如果表特别大，全量阶段会锁表或者消耗大量资源，这时候可以考虑latest-offset，只从最新位置开始读，跳过历史数据。选哪个取决于你的业务是否需要历史全量。

2.3 JDBC 轮询 source 的适用边界

JDBC source 的配置就简单多了，核心是四个东西：连接串、查询语句、轮询间隔、以及增量字段。

CREATE TABLE mysql_source ( id BIGINT, order_no VARCHAR, status INT, update_time TIMESTAMP, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://localhost:3306/biz_db', 'table-name' = 't_order', 'username' = 'flink_user', 'password' = 'xxxx', 'scan.partition.column' = 'id', 'scan.partition.num' = '4', 'scan.partition.lower-bound' = '1', 'scan.partition.upper-bound' = '1000000', 'scan.fetch-size' = '1000' );

这里scan.partition.*系列参数是为了并行读取，把一张表按 id 范围切成 4 份，4 个并行度同时拉。如果不配这些，默认单并行度，大表会非常慢。scan.fetch-size控制每次从结果集取多少行，设太小网络往返多，设太大内存压力大，1000 到 5000 是比较稳的区间。

但 JDBC source 有个根本性限制：它只能做有界读取，做不了真正的流式增量。虽然可以通过scan.auto-commit和定时调度模拟周期性查询，但每次查询都是独立的，无法感知删除操作，也无法保证 exactly-once。所以我的建议是，JDBC source 只用于维表关联或者小批量初始化，真正的实时同步一律走 CDC。

2.4 两种方案的对比与选型建议

选型逻辑我总结成一句话：要实时、要变更明细、要删除感知，选 CDC；要简单、要一次性快照、数据量小，选 JDBC。两者不是替代关系，很多时候是配合使用的——CDC 同步主表，JDBC 拉维表做关联。

3. MySQL Sink 的写入模式：从攒批到幂等

3.1 JDBC sink 的三种写入语义

Flink 写 MySQL 主要靠 JDBC connector，它支持三种写入模式，对应不同的语义保证，这个必须搞清楚，否则数据重复或丢失了都不知道为什么。

Append 模式是最简单的，就是纯 insert。适合日志类、流水类数据，只增不改。配置上不需要主键，也不需要额外参数。

Upsert 模式是实际业务里用得最多的。它要求表定义里有主键，写入时根据主键判断是 insert 还是 update。底层实现是INSERT ... ON DUPLICATE KEY UPDATE，MySQL 原生支持，效率不错。订单状态更新、用户信息同步这类场景都用它。

幂等写入严格来说不算一种独立模式，而是 Upsert 在特定条件下的表现。当主键确定且更新是幂等的（比如把 status 设成某个值，重复执行结果一样），那么即使任务重启导致部分数据重放，最终结果也是正确的。这也是为什么 Upsert 模式配合主键能提供 at-least-once 语义下的"结果正确"。

配置 Upsert 模式的关键是表定义里声明主键：

CREATE TABLE mysql_sink ( id BIGINT, order_no VARCHAR, status INT, update_time TIMESTAMP, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://localhost:3306/biz_db', 'table-name' = 't_order_result', 'username' = 'flink_user', 'password' = 'xxxx', 'sink.buffer-flush.max-rows' = '1000', 'sink.buffer-flush.interval' = '2s', 'sink.max-retries' = '3' );

注意PRIMARY KEY (id) NOT ENFORCED里的NOT ENFORCED，这是告诉 Flink 这个主键只用于优化，不需要它去实际校验。如果漏了这个关键字，建表会报错。

3.2 攒批参数怎么调才不翻车

sink.buffer-flush.max-rows和sink.buffer-flush.interval这两个参数直接决定了写入性能和延迟的平衡。

max-rows是攒够多少条就刷一次，interval是隔多久强制刷一次。两个条件是"或"的关系，谁先满足就触发。默认值分别是 100 和 0（0 表示不按时间刷，只按条数）。这个默认值在生产环境基本不能用——100 条一刷太频繁，网络往返开销大；而且如果数据流很稀疏，可能半天攒不够 100 条，数据就一直卡在缓冲区里。

我的经验配置是max-rows设 1000 到 5000，interval设 1 到 5 秒。这样既能保证吞吐，又能把延迟控制在可接受范围。如果下游对延迟极其敏感，interval 可以压到 200 毫秒，但吞吐会下降。

还有一个隐藏的坑：攒批是在每个并行度内独立进行的。假设你并行度是 8，每个并行度各自攒 1000 条，那实际一次批量写入可能是 8000 条分散在 8 个连接上。所以调大并行度时，单批的 max-rows 可以适当调小，避免单次写入过大导致 MySQL 压力集中。

3.3 主键冲突与重试机制

sink.max-retries控制写入失败后的重试次数，默认是 3。这个参数在遇到主键冲突、连接闪断时很有用。但要注意，重试只对可恢复的错误有效，比如网络抖动、死锁。如果是表结构不匹配、字段类型错误这种，重试多少次都没用，只会拖慢失败暴露的时间。

主键冲突在 Upsert 模式下一般不会发生，因为底层用的是ON DUPLICATE KEY UPDATE。但如果你的表定义声明了主键，实际写入的数据里主键有重复，且下游表的主键约束和 Flink 声明的不一致，就可能出问题。我遇到过一次，Flink 侧声明的主键是id，但 MySQL 表实际主键是(id, dt)联合主键，结果写入时频繁报主键冲突。后来把 Flink 表定义改成联合主键才解决。

所以这里有条铁律：Flink 表定义里的主键必须和 MySQL 实际主键完全一致，包括字段和顺序。

3.4 批量写入的性能实测数据

我在测试环境做过一组对比，同样的数据量（100 万条订单更新），不同的攒批配置，写入耗时差异很明显：

从数据能看出，从默认配置调到 1000/2s，耗时直接降到三分之一。并行度从 4 提到 8 也有收益，但再往上加，受限于 MySQL 的写入能力，提升就不明显了。所以并行度不是越大越好，一般设成 MySQL 实例 CPU 核数的一半到一倍比较合适。

4. 一条完整链路的搭建与验证

4.1 环境准备清单

在动手写 SQL 之前，先把环境理清楚。我列一个清单，照着检查一遍能省很多调试时间。

MySQL 侧：确认 binlog 已开且为 ROW 模式，确认同步账号有 REPLICATION 权限，确认源表和目标表都已建好，确认目标表的主键和 Flink 声明一致。

Flink 侧：确认连接器 jar 包已放到 lib 目录。CDC 需要flink-connector-mysql-cdc，JDBC 需要flink-connector-jdbc加上对应版本的 MySQL driver。这两个 jar 的版本必须和 Flink 版本匹配，否则会报NoSuchMethodError这类让人头大的错误。

依赖管理上我踩过一个坑：MySQL driver 的版本。5.x 和 8.x 的 driver 类名不同（com.mysql.jdbc.Drivervscom.mysql.cj.jdbc.Driver），而且 8.x 对时区和 SSL 的处理更严格。如果连接串里不加serverTimezone和useSSL=false，8.x driver 可能直接连不上。建议统一用 8.x driver，连接串写成：

jdbc:mysql://host:3306/db?useSSL=false&serverTimezone=Asia/Shanghai&characterEncoding=utf8

4.2 从源表到目标表的完整 SQL

下面是一条完整的链路，从 MySQL 源表 CDC 读取，经过简单处理后写入另一张 MySQL 结果表。我用 Flink SQL 写，因为这是最直观的方式。

-- 源表：订单表 CREATE TABLE source_order ( id BIGINT, order_no VARCHAR, user_id BIGINT, amount DECIMAL(10,2), status INT, update_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'localhost', 'port' = '3306', 'username' = 'flink_user', 'password' = 'xxxx', 'database-name' = 'biz_db', 'table-name' = 't_order', 'server-id' = '5401', 'server-time-zone' = 'Asia/Shanghai', 'scan.startup.mode' = 'initial' ); -- 结果表：订单汇总 CREATE TABLE sink_order_summary ( order_no VARCHAR, user_id BIGINT, amount DECIMAL(10,2), status INT, update_time TIMESTAMP(3), PRIMARY KEY (order_no) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://localhost:3306/biz_db?useSSL=false&serverTimezone=Asia/Shanghai', 'table-name' = 't_order_summary', 'username' = 'flink_user', 'password' = 'xxxx', 'sink.buffer-flush.max-rows' = '1000', 'sink.buffer-flush.interval' = '2s', 'sink.max-retries' = '3' ); -- 转换并写入 INSERT INTO sink_order_summary SELECT order_no, user_id, amount, status, update_time FROM source_order WHERE status IN (1, 2, 3);

这段 SQL 里，server-id我用了 5401，这个数字只要在同一个 MySQL 的所有 CDC 任务里唯一就行。scan.startup.mode设为initial，会先全量再增量。如果源表很大，第一次启动会跑比较久，这是正常的。

4.3 验证数据一致性的方法

任务跑起来之后，怎么确认数据是对的？我一般分三步验证。

第一步，看 Flink 任务的指标。在 Flink UI 上关注numRecordsIn和numRecordsOut，如果 in 和 out 数量对得上（考虑过滤条件），说明没有丢数据。再看 checkpoint 是否正常，如果 checkpoint 一直失败，说明有反压或者状态太大。

第二步，在 MySQL 里对账。源表改一条数据，等几秒，去结果表查对应的记录，看字段值是否一致。特别要测 update 和 delete——update 看结果表是否更新，delete 看结果表是否也删了（如果 sink 是 Upsert 模式，delete 需要额外配置，默认不传播删除）。

第三步，压测边界情况。比如源表短时间内大量更新，看 sink 端会不会积压；比如手动 kill 任务再重启，看数据会不会重复或丢失。Upsert 模式下重启后数据应该是一致的，因为主键保证了幂等。

这里有个细节：delete 操作的传播。CDC source 能捕获 delete，但 JDBC sink 默认不处理 delete 事件。如果业务需要同步删除，得在 sink 表定义里配置sink.delete-enabled或者用支持 delete 的 connector。这个点很多人会忽略，导致源表删了数据，结果表还留着。

5. 生产环境踩过的坑与调优经验

5.1 server-id 冲突导致任务反复重启

这个坑我印象最深。当时有两个 Flink 任务连同一个 MySQL，配置的时候随手写了两个 server-id，结果其中一个写重复了。表现是任务启动后跑几分钟就断，日志里报A slave with the same server_uuid/server_id as this slave has connected to the master。因为 MySQL 认为两个从库用了同一个 id，会踢掉先连的那个。

排查的时候一开始没往 server-id 上想，因为报错信息不够直白。后来查了 MySQL 的SHOW PROCESSLIST，看到两个 Binlog Dump 线程，才定位到问题。解决办法就是给每个 CDC 任务分配唯一的 server-id，我现在的习惯是用一个范围，比如 5400 到 5500，每个任务递增分配，并在任务配置里注释清楚。

5.2 时区不一致导致时间字段偏移 8 小时

另一个高频坑是时区。MySQL 里存的是2024-01-15 10:00:00，同步到结果表变成了2024-01-15 02:00:00，差了 8 小时。原因是 Flink 任务运行环境的时区、MySQL 服务端时区、连接串时区三者不一致。

解决方法是三处统一：MySQL 服务端设Asia/Shanghai，CDC source 配server-time-zone = Asia/Shanghai，JDBC 连接串加serverTimezone=Asia/Shanghai。三处对齐后时间就正常了。如果用的是 TIMESTAMP 类型，还要注意 Flink 里声明成TIMESTAMP(3)而不是TIMESTAMP_LTZ，后者会带时区转换，更容易出问题。

5.3 全量阶段锁表与资源抢占

CDC 的initial模式在全量阶段会执行类似SELECT * FROM table的操作。如果表很大且没有合适索引，这个查询可能锁表或者拖慢主库。我遇到过一次，源表几千万行，全量阶段直接把主库 CPU 打满，业务查询变慢，被 DBA 找上门。

后来调整了策略：全量阶段用scan.snapshot.fetch.size控制每次拉取的行数，避免一次性拉太多；同时把全量任务安排在业务低峰期执行；对于超大表，改用latest-offset跳过全量，历史数据用离线方式补。这几个措施下来，对主库的影响就小多了。

5.4 checkpoint 配置与状态后端选择

CDC 任务的状态会随着同步的表和时间增长。如果 checkpoint 间隔太短，频繁做快照会影响吞吐；间隔太长，故障恢复时要重放的数据多。我的经验是 checkpoint 间隔设 1 到 3 分钟，配合增量 checkpoint（RocksDB 状态后端支持）。

状态后端方面，数据量小的时候用内存就行，但 CDC 任务的状态可能到 GB 级别，建议用 RocksDB。配置上：

state.backend: rocksdb state.backend.incremental: true execution.checkpointing.interval: 60s execution.checkpointing.timeout: 10min

state.backend.incremental开启后，每次 checkpoint 只上传变化的部分，能显著减少 checkpoint 耗时。这个参数在状态大的任务里效果非常明显。

5.5 反压的定位与缓解

反压是流任务最常见的问题。CDC source 读得快，sink 写得慢，中间就会积压。表现是 Flink UI 上 source 的busy或backPressured指标变高，checkpoint 时间变长甚至超时。

定位反压要看是哪个算子慢。如果是 sink 慢，通常是 MySQL 写入跟不上，可以调大攒批、增加并行度、或者优化目标表索引。如果是中间算子慢，比如有聚合或 join，要看状态大小和 key 分布是否倾斜。

我处理过一次 sink 反压，原因是目标表没有主键索引，每次 upsert 都要全表扫描。加上主键后，写入速度从每秒几百条提升到几万条。所以目标表的索引设计很关键，别只顾着 Flink 侧调参，下游表结构同样重要。

6. 几个容易被忽略的细节

6.1 表结构变更的同步问题

源表加了一列，CDC 能捕获到 DDL 吗？答案是：取决于连接器版本和配置。较新的 CDC 版本支持 schema evolution，能自动感知列的增加。但 Flink 表定义是静态的，如果源表加了列而 Flink 表定义没更新，新列的数据会被忽略。

我的做法是，源表结构变更时，同步更新 Flink 表定义并重启任务。如果用的是initial模式，重启会重新全量，代价大；所以生产环境更推荐把 schema 变更纳入发布流程，提前规划好。

6.2 大事务的处理

MySQL 里一个事务如果更新了几十万行，binlog 里会是一个巨大的事件。CDC 解析这种大事务时可能内存溢出或者超时。遇到过几次，日志报The transaction is too large。

缓解办法是调整 CDC 的debezium.max.queue.size和debezium.max.batch.size参数，给足缓冲。但根本解决还是要业务侧避免超大事务，分批提交。这个需要和业务开发沟通，不是 Flink 单方面能解决的。

6.3 并行度与数据倾斜

CDC source 的并行度受限于表的数量，单表同步时并行度只能是 1（因为 binlog 是单线程的）。这时候如果数据量大，source 会成为瓶颈。解决办法是做分流，把一张大表按业务维度拆成多张表分别同步，或者用支持增量快照的 CDC 版本，它能把全量阶段并行化。

sink 端的并行度可以大于 source，但要注意主键分布。如果主键是自增 id，写入会集中在少数几个分区，造成倾斜。可以在写入前对主键做 hash 打散，或者用联合主键让分布更均匀。

6.4 监控指标该看哪些

任务上线不是终点，监控才是。我一般盯这几个指标：source 的numRecordsIn是否持续增长（判断是否卡住）、sink 的numRecordsOut和numRecordsIn是否匹配（判断是否丢数据）、checkpoint 的duration和failed次数（判断稳定性）、以及currentFetchEventTimeLag（判断 CDC 延迟）。

这些指标接到监控告警系统里，设置合理的阈值。比如 checkpoint 连续失败 3 次就告警，CDC 延迟超过 5 分钟就告警。有了监控，很多问题能在影响业务之前发现。

7. 写在最后的一点个人体会

Flink 读写 MySQL 这套组合，表面上看就是配几个参数的事，但真正跑在生产环境里，细节多到超出预期。我自己的经验是，配置只是入场券，对 MySQL 机制的理解才是稳定运行的关键。binlog 怎么工作、事务怎么提交、索引怎么影响写入，这些底层知识决定了你遇到问题时能不能快速定位。

另外，别迷信默认配置。Flink 连接器的默认值大多是为了兼容性，不是为了性能。攒批参数、并行度、checkpoint 间隔，这些都需要根据实际数据量和集群资源去调。调参没有银弹，只有不断测试和观察。

最后分享一个小习惯：每次上线新的同步任务，我都会先在小流量环境跑一天，观察指标稳定后再切全量。这个习惯帮我避免了好几次可能影响主库的事故。实时链路一旦出问题，排查成本远高于提前验证的成本。