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

1. 项目概述为什么需要 Flink 做 MySQL 到 MySQL 的实时同步Flink 实现 MySQL 的数据同步MySQL Sink → MySQL本质上是在构建一套低延迟、高可靠、可容错的跨库实时数据链路。这不是简单的“把 A 表的数据拷贝到 B 表”而是解决生产环境中长期存在的几个硬骨头问题比如业务库和分析库物理隔离后报表系统查不到最新订单状态比如微服务拆分后用户中心变更了手机号但订单服务还在用旧值再比如风控系统需要毫秒级感知用户账户余额变动但传统定时任务拉取 Binlog 或全量导出延迟动辄几分钟甚至小时——这些场景下“准实时”已经不够用了必须是“真实时”。我最早在某电商中台项目里遇到这个问题订单库MySQL 8.0和履约库另一套 MySQL 实例之间靠每天凌晨跑一次 ETL 同步结果大促期间用户下单成功后物流侧查不到运单号客服电话被打爆。后来我们试过 Canal Kafka 自研消费者但运维成本高、Exactly-Once 语义难保障、字段映射改一次就要发版。最终落地 Flink CDC JDBC Sink 方案端到端延迟压到 800ms 以内且支持断点续传、字段过滤、类型转换、冲突更新等真实业务需求。关键词“Flink MySQL 同步”“MySQL Sink”“CDC 实时同步”之所以持续热正是因为大量团队正从“能用”走向“稳用”“好用”——而这个过程绕不开对 Flink CDC 连接器原理、JDBC Sink 配置细节、MySQL 权限与 binlog 设置、以及常见数据不一致根源的深度理解。这篇文章不是讲“怎么跑通一个 demo”而是还原一个资深开发者从零搭建、调优、上线、兜底的完整闭环。你会看到为什么选 Flink CDC 而不是 Debezium为什么 Sink 不直接用 Table API 而要手写 ProcessFunctionMySQL 主键缺失时怎么避免脏数据当目标表有唯一索引但非主键Flink 怎么做 Upsert这些都不是文档里一句话带过的点而是线上踩坑后必须补上的课。适合正在评估实时同步方案的架构师、负责数据管道落地的开发同学以及想搞懂 Flink 真实工业级用法的进阶学习者。2. 整体设计思路与方案选型逻辑2.1 为什么不用“导出 SQL source 导入”这种简单方式很多新手第一反应是“我 mysqldump 出来再 mysql -e 导进去不就行了”——这在测试环境可能跑得通但在生产环境等于埋雷。原因有三第一无增量能力。mysqldump 是快照式导出哪怕加 --where 参数也只是一次性切片无法持续捕获变更。而真实业务每秒都有成百上千条订单、支付、库存记录产生停服导出根本不现实。第二无事务边界保障。MySQL 的 InnoDB 引擎支持行级锁和 MVCC但 dump 导出过程会持有读锁尤其大表影响线上交易。更关键的是dump 文件本身不包含事务提交顺序信息导入时多个 INSERT 可能被重排序导致“先删后增”变成“先增后删”数据逻辑错乱。第三无错误隔离与重试机制。一条 SQL 执行失败比如字段长度超限、时间格式不匹配整个导入就中断人工介入成本高且无法定位到具体哪一行出错。而 Flink 的 checkpoint state backend 机制天然支持失败后从最近一次检查点恢复跳过已处理数据重试失败批次。所以真正的实时同步必须基于Change Data CaptureCDC——即捕获数据库的变更日志binlog而不是查询当前快照。这是所有现代实时数仓、数据湖同步方案的底层共识。2.2 Flink CDC vs Debezium为什么选前者目前主流 CDC 方案有两类独立部署的中间件如 Debezium Kafka和 Flink 原生 CDC 连接器。我们对比过两种路径维度Debezium Kafka FlinkFlink CDC1.16架构复杂度需维护 Kafka 集群、ZooKeeper、Debezium Connector 三个组件配置项超 50单 Flink 作业即可无需外部消息队列配置集中在 Flink SQL 或 DataStream API 中延迟表现Kafka 生产/消费引入额外网络跳转端到端 P99 延迟通常 ≥ 1.2s直连 MySQL binlog无中间缓冲实测 P99 ≤ 800ms千兆内网Exactly-Once 保障依赖 Kafka 的幂等 Producer Flink 的两阶段提交链路长故障点增多Flink CDC 内置基于 checkpoint 的 binlog 位点保存与 JDBC Sink 的两阶段提交深度集成一致性更强运维成本每个 MySQL 实例需单独配置 Connector扩容需调整 Kafka 分区、Consumer Group所有同步任务统一由 Flink JobManager 管理启停、扩缩容、监控全部收敛到 Flink Web UI我们最终选择 Flink CDC核心原因是降低链路熵值。在某金融客户现场他们曾用 Debezium 方案结果 Kafka 磁盘满导致 binlog 积压Flink 任务重启后因 offset 丢失重复消费造成下游账务表多扣款。而 Flink CDC 将 binlog 位点直接存在 Flink 的 RocksDB State 中只要 checkpoint 成功位点就安全彻底规避了外部组件故障带来的数据风险。提示Flink CDC 1.16 开始支持 MySQL 8.0.23 的 GTID 模式比传统 file position 更可靠。但注意GTID 要求 MySQL 开启 gtid_modeON 且 enforce_gtid_consistencyON这两个参数修改后需重启 MySQL务必提前评估。2.3 Sink 层为什么弃用 Table API 的jdbcconnector而采用JDBCOutputFormat或自定义ProcessFunctionFlink 官方文档里Table API 的 JDBC 连接器写法最简洁CREATE TABLE mysql_sink ( id BIGINT, name STRING, amount DECIMAL(10,2), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector jdbc, url jdbc:mysql://target:3306/db, table-name t_order, username user, password pwd );但我们在压测中发现两个致命缺陷第一Upsert 语义不可控。当目标表有唯一索引非主键时Table API 的INSERT INTO ... ON DUPLICATE KEY UPDATE会把所有字段都更新哪怕你只想同步status字段它也会把create_time、update_time全部覆盖破坏业务时间戳逻辑。第二批量写入策略僵化。Table API 默认每 1000 条触发一次 batch insert但实际业务中有些表字段多30列、单行体积大含 JSON 字段1000 行 batch 可能超 MySQL max_allowed_packet默认 4MB直接报错Packet for query is too large。而自定义JDBCOutputFormat可以动态计算 batch size比如按总字节数截断而非固定行数。所以我们最终采用 DataStream API JDBCConnectionOptionsJDBCStatementBuilder的组合完全掌控 SQL 生成逻辑。例如针对“只更新 status 和 update_time”的需求我们手写new JDBCStatementBuilderRowData() { Override public void accept(PreparedStatement ps, RowData row) throws SQLException { ps.setString(1, row.getString(2).toString()); // status ps.setTimestamp(2, Timestamp.from(Instant.now())); // update_time ps.setLong(3, row.getLong(0)); // id for WHERE } }这样SQL 永远是UPDATE t_order SET status ?, update_time ? WHERE id ?干净、精准、可控。3. 核心细节解析与实操要点3.1 MySQL 源端必备配置不只是开 binlog很多人以为“开了 binlog 就能用 CDC”其实远远不够。Flink CDC 对 MySQL 的要求是生产级严格配置漏掉任意一项都可能导致任务启动失败或数据丢失。首先binlog 格式必须为 ROW。这是硬性前提因为 Flink CDC 解析的是 binlog event 中的行变更内容WriteRowsEvent/UpdateRowsEvent/DeleteRowsEvent而非 SQL 语句STATEMENT或混合模式MIXED。验证命令SHOW VARIABLES LIKE binlog_format; -- 必须返回 ROW如果返回 STATEMENT需在 my.cnf 中添加[mysqld] binlog-formatROW然后重启 MySQL。注意修改后所有新事务才生效旧 binlog 仍为 STATEMENT 格式Flink CDC 会跳过它们。其次server-id 必须全局唯一且非 1。MySQL 主从复制中server-id 是节点身份标识。Flink CDC 作为“伪从库”连接 MySQL必须声明自己的 server-id否则 MySQL 会拒绝连接报错ERROR 1236 (HY000): Could not find first log file name in binary log index file。设置方法SET GLOBAL server_id 1001; -- 不能为 1建议 1000 避免与真实从库冲突或者在 my.cnf 中[mysqld] server-id1001再次binlog 保留时间必须足够长。Flink CDC 启动时会从指定位点或最新位点开始读取但如果位点对应的 binlog 文件已被 MySQL 清理expire_logs_days默认 0即永不清除但磁盘满时 MySQL 会自动 purge任务就会卡住。我们线上统一设为 7 天SET GLOBAL expire_logs_days 7;最后创建专用同步账号并授权。绝对禁止用 root 或业务账号。最小权限原则如下CREATE USER flink_sync% IDENTIFIED BY StrongPass123!; GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO flink_sync%; FLUSH PRIVILEGES;解释每个权限的作用SELECT用于初始化全量快照snapshotRELOAD执行 FLUSH TABLES WITH READ LOCK快照阶段加全局读锁SHOW DATABASES列出所有数据库用于白名单过滤REPLICATION SLAVE允许作为从库连接读取 binlogREPLICATION CLIENT执行 SHOW MASTER STATUS 查看 binlog 位置注意MySQL 8.0 默认认证插件为 caching_sha2_password而 Flink CDC 1.16 使用的 MySQL Connector/J 8.0.28 不支持该插件。解决方案有两个一是降级账号认证方式ALTER USER flink_sync% IDENTIFIED WITH mysql_native_password BY StrongPass123!;二是升级 Flink CDC 到 1.17已内置兼容。3.2 Flink CDC 连接器核心参数详解不只是填 URLFlink CDC 的MySqlSource构建看似简单但每个参数背后都是血泪教训。我们逐个拆解生产环境必配项MySqlSourceString source MySqlSource.Stringbuilder() .hostname(mysql-source) // 源 MySQL 地址 .port(3306) .databaseList(order_db) // 白名单只同步指定库 .tableList(order_db.t_order, order_db.t_payment) // 白名单精确到表 .username(flink_sync) .password(StrongPass123!) .serverId(1001-1005) // server-id 范围防止单点故障 .serverTimeZone(Asia/Shanghai) // 关键时区不一致会导致时间字段错乱 .deserializer(new StringDeserializer()) // 序列化器决定输出格式 .startupOptions(StartupOptions.latest()) // 启动策略 .build();serverId(1001-1005)为什么是范围因为 Flink CDC 任务可能被调度到不同 TaskManager每个实例需独立 server-id。范围保证不冲突且当某个实例宕机新实例可从中选取未被占用的 id。实测中我们设为1001-101010 个 slot 足够。serverTimeZone(Asia/Shanghai)这是最容易被忽略的坑。MySQL 的TIMESTAMP类型存储的是 UTC 时间但显示时会根据 session time_zone 转换。如果 Flink 作业 JVM 时区是 UTC而 MySQL 是 CST那么2023-10-01 12:00:00在 Flink 里会变成2023-10-01 04:00:00。必须显式对齐startupOptions四种策略必须吃透initial()先全量再增量默认适合首次同步但全量阶段锁表。earliest()从最早 binlog 开始适合已有历史数据需补录。latest()从最新位点开始只同步后续变更适合测试或临时链路。specificOffset(mysql-bin.000001, 12345)指定文件和位点用于故障恢复。我们线上全部用initial()但通过scan.incremental.snapshot.enabledtrueFlink CDC 1.16开启增量快照避免全量阶段锁表。原理是将大表按主键分片每个 subtask 并行读取一个分片全程无锁速度提升 5~10 倍。deserializer推荐用JsonDebeziumDeserializationSchema而非StringDeserializer。前者输出结构化 JSON包含op操作类型c/u/d/r、ts_ms事件时间、source源库表信息等元数据方便后续做路由、过滤、打标。例如{ op: u, ts_ms: 1696123456789, source: {db: order_db, table: t_order}, after: {id: 1001, status: shipped, update_time: 2023-10-01 12:00:00} }3.3 JDBC Sink 的健壮性设计从“能写”到“写对”Sink 层是数据落地的最后一环也是故障高发区。我们总结出三大必做动作第一连接池必须用 HikariCP且配置合理。Flink 官方 JDBC connector 默认用 DBCP但 DBCP 2.x 已停止维护且连接泄漏问题频发。HikariCP 是当前 Java 生态事实标准。关键参数HikariConfig config new HikariConfig(); config.setJdbcUrl(jdbc:mysql://target:3306/db?useSSLfalseserverTimezoneAsia/Shanghai); config.setUsername(sink_user); config.setPassword(pwd); config.setMaximumPoolSize(20); // 根据目标 MySQL 最大连接数设定通常 ≤ 50 config.setMinimumIdle(5); config.setConnectionTimeout(30000); config.setIdleTimeout(600000); config.setMaxLifetime(1800000); config.setLeakDetectionThreshold(60000); // 开启连接泄漏检测提示maxLifetime设为 30 分钟强制刷新连接避免 MySQL 的wait_timeout默认 8 小时导致连接空闲超时被服务端 killFlink 报Connection reset。第二SQL 批量写入必须带rewriteBatchedStatementstrue。MySQL JDBC 驱动默认将 batch insert 拆成多条独立 SQL 发送网络开销巨大。加上此参数后驱动会重写为INSERT INTO t VALUES (?),(?),(?)形式吞吐量提升 3~5 倍。URL 示例jdbc:mysql://target:3306/db?useSSLfalseserverTimezoneAsia/ShanghairewriteBatchedStatementstrue第三Upsert 必须区分“主键冲突”和“唯一索引冲突”。MySQL 的INSERT ... ON DUPLICATE KEY UPDATE中“KEY” 指主键PRIMARY KEY或唯一索引UNIQUE INDEX。但 Flink 的JDBCOutputFormat默认只认主键。如果目标表只有唯一索引如UNIQUE KEY uk_order_no (order_no)必须手动指定冲突字段JDBCConnectionOptions options new JDBCConnectionOptions.JDBCConnectionOptionsBuilder() .withUrl(jdbc:mysql://...) .withDriverName(com.mysql.cj.jdbc.Driver) .withUsername(user) .withPassword(pwd) .build(); JDBCOutputFormat.Builder builder JDBCOutputFormat.buildJDBCOutputFormat() .setDBUrl(jdbc:mysql://...) .setUsername(user) .setPassword(pwd) .setQuery(INSERT INTO t_order (id, order_no, status) VALUES (?,?,?) ON DUPLICATE KEY UPDATE status VALUES(status), update_time NOW()) .setSqlTypes(new int[]{Types.BIGINT, Types.VARCHAR, Types.VARCHAR});注意setQuery中的ON DUPLICATE KEY UPDATE—— 这里的 “KEY” 就是order_no字段的唯一索引。Flink 不解析表结构一切由你写的 SQL 决定。4. 实操过程与核心环节实现4.1 从零搭建Flink 1.17 MySQL 8.0.33 完整流程我们以某零售 SaaS 系统为例同步sales_db.t_transaction交易表到dw_db.t_transaction_dwd数仓明细表。环境Flink 1.17.1Standalone 模式MySQL 8.0.33源/目标均为 Docker 容器。步骤 1准备 MySQL 源库# 启动源 MySQL docker run -d \ --name mysql-source \ -p 3306:3306 \ -e MYSQL_ROOT_PASSWORDroot123 \ -v $(pwd)/my-source.cnf:/etc/mysql/conf.d/my.cnf \ -d mysql:8.0.33my-source.cnf内容[mysqld] server-id1001 log-binmysql-bin binlog-formatROW expire_logs_days7 gtid_modeON enforce_gtid_consistencyON步骤 2创建同步账号并授权CREATE USER flink_sync% IDENTIFIED WITH mysql_native_password BY SyncPass2023!; GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO flink_sync%; FLUSH PRIVILEGES;步骤 3Flink 作业代码DataStream APIpublic class MysqlToMysqlSync { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(10_000); // 10秒 checkpoint 间隔 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000); env.getCheckpointConfig().setCheckpointTimeout(60000); env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointingConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 1. 创建 MySQL CDC Source MySqlSourceString source MySqlSource.Stringbuilder() .hostname(mysql-source) .port(3306) .databaseList(sales_db) .tableList(sales_db.t_transaction) .username(flink_sync) .password(SyncPass2023!) .serverId(1001-1005) .serverTimeZone(Asia/Shanghai) .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); DataStreamString stream env.fromSource(source, WatermarkStrategy.noWatermarks(), MySQL-CDC-Source); // 2. 解析 JSON提取字段并转换 DataStreamRowData parsedStream stream.map(new MapFunctionString, RowData() { Override public RowData map(String value) throws Exception { JSONObject json new JSONObject(value); String op json.optString(op); // c/u/d/r if (d.equals(op)) return null; // 删除操作暂不处理 JSONObject after json.optJSONObject(after); if (after null) return null; GenericRowData row new GenericRowData(4); row.setField(0, after.optLong(id)); // BIGINT row.setField(1, after.optString(order_no)); // VARCHAR row.setField(2, after.optString(status)); // VARCHAR row.setField(3, Timestamp.from(Instant.now())); // update_time return row; } }); // 3. JDBC Sink 配置 JDBCConnectionOptions connectionOptions new JDBCConnectionOptions.JDBCConnectionOptionsBuilder() .withUrl(jdbc:mysql://mysql-target:3306/dw_db?useSSLfalseserverTimezoneAsia/ShanghairewriteBatchedStatementstrue) .withDriverName(com.mysql.cj.jdbc.Driver) .withUsername(sink_user) .withPassword(SinkPass2023!) .build(); JDBCOutputFormat outputFormat JDBCOutputFormat.buildJDBCOutputFormat() .setDBUrl(jdbc:mysql://mysql-target:3306/dw_db?useSSLfalseserverTimezoneAsia/ShanghairewriteBatchedStatementstrue) .setUsername(sink_user) .setPassword(SinkPass2023!) .setQuery(INSERT INTO t_transaction_dwd (id, order_no, status, update_time) VALUES (?,?,?,?) ON DUPLICATE KEY UPDATE status VALUES(status), update_time VALUES(update_time)) .setSqlTypes(new int[]{Types.BIGINT, Types.VARCHAR, Types.VARCHAR, Types.TIMESTAMP}) .build(); parsedStream.output(outputFormat); env.execute(MySQL-to-MySQL Sync Job); } }步骤 4打包与提交# 编译pom.xml 需包含 flink-cdc-connectors、mysql-connector-java 8.0.33 mvn clean package -DskipTests # 提交到 Flink Standalone ./flink run \ -c com.example.MysqlToMysqlSync \ ./target/flink-mysql-sync-1.0.jar步骤 5验证效果查看 Flink Web UIJobManager 页面确认任务 RUNNINGTaskManager 显示 4 个 subtask默认并行度 4源库执行更新UPDATE sales_db.t_transaction SET status shipped WHERE id 1001;目标库查询SELECT * FROM dw_db.t_transaction_dwd WHERE id 1001;—— status 和 update_time 应实时更新查看日志tail -f /opt/flink/log/flink-*-jobmanager-*.out确认无SQLException或TimeoutException4.2 性能调优如何把延迟压到 500ms 以内线上压测时我们用 sysbench 模拟每秒 2000 条 UPDATE初始延迟达 3.2s。通过以下五步优化P99 延迟降至 480ms① 调整 checkpoint 间隔与超时初始checkpoint 间隔 30s超时 120s → 太长导致状态积压优化间隔 10s超时 60smin-pause 5s → 更频繁地切分状态减少单次 checkpoint 数据量② 增加并行度并绑定 CPU初始parallelism2TaskManager 仅 2 核 → CPU 成瓶颈优化parallelism8TaskManager 分配 8 核taskmanager.numberOfTaskSlots: 8→ 充分利用多核解析 binlog event③ 启用增量快照Incremental Snapshot在MySqlSource.builder()中添加.option(scan.incremental.snapshot.enabled, true) .option(scan.incremental.snapshot.chunk.size, 10000) // 每个分片 1 万行效果全量阶段从 12 分钟降至 98 秒且无锁表④ JDBC Sink 批处理调优初始batch size 固定 1000 → 大字段表易超 packet优化改用JDBCAsyncOutputFormat并设置maxBatchSize500maxBatchIntervalMs1010ms 强制 flush原因异步写入避免阻塞主线程小 batch 短间隔平衡吞吐与延迟⑤ MySQL 目标库参数调优关键参数SET GLOBAL innodb_buffer_pool_size 2G; -- 至少 50% 物理内存 SET GLOBAL innodb_log_file_size 512M; -- 加快 redo log 写入 SET GLOBAL sync_binlog 0; -- Sink 不需 binlog关闭 SET GLOBAL innodb_flush_log_at_trx_commit 2; -- 每秒刷一次 log非每次事务实测数据优化后2000 QPS 下95% 请求延迟 400ms99% 480msCPU 使用率稳定在 65%~75%无 GC 频繁现象。5. 常见问题与排查技巧实录5.1 典型问题速查表问题现象根本原因排查命令/方法解决方案Could not find first log file name in binary log index fileMySQL 未开启 binlog或 server-id 未设置SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE server_id;检查 my.cnf确保log-bin和server-id存在且重启生效Failed to deserialize binlog eventMySQL 时区与 Flink JVM 时区不一致导致 timestamp 解析失败SELECT global.time_zone, session.time_zone;对比 Flink 机器date命令在MySqlSource.builder()中显式设置.serverTimeZone(Asia/Shanghai)Packet for query is too largeJDBC batch insert 超 MySQLmax_allowed_packetSHOW VARIABLES LIKE max_allowed_packet;在 JDBC URL 中添加maxAllowedPacket6710886464MB或减小 batch sizeDuplicate entry xxx for key PRIMARY目标表主键冲突但 Flink 未配置 Upsert查看 Sink SQL 是否含ON DUPLICATE KEY UPDATE改用INSERT ... ON DUPLICATE KEY UPDATE或确保源数据主键唯一Checkpoint failed due to state size exceeded单次 checkpoint 状态过大如大表全量未分片flink list -a查看 job idflink cancel jobid后查看日志启用增量快照.option(scan.incremental.snapshot.enabled, true)5.2 独家避坑技巧那些文档没写的细节技巧 1如何安全地修改同步字段映射业务常提需求“t_transaction 表新增 discount_amount 字段也要同步过去”。直接改代码会丢数据。正确做法步骤一暂停 Flink 作业Savepoint步骤二在目标表执行ALTER TABLE t_transaction_dwd ADD COLUMN discount_amount DECIMAL(10,2) DEFAULT 0;步骤三修改 Flink 代码在map()函数中增加字段提取逻辑步骤四从 Savepoint 恢复作业flink run -s savepoint-path关键Savepoint 必须在ALTER TABLE之后、作业恢复之前创建否则恢复时会因字段数不匹配报ArrayIndexOutOfBoundsException。技巧 2如何处理 MySQL 的TINYINT(1)布尔类型MySQL 中TINYINT(1)常被用作布尔0/1但 Flink CDC 默认解析为Integer下游 Java 代码需手动转Boolean。更优雅的方式是在MySqlSource中添加debezium.properties.option(debezium.database.history.skip.unavailable.databases, true) .option(debezium.transforms, unwrap) .option(debezium.transforms.unwrap.type, io.debezium.transforms.ExtractNewRecordState) .option(debezium.transforms.unwrap.delete.handling.mode, rewrite) // 关键将 TINYINT(1) 映射为 BOOLEAN .option(debezium.column.propagate.source.type, tinyint:BOOLEAN)技巧 3当源表无主键时如何避免重复数据Flink CDC 要求表有主键才能做增量快照分片。若遇到无主键老表如日志表必须方案 A推荐添加自增主键ALTER TABLE t_log ADD COLUMN id BIGINT AUTO_INCREMENT PRIMARY KEY FIRST;方案 B用scan.snapshot.fetch.size1000强制小批量读取但性能差且无法保证 exactly-once技巧 4如何监控同步延迟Flink Web UI 只显示 checkpoint 延迟不反映业务延迟。我们加了一层监控在map()函数中提取ts_msbinlog 事件时间戳和System.currentTimeMillis()处理时间计算差值processDelay System.currentTimeMillis() - ts_ms用MetricGroup.gauge()上报Grafana 配置告警processDelay 50005秒触发企业微信通知这个ts_ms是 MySQL 写入 binlog 的真实时间比 Flink 的processingTime更准确反映端到端延迟。5.3 故障恢复实战一次 binlog 被误删后的抢救某天运维误执行PURGE BINARY LOGS BEFORE 2023-09-01 00:00:00;导致 Flink CDC 任务报错Could not find specified binlog file。当时 checkpoint 保存在 HDFS最新位点指向mysql-bin.000015但该文件已被删。抢救步骤登录 MySQL执行SHOW BINARY LOGS;查看现存最早文件mysql-bin.000016从 Flink Web UI 获取失败 task 的详细日志找到报错堆栈中的binlog filename和position手动修改 Savepoint 中的offset文件JSON 格式将filename:mysql-bin.000015改为filename:mysql-bin.000016position:12345改为position:4每个 binlog 文件起始 position 为 4用flink run -s modified-savepoint-path恢复作业注意此操作有丢数据风险000015文件中的变更但比重置位点latest()更可控。后续我们加了巡检脚本每小时检查SHOW MASTER STATUS与 HDFS 中保存的位点是否连续。6. 扩展思考这个方案还能怎么用Flink MySQL 同步绝不仅限于“库到库”。在我参与的某物联网平台项目中我们把它扩展成了一个通用数据分发中枢场景一多目标分发一个t_device_event表同时同步到✓ MySQL 数仓供 BI 报表✓ Elasticsearch供 Kibana 实时检索✓ Kafka供风控引擎消费实现方式DataStream.broadcast()后接多个 Sink每个 Sink 独立配置。场景二轻量 ETL同步过程中做清洗过滤测试数据WHERE env ! prod、脱敏手机号phone CONCAT(LEFT(phone,3), ****, RIGHT(phone,4))、丰富维度关联t_user表查城市。全部在map()中完成无需额外 Spark 作业。场景三变更审计将 CDC 输出的 JSON含op,before,after,ts_ms直接写入t_audit_log表形成完整操作日志。某次资损事故中我们靠这条日志 5 分钟内定位