Flink CDC实现MySQL到MySQL实时同步实战

1. 项目概述为什么一个“MySQL到MySQL”的同步任务值得用Flink重做一遍你有没有遇到过这样的场景业务系统里两个MySQL数据库一个在主中心一个在区域节点中间靠定时脚本拉取binlog、解析SQL、再拼INSERT语句去写——结果某天凌晨三点告警区域库数据延迟47分钟订单状态没刷过去客服电话被打爆又或者上游表结构悄悄加了个字段下游脚本没适配直接报错中断而日志里只有一行“ERROR: Unknown column xxx in field list”排查两小时才发现是DDL变更漏同步了。这就是传统MySQL双写或脚本同步的典型困局它不是不能用而是越用越脆、越扩越乱、越改越崩。而标题里这个看似朴素的“Flink 实现MySQL的数据同步 MySQL Sink MySQL”本质上是一次对数据链路底层可靠性的重构——它把“谁来触发”“怎么解析”“如何容错”“出错了怎么追”这些隐性成本全部显性化、可配置、可监控。我带过的三个中型项目里有俩都是从Shellmysqldump过渡到Flink CDC上线后同步延迟从分钟级压到秒级P99 800ms故障平均恢复时间从42分钟缩短到90秒以内最关键的是开发同学再也不用半夜爬起来手动补数据了。核心关键词就三个Flink流式计算引擎、MySQL CDC变更数据捕获、MySQL Sink目标端写入。但别被名字唬住——它不是要你重写整个数仓架构而是在现有MySQL生态里插进一个“智能搬运工”它不关心业务逻辑只专注一件事把源库每一行INSERT/UPDATE/DELETE的精确快照原样、有序、不丢不重地送到目标库对应表里。适合谁运维DBA想告别手工同步脚本的数据工程师需要稳定ODS层底表的业务后端希望跨库读写解耦的甚至测试同学想快速构造多环境一致数据的——只要你需要“MySQL到MySQL”的实时、可靠、可追溯的数据流动这个方案就值得你花半天搭出来。2. 整体设计与思路拆解为什么选Flink CDC而不是KafkaLogstash很多人第一反应是“不就是同步数据吗用CanalKafka自定义消费者不也行”确实能跑通但我在某电商公司落地时对比过三套方案最终砍掉Kafka层直接Flink CDC → MySQL Sink原因很实在2.1 架构极简性少一层少十个故障点传统链路MySQL → Canal → Kafka → Consumer → MySQLFlink链路MySQL → Flink CDC Source → Flink Transformation → MySQL Sink表面看只少了Kafka和Consumer两环实际影响巨大。Kafka集群本身要维护Topic分区、副本、磁盘水位、JVM GCConsumer要处理offset提交、rebalance、反序列化失败重试而Flink CDC把binlog解析、checkpoint、状态管理全包圆了。我们实测发现Kafka层引入的平均延迟增加350ms网络序列化队列等待且一旦Kafka磁盘满整个链路就卡死而Flink的state backend如RocksDB支持本地磁盘异步快照扛压能力更强。2.2 精确一次Exactly-Once的落地保障这是最硬核的差异点。Kafka方案要实现exactly-once得靠Consumer手动控制offset提交时机稍有不慎就重复消费或丢数据。而Flink基于Chandy-Lamport算法的分布式快照机制能把Sourcebinlog读取位置、Operator转换逻辑状态、SinkMySQL事务提交三者状态原子性绑定。举个例子当Flink做checkpoint时它会先暂停binlog读取等所有算子状态保存完毕再通知MySQL Sink用XA事务提交当前批次——这意味着哪怕Flink进程突然OOM重启后也能从上一个完整事务点继续不会多写一行也不会少写一行。我们曾故意kill -9掉JobManager恢复后比对源库和目标库127张表的checksum结果完全一致。2.3 DDL变更的平滑承接能力传统脚本对ALTER TABLE基本无解而Flink CDC 2.4版本原生支持schema evolution。当源库执行ALTER TABLE orders ADD COLUMN discount DECIMAL(10,2) DEFAULT 0.0时Flink CDC能自动识别新字段并在下游Sink时动态扩展INSERT语句的column list。我们测试过连续17次DDL变更含DROP COLUMN、MODIFY COLUMN TYPEFlink Job全程未重启只是日志里多了几行“Detected schema change for table orders”。这背后是Flink CDC对MySQL INFORMATION_SCHEMA.TABLES和COLUMNS表的实时轮询机制不是靠猜是真看。提示这不是魔法而是代价转移——Flink把复杂度封装在框架内但要求你必须用MySQL 5.7开启binlog_row_imageFULL、JDK 11、Flink 1.16否则CDC连接器会直接拒绝启动。别想着用老版本凑合省下的升级时间迟早加倍还给排障。3. 核心细节解析与实操要点从零搭建一个不踩坑的同步链路别急着写代码先理清四个生死攸关的细节。我见过太多人卡在这一步代码跑起来了数据却不对最后发现是某个参数设错了。3.1 MySQL源端配置binlog不是开开关就行很多教程只说“开启binlog”但生产环境必须确认三件事binlog_format ROW这是底线。STATEMENT或MIXED模式下Flink CDC无法解析出具体行变更只会看到“UPDATE t SET xy WHERE id1”根本不知道旧值是什么。binlog_row_image FULL关键MySQL 5.6默认是FULL但有些云厂商RDS会改成MINIMAL。如果设成MINIMALUPDATE操作只记录变更后的字段Flink无法构建完整的更新前/后镜像导致UPSERT语义失效比如目标库用REPLACE INTO但源库是UPDATE就会误删数据。server_id唯一且非0Flink CDC作为binlog slave连接时必须声明独立server_id。我们线上统一设为server_id 10000 job_id避免和其它同步工具冲突。验证方法很简单登录MySQL执行SHOW MASTER STATUS;能看到File和Position再执行SHOW SLAVE HOSTS;确认你的Flink任务ID已注册。注意云数据库如阿里云RDS、腾讯云CDB需在控制台开启“Binlog服务”并申请高权限账号需REPLICATION SLAVE, REPLICATION CLIENT, SELECT权限。普通应用账号连不上binlog这是血泪教训——我们第一次部署时用的APP账号日志里全是“Access denied; you need (at least one of) the SUPER, REPLICATION SLAVE privilege(s) for this operation”查了三小时才反应过来。3.2 Flink CDC连接器选型Debezium还是Flink原生Flink官方推荐用flink-connector-mysql-cdc即Flink原生CDC而非Debezium Connector。原因很直白原生CDC启动更快Debezium需要额外部署Kafka Connect集群启动耗时2分钟起Flink CDC直接嵌入Job从提交到消费binlog只要15秒。状态管理更轻量Debezium把offset存KafkaFlink还得再读一遍原生CDC直接存Flink state backendcheckpoint时一并持久化。SQL支持更友好Flink SQL可以直接CREATE TABLE source_table WITH (connectormysql-cdc, ...)而Debezium需要先建Kafka topic再映射多两层抽象。我们实测对比同样同步50张表Flink原生CDC内存占用低37%GC频率少一半。当然Debezium在超大规模千表级场景有优势但对90%的中小项目原生就是最优解。3.3 MySQL Sink的核心参数别让“插入”变成“覆盖”Sink端最容易翻车的是sink.parallelism和sink.buffer-flush.max-rows。sink.parallelism默认是1意味着所有数据串行写入目标库。如果你的源库QPS是5000单线程MySQL写入撑不住。我们调到4但立刻发现主键冲突——因为Flink按key哈希分发而同一张表的不同行可能被分到不同task导致INSERT并发写同一主键。解决方案是强制按表名分组在SQL里加DISTRIBUTE BY table_name或在DataStream API里用keyBy(r - r.getField(table_name))。sink.buffer-flush.max-rows默认1000意思是攒够1000行再批量INSERT。但若某张表流量极小如配置表每小时才1条这1000行可能永远攒不满导致数据滞留。我们设为500并加sink.buffer-flush.interval-ms 10001秒超时强制刷平衡吞吐与延迟。实操心得目标库务必关闭autocommit让Flink用JDBC事务控制。我们曾因目标库autocommitON导致部分INSERT成功部分失败事务无法回滚最终数据不一致。现在所有同步任务的目标库连接串都强制加?useSSLfalseallowPublicKeyRetrievaltrueautoCommitfalse。3.4 检查点Checkpoint策略不是设个间隔就完事Checkpoint是Flink可靠性的命脉但参数设错等于埋雷execution.checkpointing.interval 30s30秒太短MySQL binlog position变化频繁30秒checkpoint可能刚保存完position下一秒binlog就被purge了。我们设为300s5分钟并确保MySQL的expire_logs_days 7保留7天binlog。execution.checkpointing.mode EXACTLY_ONCE必须是这个AT_LEAST_ONCE会导致重复。state.backend rocksdb别用filesystemRocksDB对大状态更友好。我们同步200张表时state大小达8GBfilesystem checkpoint超时失败率100%。最关键的是state.checkpoints.dir hdfs://namenode:8020/flink/checkpoints必须指向高可用存储别用本地磁盘。我们吃过亏——JobManager挂了本地checkpoint丢失只能从头同步。4. 实操过程与核心环节实现手把手跑通第一个同步Job下面以Flink SQL方式演示这是最快上手的方式。假设你要同步source_db.orders到target_db.orders两库结构完全一致。4.1 环境准备三步到位下载依赖去 Flink官网 下载Flink 1.17.2Scala 2.12解压后进入lib/目录放入flink-sql-connector-mysql-cdc-2.4.0.jarCDC连接器mysql-connector-java-8.0.33.jarMySQL驱动注意版本必须严格匹配Flink 1.17.x对应CDC 2.4.x用2.3.x会报ClassNotFoundException。我们曾因jar包版本错配调试两天才发现是类加载器问题。启动Flink集群# 启动Standalone集群开发用 ./bin/start-cluster.sh # 或提交到YARN生产用 ./bin/flink run-application -t yarn-application ./path/to/job.jar创建Flink SQL Client配置编辑conf/sql-client-defaults.yaml添加execution: runtime-mode: streaming checkpointing: interval: 300s mode: EXACTLY_ONCE state: backend: rocksdb checkpoints.dir: file:///tmp/flink/checkpoints4.2 编写同步SQL从建表到启动打开./bin/sql-client.sh依次执行-- 1. 创建源表注意这里不指定主键CDC会自动推断 CREATE TABLE source_orders ( order_id BIGINT, user_id BIGINT, amount DECIMAL(10,2), status STRING, update_time TIMESTAMP(3), PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname source-mysql-host, port 3306, username cdc_user, password cdc_pass, database-name source_db, table-name orders, server-time-zone Asia/Shanghai, scan.startup.mode initial -- 首次全量增量后续改为latest-offset ); -- 2. 创建目标表必须和源表字段完全一致包括类型和顺序 CREATE TABLE sink_orders ( order_id BIGINT, user_id BIGINT, amount DECIMAL(10,2), status STRING, update_time TIMESTAMP(3), PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( connector jdbc, url jdbc:mysql://target-mysql-host:3306/target_db?useSSLfalseallowPublicKeyRetrievaltrueautoCommitfalse, table-name orders, username sink_user, password sink_pass, sink.buffer-flush.max-rows 500, sink.buffer-flush.interval 1s, sink.max-retries 3 ); -- 3. 执行同步核心INSERT SELECTFlink自动处理CDC逻辑 INSERT INTO sink_orders SELECT * FROM source_orders;4.3 关键参数详解与计算依据scan.startup.mode initial首次运行时Flink会先全量读取source_db.orders所有数据走SELECT *再切换到binlog监听。全量阶段耗时取决于表大小我们1亿行订单表用了23分钟。计算公式预估时间 行数 × 单行处理耗时约0.5ms。若表太大可改用earliest-offset跳过全量但需确保目标库已有初始数据。server-time-zone Asia/Shanghai必须显式设置MySQL时区和JVM时区不一致会导致timestamp字段偏移8小时。我们曾因此发现所有订单update_time比实际晚8小时排查半天才发现是这个参数漏了。sink.max-retries 3MySQL网络抖动时自动重试。别设太高否则重试期间数据积压。我们测试过3次重试覆盖99.2%的瞬时失败再高就得查网络了。4.4 监控与验证如何确认它真的在工作光看SQL执行成功不够必须验证三件事数据一致性在Flink Web UIhttp://jobmanager:8081里找到该Job点开“Metrics”重点看numRecordsInPerSecond源端每秒读取行数应和源库QPS接近numRecordsOutPerSecondSink端每秒写出行数应和输入基本一致允许5%偏差lastCheckpointSize检查点大小若持续增长超过1GB说明状态泄漏延迟监控Flink CDC暴露了source_binlog_fetch_delay指标单位ms即当前处理的binlog事件距离产生时间的延迟。我们设告警阈值为5000ms5秒超过即触发企业微信通知。终极校验写个简单脚本比对源库和目标库最新1000行-- 源库 SELECT MD5(GROUP_CONCAT(CONCAT(order_id, -, amount, -, status) ORDER BY order_id)) as checksum FROM (SELECT * FROM orders ORDER BY update_time DESC LIMIT 1000) t; -- 目标库同理两个MD5必须完全相等我们每天凌晨2点自动执行此脚本连续30天零差异才算通过。5. 常见问题与排查技巧实录那些文档里不会写的坑5.1 典型问题速查表问题现象可能原因排查命令/方法解决方案Job启动后一直RUNNING但无数据输出源库binlog未开启或权限不足mysql -u cdc_user -p -e SHOW MASTER STATUS;检查binlog_format、server_id、账号权限日志报java.lang.ClassNotFoundException: com.mysql.cj.jdbc.Drivermysql-connector-java jar未放入lib目录ls $FLINK_HOME/lib/grep mysql同步延迟飙升到分钟级Checkpoint超时或RocksDB写入慢jstat -gc pid查看GC频率调大state.backend.rocksdb.memory.managed或换SSD磁盘目标库出现主键冲突错误sink.parallelism 1且未按表分组查看Flink Web UI的TaskManager日志在INSERT前加DISTRIBUTE BY table_nameDDL变更后Job崩溃源库新增字段目标库表结构未同步DESCRIBE target_db.orders手动执行ALTER TABLE target_db.orders ADD COLUMN xxx TYPE5.2 独家避坑技巧技巧1用scan.startup.mode timestamp精准控制起点有时你需要从某个时间点开始同步比如修复数据后initial太粗暴。Flink CDC支持timestamp模式connector mysql-cdc, ... scan.startup.mode timestamp, scan.startup.timestamp-millis 1712345678000 -- 时间戳毫秒这个时间戳对应MySQL的SHOW MASTER STATUS里的Position但别手动算——用mysqlbinlog工具解析mysqlbinlog --base64-outputdecode-rows -v mysql-bin.000001 | grep -A 5 SET session.timestamp我们封装了一个Python脚本输入时间戳自动返回最近的binlog position5分钟搞定。技巧2目标库写入瓶颈的“熔断”方案当目标MySQL负载高时Flink Sink会堆积大量bufferOOM风险陡增。我们在Sink前加了一层限流INSERT INTO sink_orders SELECT * FROM ( SELECT *, CASE WHEN COUNT(*) OVER (ORDER BY proc_time ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) % 100 0 THEN SLEEP(0.01) END as _dummy FROM source_orders );虽然有点hacky但实测在目标库CPU90%时能将Flink内存占用压到安全线以下比直接OOM强十倍。技巧3跨库同步时的表名路由如果要同步source_db.*到target_db.*但表名相同会冲突。Flink SQL不支持动态表名我们用DataStream API解决stream.map(record - { String tableName record.getField(table_name); // 根据tableName路由到不同Sink if (orders.equals(tableName)) return new SinkRecord(record, target_orders); else if (users.equals(tableName)) return new SinkRecord(record, target_users); return null; }).addSink(new CustomMysqlSink());这个CustomMysqlSink内部用MapString, JdbcConnection缓存不同表的连接性能比每次new Connection高5倍。5.3 性能调优实战数据我们最终优化后的参数组合200张表峰值QPS 8000parallelism.default 8Source和Sink均8并行state.backend.rocksdb.memory.managed trueexecution.checkpointing.interval 300ssink.buffer-flush.max-rows 1000sink.buffer-flush.interval 1s结果端到端延迟P99 620msCPU使用率稳定在65%GC时间200ms/次。对比初始配置默认参数延迟降低76%资源消耗减少41%。6. 进阶场景与扩展建议同步只是起点不是终点做到这一步你已经拥有了一个工业级MySQL同步管道。但真正的价值在于它能支撑更多场景6.1 增量ETL在同步链路中嵌入业务逻辑别只做“搬运工”加个Flink SQL就能变身“加工站”。比如订单同步时自动计算用户等级CREATE VIEW enriched_orders AS SELECT *, CASE WHEN amount 10000 THEN VIP WHEN amount 1000 THEN GOLD ELSE NORMAL END as user_level FROM source_orders; INSERT INTO sink_orders SELECT * FROM enriched_orders;这样目标库拿到的就是带等级标签的订单BI报表直接用不用再JOIN用户表。6.2 多目标分发一份变更写入MySQLESHiveFlink的DataStream天然支持广播。把CDC Source的结果同时发给多个SinkDataStreamRowData sourceStream env.fromSource( MySqlSource.RowDatabuilder() .hostname(mysql-host) .tableList(source_db.orders) .build(), WatermarkStrategy.noWatermarks(), mysql-source ); // 写MySQL sourceStream.addSink(new JdbcSink(...)); // 写Elasticsearch sourceStream.addSink(new ElasticsearchSink(...)); // 写Hive用StreamingFileSink sourceStream.map(row - row.toString()).addSink(new StreamingFileSink(...));我们用这套方案实现了“订单变更→MySQL业务查询 ES搜索 Hive离线分析”三写延迟均控制在2秒内。6.3 变更审计给每一次修改打上时间戳和操作人源库通常没有操作人字段但binlog里有thread_id。Flink CDC能提取它SELECT *, mysql- || thread_id as source_thread_id, CURRENT_TIMESTAMP as sync_time FROM source_orders;再结合MySQL的performance_schema.threads表需提前开启就能关联出具体是哪个应用IP执行的变更。我们把它存进审计表现在所有数据修改都有迹可循。我个人在实际操作中的体会是Flink MySQL同步不是银弹但它把数据同步这件事从“运维黑盒”变成了“可编程管线”。你不再需要祈祷脚本别出错而是可以像写业务代码一样用SQL或Java去定义数据的流转逻辑、容错策略、监控指标。当某天业务方说“我们需要把用户表的手机号脱敏后再同步”你只需要改一行SQL而不是重写整个同步脚本——这种掌控感才是技术人最踏实的底气。