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 读数据主流就两种方式一种是基于CDCChange Data Capture的增量读取另一种是基于JDBC 的周期性轮询。很多人一开始分不清该用哪个其实判断标准很简单——你要的是变更通知还是定时快照。CDC 的思路是伪装成 MySQL 的从库订阅 binlog数据库里每发生一次 insert、update、deleteFlink 这边就能收到一条对应的事件。它是真正的实时延迟通常在毫秒到秒级。而 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 SourceJDBC 轮询 Source实时性毫秒到秒级取决于轮询间隔通常分钟级能否捕获删除能不能能否捕获中间状态能不能对 MySQL 压力低读 binlog高反复执行 select前置条件需开 binlog、授权只需 SELECT 权限全量增量支持需自行实现运维复杂度较高低选型逻辑我总结成一句话要实时、要变更明细、要删除感知选 CDC要简单、要一次性快照、数据量小选 JDBC。两者不是替代关系很多时候是配合使用的——CDC 同步主表JDBC 拉维表做关联。3. MySQL Sink 的写入模式从攒批到幂等3.1 JDBC sink 的三种写入语义Flink 写 MySQL 主要靠 JDBC connector它支持三种写入模式对应不同的语义保证这个必须搞清楚否则数据重复或丢失了都不知道为什么。Append 模式是最简单的就是纯 insert。适合日志类、流水类数据只增不改。配置上不需要主键也不需要额外参数。Upsert 模式是实际业务里用得最多的。它要求表定义里有主键写入时根据主键判断是 insert 还是 update。底层实现是INSERT ... ON DUPLICATE KEY UPDATEMySQL 原生支持效率不错。订单状态更新、用户信息同步这类场景都用它。幂等写入严格来说不算一种独立模式而是 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 和 00 表示不按时间刷只按条数。这个默认值在生产环境基本不能用——100 条一刷太频繁网络往返开销大而且如果数据流很稀疏可能半天攒不够 100 条数据就一直卡在缓冲区里。我的经验配置是max-rows设 1000 到 5000interval设 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 万条订单更新不同的攒批配置写入耗时差异很明显max-rowsinterval并行度总耗时备注10004186s默认配置频繁小批10002s462s推荐配置50005s448s吞吐最高延迟略高10002s841s并行度提升明显50005s839s并行度到 8 后收益递减从数据能看出从默认配置调到 1000/2s耗时直接降到三分之一。并行度从 4 提到 8 也有收益但再往上加受限于 MySQL 的写入能力提升就不明显了。所以并行度不是越大越好一般设成 MySQL 实例 CPU 核数的一半到一倍比较合适。4. 一条完整链路的搭建与验证4.1 环境准备清单在动手写 SQL 之前先把环境理清楚。我列一个清单照着检查一遍能省很多调试时间。MySQL 侧确认 binlog 已开且为 ROW 模式确认同步账号有 REPLICATION 权限确认源表和目标表都已建好确认目标表的主键和 Flink 声明一致。Flink 侧确认连接器 jar 包已放到 lib 目录。CDC 需要flink-connector-mysql-cdcJDBC 需要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和useSSLfalse8.x driver 可能直接连不上。建议统一用 8.x driver连接串写成jdbc:mysql://host:3306/db?useSSLfalseserverTimezoneAsia/ShanghaicharacterEncodingutf84.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?useSSLfalseserverTimezoneAsia/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/ShanghaiCDC source 配server-time-zone Asia/ShanghaiJDBC 连接串加serverTimezoneAsia/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 分钟配合增量 checkpointRocksDB 状态后端支持。状态后端方面数据量小的时候用内存就行但 CDC 任务的状态可能到 GB 级别建议用 RocksDB。配置上state.backend: rocksdb state.backend.incremental: true execution.checkpointing.interval: 60s execution.checkpointing.timeout: 10minstate.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 间隔这些都需要根据实际数据量和集群资源去调。调参没有银弹只有不断测试和观察。最后分享一个小习惯每次上线新的同步任务我都会先在小流量环境跑一天观察指标稳定后再切全量。这个习惯帮我避免了好几次可能影响主库的事故。实时链路一旦出问题排查成本远高于提前验证的成本。