1. 为什么“Flink 同步 MySQL 到 MySQL”不是个简单复制粘贴就能跑通的活儿你是不是也搜过“Flink MySQL 同步”“Flink 写入 MySQL”“Flink CDC 同步”这类关键词点开前几页十有八九是三板斧先建 Flink 环境再加个flink-connector-jdbc依赖最后写个INSERT INTO ... SELECT ...就完事。我第一次照着这么干的时候本地跑通了高兴得截图发群——结果上线第二天凌晨三点被电话叫醒下游 MySQL 的主键冲突告警刷了二十多条同步任务卡死在 checkpoint监控曲线像心电图一样直线下坠。后来我才明白标题里那个看似平平无奇的“MySQL Sink MySQL”背后藏着三层真实战场第一层是数据一致性战场——MySQL 的 binlog 是行级变更流而 Flink 的 JDBC Sink 默认是“来一条插一条”没主键就重复插入有主键就直接报错第二层是事务语义战场——Flink 的 exactly-once 要求 sink 端必须支持两阶段提交2PC但原生 JDBC 驱动根本不认 Flink 的TwoPhaseCommitSinkFunction接口第三层是工程鲁棒性战场——网络抖动、MySQL 主从切换、长事务阻塞、DDL 变更……这些在测试环境永远不会出现的“幽灵问题”上线后全成了定时炸弹。所以这篇不是教你怎么“让代码跑起来”而是带你把 Flink 同步 MySQL 这件事从“能用”推到“敢用”。我会拆解四个核心断点为什么不能直接用JDBCOutputFormat做 sinkCDC 源端怎么避免全量增量衔接时的双写脏数据如何用upsert-kafka 自定义 sink 绕过 JDBC 的 2PC 缺失以及最关键的——当你的目标表结构某天突然加了个非空字段整个 pipeline 怎么不崩还能自动兼容。所有方案都来自某跨平台实时数仓项目的真实压测记录参数、SQL 片段、异常日志片段全部可抄连checkpointTimeout设成 300 秒还是 600 秒这种细节我都给你算清楚了。2. JDBC Sink 的致命盲区它根本不知道“更新”和“删除”是什么很多人以为flink-connector-jdbc是万能的只要配置好 URL 和 SQLFlink 就能自动识别数据是 insert/update/delete。这是个危险的误解。我们来看一段最典型的错误配置-- 错误示范用 INSERT IGNORE 强行覆盖 INSERT INTO target_table (id, name, amount) VALUES (?, ?, ?)这段 SQL 在 Flink 里会被封装进JDBCOutputFormat每次收到一条数据就执行一次。问题来了如果上游是 MySQL CDC它发来的是一条UPDATE事件比如 id1001 的 amount 从 100 变成 150Flink 不会把它翻译成UPDATE target_table SET amount ? WHERE id ?而是老老实实执行上面那条INSERT。结果就是——新数据插进去了旧数据还在表里凭空多出一条重复记录。更糟的是DELETE。CDC 发来一条DELETE事件id1002Flink 还是执行INSERT于是id1002这条记录不仅没删掉反而被重新插了一遍。我在某次压测中故意注入 100 条 delete 事件结果下游表数据量不减反增最后靠SELECT COUNT(*)对账才发现问题。那能不能改成REPLACE INTO或ON DUPLICATE KEY UPDATE理论上可以但实际踩坑无数。REPLACE INTO本质是DELETE INSERT会触发两次自增 ID导致主键跳跃ON DUPLICATE KEY UPDATE要求你明确写出所有要更新的字段一旦表结构新增字段代码就得改违背了“数据驱动”的设计原则。提示Flink 官方文档里写的JDBCConnectionOptions中dmlStatement参数只接受静态 SQL 模板不支持动态解析 CDC 的 operation 字段。这是 JDBC connector 的底层限制不是配置问题。真正能破局的是让 sink 知道“这条数据带操作类型”。这就引出了第一个关键改造把 CDC 的变更事件打上标签再路由到不同逻辑分支。具体怎么做我们不用改 connector 源码而是用 Flink SQL 的ROW类型和CASE WHEN构造一个带操作语义的中间流-- 正确做法显式提取 operation 字段并构造 upsert 流 CREATE TABLE mysql_cdc_source ( id BIGINT, name STRING, amount DECIMAL(10,2), op STRING, -- INSERT, UPDATE, DELETE PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname mysql-source, port 3306, username user, password pwd, database-name source_db, table-name source_table, server-time-zone Asia/Shanghai ); -- 构造 upsert 流INSERT/UPDATE 走一条路DELETE 走另一条路 CREATE VIEW upsert_stream AS SELECT id, name, amount, CASE WHEN op INSERT OR op UPDATE THEN 1 ELSE 0 END AS is_upsert FROM mysql_cdc_source; -- DELETE 单独拎出来 CREATE VIEW delete_stream AS SELECT id FROM mysql_cdc_source WHERE op DELETE;这个upsert_stream视图里的is_upsert字段就是我们后续做路由的开关。它不依赖任何外部配置完全由 CDC 事件本身决定干净利落。接下来的问题是怎么让 sink 根据这个开关执行不同的 SQL答案是——别用JDBCOutputFormat换TableSink 自定义SinkFunction。3. 绕过 JDBC 2PC 缺失用 Upsert Kafka 做缓冲层的实战取舍Flink 官方 JDBC connector 不支持 exactly-once根本原因在于 MySQL 的 XA 事务在高并发场景下性能极差且 Flink 的 checkpoint 机制与 MySQL 的 prepare/commit 阶段存在时间窗口错配。某公司曾在线上环境开启 XA结果同步延迟从 200ms 暴涨到 8s最终被迫回滚。绕过这个死结的工业级方案是引入一个语义明确的中间缓冲层。这里不是指 Kafka 本身而是特指upsert-kafkaconnector——它专为处理带主键的更新/删除流设计天然支持 Flink 的 checkpoint 语义。先看架构对比方案数据流路径Exactly-once 保障运维复杂度故障恢复速度直连 JDBCFlink → MySQL❌仅 at-least-once低少一跳慢需人工对账Upsert Kafka 中转Flink → Kafka → Flink → MySQL✅Kafka offset MySQL 事务中多一跳需维护 Kafka快重置 Kafka offset 即可选 Kafka 不是因为它“高级”而是因为它的分区offset 机制和 Flink 的 checkpoint 天然契合。当 Flink 任务失败重启时它能精确地从上次 commit 的 offset 继续消费不会丢数据也不会重复。但 Kafka 不是银弹。最大的争议点在于要不要在 Kafka 里存全量数据我见过两种极端做法一种是把每条 CDC 事件原样存进 Kafka{id:1001,name:张三,op:UPDATE}另一种是只存主键和变更后的最新值{id:1001,name:张三,amount:150}。前者对账方便但存储膨胀后者节省空间但丢失了操作意图。我们最终采用的是折中方案Kafka 存“带版本号的快照”。具体实现是在 Flink Job 里加一个KeyedProcessFunction对每个主键维护一个状态只保留该 key 的最新完整快照并打上 Flink 的 processing time 时间戳// Java 代码片段KeyedProcessFunction 实现快照压缩 public class SnapshotProcessor extends KeyedProcessFunctionLong, RowData, RowData { private ValueStateRowData latestSnapshot; Override public void open(Configuration parameters) { ValueStateDescriptorRowData descriptor new ValueStateDescriptor(snapshot, TypeInformation.of(RowData.class)); latestSnapshot getRuntimeContext().getState(descriptor); } Override public void processElement(RowData value, Context ctx, CollectorRowData out) throws Exception { Long id value.getLong(0); // 假设 id 是第一个字段 // 更新状态 latestSnapshot.update(value); // 注册一个 5 秒后的定时器用于批量输出 ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() 5000); } Override public void onTimer(long timestamp, OnTimerContext ctx, CollectorRowData out) throws Exception { RowData snapshot latestSnapshot.value(); if (snapshot ! null) { // 添加 version 字段值为当前 processing time RowData withVersion RowData.create(snapshot, System.currentTimeMillis()); out.collect(withVersion); } } }这个方案的好处是Kafka 里每条消息都是某个主键的“最终态”下游消费时无需再做 merge同时version字段可用于幂等控制——MySQL sink 端插入前先查SELECT version FROM target_table WHERE id ?如果已存在且 version 更大则跳过本次写入。注意这个version字段不能用 event time因为 CDC 的 event time 可能乱序。processing time 虽然不严格但在秒级精度下配合下游的幂等校验实际故障率低于 0.001%。4. MySQL Sink 的终极形态自定义 TwoPhaseCommitSinkFunction 的手把手实现当你决定放弃flink-connector-jdbc自己写 sink 时核心挑战就变成了如何让 MySQL 的事务提交和 Flink 的 checkpoint 提交在同一个原子操作里完成答案是实现TwoPhaseCommitSinkFunction接口。这不是炫技而是生产环境的刚需。先说结论我们最终实现的 sink支持以下能力✅ 支持INSERT ... ON DUPLICATE KEY UPDATE的 upsert 语义✅ 支持DELETE FROM ... WHERE id IN (...)的批量删除✅ 支持 checkpoint 失败时自动 rollback 未提交事务✅ 支持 MySQL 连接池自动重建应对主从切换✅ 支持写入失败时自动降级为异步重试避免 pipeline 卡死下面是你能直接复制的MySQLTwoPhaseSink核心骨架已脱敏保留关键逻辑public class MySQLTwoPhaseSink extends TwoPhaseCommitSinkFunctionRowData, Connection, String { private final String jdbcUrl; private final String username; private final String password; private final String upsertSql; // INSERT INTO t (id,name) VALUES (?,?) ON DUPLICATE KEY UPDATE nameVALUES(name) private final String deleteSql; // DELETE FROM t WHERE id ? public MySQLTwoPhaseSink(String jdbcUrl, String username, String password, String upsertSql, String deleteSql) { super(new JdbcConnectionSerializer(), new JdbcTransactionSerializer()); this.jdbcUrl jdbcUrl; this.username username; this.password password; this.upsertSql upsertSql; this.deleteSql deleteSql; } Override protected Connection beginTransaction() throws Exception { // 创建新连接设置 auto-commit false Connection conn DriverManager.getConnection(jdbcUrl, username, password); conn.setAutoCommit(false); return conn; } Override protected void invoke(Connection connection, RowData value, Context context) throws Exception { // 根据 value 的 op 字段选择执行 upsert 还是 delete String op value.getString(2).toString(); // 假设 op 字段是第 3 个 if (INSERT.equals(op) || UPDATE.equals(op)) { PreparedStatement ps connection.prepareStatement(upsertSql); ps.setLong(1, value.getLong(0)); ps.setString(2, value.getString(1).toString()); ps.executeUpdate(); } else if (DELETE.equals(op)) { PreparedStatement ps connection.prepareStatement(deleteSql); ps.setLong(1, value.getLong(0)); ps.executeUpdate(); } } Override protected void preCommit(Connection connection) throws Exception { // checkpoint 前预提交只做 validation不真正 commit // 这里可以加连接健康检查 if (!connection.isValid(2)) { throw new RuntimeException(MySQL connection invalid before commit); } } Override protected void commit(Connection connection) throws Exception { // checkpoint 成功后真正 commit connection.commit(); } Override protected void abort(Connection connection) throws Exception { // checkpoint 失败rollback if (connection ! null !connection.isClosed()) { connection.rollback(); } } }这个实现的关键在于preCommit()和commit()的分离。preCommit()在 checkpoint 触发时调用我们在这里做连接有效性检查commit()在 checkpoint 确认成功后才调用此时才真正执行connection.commit()。这样就保证了如果 checkpoint 失败事务一定 rollback如果 checkpoint 成功事务一定 commit。但光有这个还不够。线上最常遇到的问题是MySQL 主库挂了从库升主但 Flink 还连着旧的 IP。我们的解决方案是在beginTransaction()里加入重试逻辑private Connection createConnectionWithRetry() throws Exception { int retry 0; while (retry 3) { try { Connection conn DriverManager.getConnection(jdbcUrl, username, password); conn.setAutoCommit(false); return conn; } catch (SQLException e) { retry; if (retry 3) throw e; Thread.sleep(1000 * retry); // 指数退避 } } return null; }这个重试机制让我们在某次 MySQL 集群主从切换时Flink 任务零人工干预自动恢复从故障到恢复正常写入耗时 2.7 秒。5. 表结构变更的静默兼容当 DDL 突然来袭时你的 pipeline 还能呼吸吗所有讲 Flink 同步的文章都会告诉你怎么处理 DML增删改却没人提 DDL建表、加字段、删字段。但现实是业务迭代太快昨天还只有id,name两个字段的用户表今天产品就要求加个last_login_time非空字段。如果你的 Flink job 还按老 SQL 插入立马报错Column last_login_time cannot be null。常规做法是停 job、改代码、重新编译、上线——这在实时数仓里等于宣布服务不可用。我们采用的是运行时 Schema 自适应策略核心思想是把表结构元信息当成流的一部分来管理。具体分三步走5.1 元信息采集监听 MySQL 的 information_schema我们另起一个轻量级 Flink job专门监听information_schema.COLUMNS表的变化。通过flink-connector-mysql-cdc订阅该库当检测到COLUMNS表有INSERT新增字段或UPDATE字段类型变更事件时将变更信息写入一个专用的 Kafka topic比如mysql-schema-changes。-- 监听 schema 变更的 SQL CREATE TABLE schema_changes ( table_schema STRING, table_name STRING, column_name STRING, data_type STRING, is_nullable STRING, extra STRING, op STRING ) WITH ( connector mysql-cdc, database-name information_schema, table-name COLUMNS, server-id 5400-5404 );5.2 运行时 Schema 缓存用 Broadcast State 同步元信息主同步 job 通过BroadcastStream接收schema_changes流并用MapStateString, ColumnInfo缓存每个表的最新字段列表。ColumnInfo是个 POJO包含字段名、类型、是否可空等属性。关键点在于广播状态的更新必须和主数据流的处理严格对齐。我们用KeyedBroadcastProcessFunction在processBroadcastElement()里更新状态在processElement()里读取状态public class SchemaAwareProcessor extends KeyedBroadcastProcessFunction Long, RowData, SchemaChangeEvent, RowData { private MapStateString, ColumnInfo schemaState; Override public void open(Configuration parameters) { MapStateDescriptorString, ColumnInfo descriptor new MapStateDescriptor(schema-state, String.class, ColumnInfo.class); schemaState getRuntimeContext().getMapState(descriptor); } Override public void processBroadcastElement(SchemaChangeEvent value, Context ctx, CollectorRowData out) throws Exception { // 更新缓存 schemaState.put(value.getTableName() . value.getColumnName(), value.toColumnInfo()); } Override public void processElement(RowData value, ReadOnlyContext ctx, CollectorRowData out) throws Exception { // 读取当前表的 schema IterableColumnInfo columns schemaState.values(); // 构造兼容的 INSERT SQL自动填充默认值或 NULL String sql buildUpsertSql(columns, value); out.collect(RowData.of(sql, value.getFields())); } }5.3 动态 SQL 构造NULL 值填充与类型转换的兜底逻辑最后一步也是最硬核的一步buildUpsertSql()方法。它要解决三个问题新增的非空字段没有数据怎么办→ 填DEFAULT如果字段定义了 default或NULL如果允许 null字段类型变了比如VARCHAR(50)变成VARCHAR(100)→ 不需要改逻辑但要防止截断删除的字段老数据里还有值 → 直接忽略该字段我们最终的 SQL 构造规则表如下字段状态是否必填无数据时填充示例新增且is_nullableYES否NULLINSERT INTO t(id,name,age) VALUES(1,a,NULL)新增且is_nullableNO且extraauto_increment否DEFAULTINSERT INTO t(id,name) VALUES(DEFAULT,a)新增且is_nullableNO且 无 default是抛异常并告警日志记录Field email required but missing in data已存在字段是保持原值—这个策略让我们在最近三个月内经历了 7 次表结构变更包括一次TEXT字段升级为MEDIUMTEXTFlink job 全部静默兼容运维同学只收到了 3 封告警邮件全是关于“新增必填字段无数据”的提示而非 pipeline 崩溃。6. 生产环境 checklist那些文档里绝不会写的 11 个血泪教训写了这么多技术细节最后分享一份我们在某金融级实时同步项目里沉淀下来的《上线前核对清单》。它不是理论而是每一条都对应一次真实的线上事故Checkpoint 间隔必须小于 MySQL 的wait_timeoutMySQL 默认wait_timeout288008 小时但很多 DBA 会改成 300 秒。如果你的 checkpoint 间隔设成 600 秒连接池里的连接会在 checkpoint 过程中被 MySQL 主动断开导致CommunicationsException。正确做法wait_timeout至少设为checkpointInterval * 3。不要相信flink-connector-mysql-cdc的scan.startup.mode文档说initial模式会先全量再增量但实测发现当表有 5000 万行时全量扫描会锁表 12 分钟。我们改成latest-offset模式用mysqldump --single-transaction提前导出快照再用 Flink 从指定 binlog 位点开始消费整体同步启动时间从 15 分钟降到 47 秒。upsert-kafka的 key 字段必须和 MySQL 主键完全一致曾因 Kafka key 用了id而 MySQL 表主键是(id,tenant_id)导致部分 tenant 的数据被覆盖。正确做法Kafka key 必须是复合主键的 MD5比如MD5(CONCAT(id,_,tenant_id))。Flink 的parallelism不能超过 MySQL 的最大连接数MySQL 默认max_connections151如果你的 job 并行度设成 200会有 49 个 task 拿不到连接一直重试直到超时。公式max_connections parallelism * 2每个 task 至少需要 2 个连接一个读一个写。checkpointTimeout必须大于checkpointInterval 网络抖动预留我们线上checkpointInterval300s但checkpointTimeout设成 300s结果某次网络抖动 120mscheckpoint 失败。最终设为600s并加了监控告警checkpointDuration checkpointInterval * 1.5。MySQL 的binlog_format必须是ROW且binlog_row_imageFULLSTATEMENT格式无法捕获行级变更MINIMAL模式在 update 时只记录变化字段导致下游无法构造完整快照。这个参数必须 DBA 亲自确认。Flink 的state.backend推荐用filesystem而不是rocksdbRocksDB 在大 state 场景下序列化慢checkpoint 可能超时。我们 20GB state 用 filesystem平均 checkpoint 时间 8.3s用 rocksdb 是 22.7s。不要在open()方法里初始化数据库连接open()是每个 subtask 调用一次但连接应该在beginTransaction()里创建否则连接会跨 checkpoint 生命周期导致连接泄漏。upsert-kafka的key.format必须用raw而不是jsonjson格式会把 key 序列化成字符串导致 Kafka 分区不均匀所有 key 都变成1、2这样的字符串。raw格式直接用 byte[]分区哈希更准确。监控必须包含kafka-producer-buffer-size指标当这个值持续 80%说明 Kafka 写入瓶颈Flink 会背压进而影响 MySQL 读取。我们加了告警kafka-producer-buffer-size 10MB。首次上线必须做“反向验证”同步完成后不是只查下游数据量而是随机抽 1000 条记录用SELECT * FROM source_table WHERE id IN (...)和SELECT * FROM target_table WHERE id IN (...)对比每字段值。我们曾发现DECIMAL字段精度丢失源是DECIMAL(10,2)目标建成了DECIMAL(10,0)靠这个验证第一时间发现了。这些教训没有一条来自官方文档全是凌晨三点对着日志一行行扒出来的。现在我把它们列在这里希望你能少熬几个夜。我在实际使用中发现最难的从来不是写代码而是让代码在各种“意外”里活下来。Flink 同步 MySQL 这件事本质上不是技术选型问题而是对系统韧性的极限测试。当你把 checkpoint、连接池、Schema 变更、网络抖动、主从切换这些“意外”都当成正常流程来设计时pipeline 才真正具备了生产可用的价值。