接到一个活儿把业务库的MySQL数据实时同步到Elasticsearch给搜索和报表用。一开始我想得挺简单直接定时任务全量拉一遍不就行了结果一上线就被延迟数据打脸用户这边刚下完单搜索那边半小时后才看得到。后来换了Flink CDC这条链路才真正把同步这件事做到分钟级以内。这篇就顺着这个实战过程把Flink CDC从选型到落地的完整细节拆开讲包括binlog拉取原理、环境版本搭配、任务开发、以及生产环境里那些迟早会踩的坑希望能给打算做MySQL到ES同步的开发者省点试错时间。1. 同步MySQL到ES为什么最终选Flink CDC而不是传统ETL1.1 业务痛点全量同步的延迟代价很多团队一开始做数据同步脑子里第一反应就是写个XXXL每半小时全量同步一次。这个方案确实简单MySQL里SELECT全部数据落到临时文件再灌进Elasticsearch。但问题很明显首先全量同步的耗时跟数据量成正比几十G的表每次跑完都要几分钟甚至十几分钟其次为了不锁业务表通常要用mysqldump或者游标分批读这中间新产生的增量数据根本抓不到。结果就是业务系统里面的最新状态ES里面永远差一个批次。我当时那个场景是电商订单搜索用户下单后需要在订单列表里立刻看到新订单也要能按商品名搜到。领导要求延迟不超过1分钟。全量同步半小时才跑一次延迟是10分钟到20分钟。后来加了个双写方案业务代码里同时写MySQL和ES但业务系统里所有改动都要动底层DAO层风险大、侵入高还容易掉消息。试了几天直接否了。1.2 常见方案比较定时同步、双写、监听binlog与Flink CDC既然要低延迟、低侵入核心思路就落到监听MySQL的binlog上。MySQL里所有增删改都会记录到binlog只要能从binlog里拿到这些变更事件再格式化成ES需要的JSON结构同步就成立了。这个链路里有几个经典实现路线方案实现方式延迟侵入性适用规模定时全量同步Quartz定时任务分钟级~小时级低小数据量对实时性无要求业务代码双写在DAO层同步写ES毫秒级高业务简单且团队能接受改动解析binlog自定义工具自己写客户端订阅binlog秒级低需要维护一套解析分发逻辑Canal Kafka 消费通过Canal伪装从库拉取binlog秒级低已经有Kafka基础设施Flink CDC直接内建binlog订阅与状态流转秒级低数据量大、需要精确一次语义在这几个方案里Flink CDC的亮点在于它把订阅binlog和流式计算两个事合并成了一个引擎。你不用自己维护一个解析进程再把事件丢到消息队列再写另一个消费者去写ES。Flink CDC连接器会把MySQL的binlog变成DataStream或动态表你只需要声明Source和Sink中间还能用SQL做字段映射、类型转换、维表关联。对做实时数仓的团队来说这是一套标准化程度很高的模式。顺带说一下为什么当时没用CanalKafka。Canal本身成熟但你得额外部署一套Canal集群还要管理Kafka的Topic、消费位点、offset重置逻辑。Flink CDC天然有Checkpoint和状态存储故障恢复时binlog位点会记录在状态里能做到精确一次语义或至少是至少一次省掉了大量手工运维成本。对于我这种后端团队、没有专职大数据运维的环境来说Flink CDC是性价比最合适的。2. Flink CDC的运作机制从binlog到ES的完整链路2.1 核心组件与角色分工Flink CDC能做什么本质上取决于它对MySQL binlog的订阅能力。整条链路里角色是这样的MySQL主库负责产生binlogFlink CDC的Source连接器会伪装成一个MySQL从库slave向主库请求binlog流Flink的任务算子负责解析binlog日志事件并将其转换成RowData结构最后通过Sink连接器写入Elasticsearch。这样伪装从库的设计有个好处完全走MySQL主从复制的标准协议不需要侵入业务库的插件或agent。你不需要改业务代码也不需要在MySQL上装任何额外的东西只要给同步账号开好复制权限就行。Flink CDC连接器内部其实还包含多个层次比如连接池负责与MySQL通信快照器snapshotter负责做初始全量数据的一致性快照增量读取器binlogReader负责持续拉取增量。在Flink的框架里这两个阶段被封装成了无边界的流全量快照完成之后自动无缝切换到增量读取用户感知不到中间的分界。2.2 基于binlog的增量抓取原理要搞懂Flink CDC不能绕开binlog里面到底存了什么。MySQL的binlog里以事件event的形式记录着所有数据变更比如TABLE_MAP事件记录了表结构信息WRITE_ROWS_EVENT表示插入UPDATE_ROWS_EVENT表示更新DELETE_ROWS_EVENT表示删除。Flink CDC连接器订阅到这些事件后把它们解析成带有模式信息schema的日志流。每行变更数据里不仅有变更后的值还有变更前后的镜像before/after。比如一条UPDATE事件它既会带更新前的旧值也会带更新后的新值。这个特性非常关键如果你只想把最新状态同步到ES可以取after字段如果你要做数据审计或者变化流的二次加工可以同时保留before字段。在Flink SQL里面你甚至可以直接用正则表达式去处理CDC关联的Time字段、用户ID、状态值。我当时就靠这个做了个只同步状态为已支付或已发货的条件过滤不需要在业务代码里加任何东西。当然binlog也有三种格式STATEMENT记录原始SQL、ROW记录行的实际变更、MIXED混合模式。做CDC同步一定要用ROW格式因为只有ROW格式才能拿到哪一行哪一列变了的完整数据。如果业务库还在用STATEMENT格式Flink CDC会直接拒绝启动或者只能靠旁路解析SQL才能恢复变更数据。排查这个问题的过程现在想起来都是一把辛酸泪。2.3 状态管理与Checkpoint的配合Flink CDC跟普通的ETL工具最大的不同在于它把同步进度作为状态保存在Flink的Checkpoint里。每处理一批binlog记录Flink会定期触发一个Checkpoint把当前消费到的binlog文件名位置offset保存到远端存储HDFS、S3或本地这样只要任务够稳重启后就能从Checkpoint里恢复位点不丢数据、不重复消费。这里有个细节需要理解binlog位置是消费进度而不是数据库事务完整进度。如果在处理一个事务的半途触发了Checkpoint恢复时可能会从该事务的中间位置重新开始造成部分重复。因此生产环境里通常会把Checkpoint之间的间隔设置得比单事务处理耗时稍大以降低出现半事务的窗口。配合MySQL主键去做ES文档更新的幂等性即使偶发重复最后也能收敛到一致状态。另外你还需要知道Flink CDC在恢复时的行为它通过StateBackend里保存的偏移量向MySQL请求从那个位点开始的binlog。如果binlog已经被MySQL的expire_logs_days策略清掉了任务就无法恢复只能重新做一次全量快照。所以这个参数必须调大至少要覆盖你允许的最长宕机时间。3. 实战前的选型与环境准备版本搭配最容易被忽视3.1 Flink、CDC connector、Java的一整套兼容矩阵版本问题在Flink CDC项目里是新手最容易踩的第一道门。某次我照着老教程写了一个Flink SQL作业本地跑得好好的换到集群环境就各种NoSuchMethodError查来查去才发现是Flink CDC的jar包和Flink核心版本对不上。以我当时用的Flink 1.15.2为例那会Flink CDC还主要依赖flink-cdc-connector-mysql。后来到了Flink 1.17、1.18时代CDC connector已经拆分成多个独立连接器包而且引入了增量快照incremental snapshot特性。所以你在pom里加依赖时一定要核对flink-connector版本、flink版本、Java版本三者properties flink.version1.17.2/flink.version cdc.version2.4.2/cdc.version /properties dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-java/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version /dependency !-- Flink SQL -- dependency groupIdorg.apache.flink/groupId artifactIdflink-table-api-java-bridge/artifactId version${flink.version}/version /dependency !-- MySQL CDC Connector -- dependency groupIdcom.ververica/groupId artifactIdflink-connector-mysql-cdc/artifactId version${cdc.version}/version /dependency !-- Elasticsearch connector -- dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-elasticsearch7/artifactId version${flink.version}/version /dependency /propertiesJava版本也要统一。Flink 1.17及以上建议用Java 8或Java 11JDK 17在某些ConcurrentHashMap的反射调用上会报错。ES客户端也分7.x和8.x两套写法差别很大8.x的RestHighLevelClient已经废弃需要走Elasticsearch Java API Client。我建议一开始就用Flink官方提供的专用连接器而不是自己在Sink里写HTTP客户端原因是连接器里面已经内置了BulkProcessor、重试、背压逻辑自己写的话性能问题很多。3.2 MySQL端必备的参数配置回收binlog时间等MySQL要同时作为CDC的数据源有些参数必须检查否则麻烦不断。第一个是binlog_formatROW像前面说的不设成ROW就解析不到准确数据第二个是binlog_row_imageFULL确保binlog里包含完整的before/after字段如果设置成MINIMAL更新事件里可能只有主键和发生变化的列ES侧做部分更新时会缺数据。第三个是binlog的保存周期。MySQL默认的expire_logs_days可能是7天甚至3天如果Flink任务因为业务改动、网络中断重启超过这段时间旧binlog就没了。官方建议设置到14天或者更长我用的时候直接设成30天毕竟磁盘多花点钱比任务无法恢复要划算。设置方法[mysqld] log-binmysql-bin binlog_formatROW binlog_row_imageFULL server_id101 expire_logs_days30这里有一个非常容易忽略的点server_id必须唯一。Flink CDC启动时会假装成一个MySQL从库如果其中一台机器上的server_id跟其他从库或另一条Flink任务重复MySQL主库会踢掉之前的连接导致同步中断。同时同一台机器上如果起多个Flink CDC作业每个作业也需要分配不同的server_id。账号权限方面同步账号至少要REPLICATION SLAVE、REPLICATION CLIENT。Web或业务账号如果权限只到SELECT是连不上binlog协议的CREATE USER cdc_user% IDENTIFIED BY StrongPassword; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO cdc_user%;3.3 ES索引设计时的字段映射考量ES这边的准备少有人提前考虑但等到走了几步才发现问题就晚了。一个常见坑是MySQL里的datetime会被自动映射成ES里的text导致无法做范围查询另一个坑是字符串字段默认被分词导致你本来想精确匹配的值被拆得乱七八糟。Flink CDC默认会根据MySQL的字段类型推断ES的字段类型。比如MySQL的bigint映射到ES是longvarchar映射到textdatetime映射到date。要注意的是MySQL的tinyint(1)常被用来表示布尔值但Flink CDC会把它当作smallintES那边就会变成long跟Java里Boolean的预期不一致。你要么在MySQL侧改字段类型要么在SQL里显式CAST。ES索引的mapping建议预先定义好不要全让Flink自动推断。比如{ mappings: { properties: { id: { type: keyword }, order_no: { type: keyword }, user_name: { type: keyword }, product_desc: { type: text, analyzer: ik_max_word }, order_amount: { type: double }, order_time: { type: date, format: yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis }, status: { type: keyword } } } }这样你在SQL里写字段映射时就能精确控制哪些字段需要分词、哪些字段精确匹配。我遇到过真实案例默认mapping把手机号这个字段变成了text用户搜索138居然能把一堆号码都匹配出来数据看起来就很不靠谱。所以ES侧一定要自己维护mapping而不是依赖自动推断。4. 第一个同步任务开发、部署与验证的完整过程4.1 写一个最小可运行的Flink SQL作业Flink CDC最大的便利是你甚至不用写Java代码就能完成同步。用Flink SQL可以在一分钟之内定义一个从MySQL到ES的管道。比如把orders表同步到一个ES索引先定义源表CREATE TABLE orders_source ( id STRING, order_no STRING, user_id STRING, product_name STRING, order_amount DECIMAL(10, 2), order_time TIMESTAMP(3), status STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname 192.168.1.10, port 3306, username cdc_user, password StrongPassword, database-name shop_db, table-name orders, server-id 5400-5404, scan.incremental.snapshot.enabled true, scan.incremental.snapshot.chunk.size 8096 );然后定义ES的Sink表。这里要注意Elasticsearch连接器是通过主键来处理更新的所以ES文档的_id要和MySQL主键对应上CREATE TABLE es_sink ( id STRING, order_no STRING, user_id STRING, product_name STRING, order_amount DECIMAL(10, 2), order_time TIMESTAMP(3), status STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector elasticsearch-7, hosts http://192.168.1.20:9200, index orders_index, document-id.key-delimiter -, sink.bulk-flush.max-actions 500, sink.bulk-flush.max-size 5mb, sink.bulk-flush.interval 1000 );最后一条insert语句就完成了同步逻辑INSERT INTO es_sink SELECT id, order_no, user_id, product_name, order_amount, order_time, status FROM orders_source;这个作业放到Flink SQL Client或者SQL Gateway里执行立刻就能看到binlog变化被翻译成ES写入。SQL方式的优点是不需要编译打包改个字段映射就能上线。缺点是复杂流转比如多表join、窗口聚合会写到SQL里可读性和可调试性都稍微差一些。4.2 用DataStream API处理复杂转换的场景如果只是字段搬运SQL已经够了。但真实场景里我需要把order_time从时间戳转成字符串、把金额从分转成元、还要根据商品ID去关联维表查到分类名称。这种时候我更喜欢用DataStream API因为能精确控制每个字段的处理逻辑调试也很方便。核心代码如下import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.datastream.DataStream; import com.ververica.cdc.connectors.mysql.source.MySqlSource; import com.ververica.cdc.debezium.DebeziumDeserializationSchema; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink; public class MysqlCdcToES { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(10000); MySqlSourceString mysqlSource MySqlSource.Stringbuilder() .hostname(192.168.1.10) .port(3306) .databaseList(shop_db) .tableList(shop_db.orders) .username(cdc_user) .password(StrongPassword) .serverId(5400-5404) .deserializer(new MyDeserializationSchema()) .build(); DataStreamString stream env.fromSource(mysqlSource, org.apache.flink.api.common.eventtime.WatermarkStrategy.noWatermarks(), MySQL CDC Source); stream.map(json - transform(json)) .addSink(esSink()); env.execute(mysql-cdc-to-es); } }MyDeserializationSchema这一步很重要你需要把Debezium格式的变更事件解析成自己的业务对象。一个最简单的做法是把SourceRecord的JSON序列化成字符串然后在后面的map里用Jackson解析。稍微复杂一点你可以直接用Flink官方的JsonDebeziumDeserializationSchema它会把整个change事件包成一个JSON对象里面的before、after、op字段都能拿到。特别注意如果op是d删除你要在代码里主动从ES删除对应文档不然ES里会残留已经删除的记录。ES Sink的写法用ElasticsearchSink时最省事的是让内部连接器根据index参数自动判断索引。但要注意如果Sink里面想批量写入得在Sink函数里拿到BulkProcessor的配置。否则数据量一旦上来单条写入的吞吐会明显受限。4.3 任务提交与启动中常见的环境错误任务写好了数据却没走起来这部分我几乎每次部署都会撞到。首先一个致命问题是本地IDEA跑的时候好好的打成Jar丢到Flink集群全部ClassNotFound。原因通常是Flink SQL客户端的lib目录缺少CDC连接器jar包。在集群里要把flink-connector-mysql-cdc-*.jar、flink-connector-elasticsearch7-*.jar手动放进$FLINK_HOME/lib/然后重启集群。这个步骤官网文档写得很隐晦很多人漏了。另一个典型错误是表字段类型不匹配。比如MySQL里的DECIMAL(10,2)到了Flink SQL默认变成DECIMAL(10, 2)没错但你如果没引入flink-table-common相关依赖SQL初始化时就会报Type does not support the DecimalType。这时候去查依赖冲突把不需要的旧版本exclusion掉即可。启动之后还有两个验证动作第一手动在MySQL里insert一条数据然后立刻去ES查询看文档是否出现第二执行update操作看ES文档是否更新。如果发现insert有、update没有多半是ES Sink的主键没定义对或者索引mapping里_id没有使用MySQL主键。这个坑排查起来最耗时原因往往只在sql的PRIMARY KEY声明上面。5. 同步链路稳定运行的关键监控、断点恢复与背压处理5.1 Checkpoint配置与恢复策略线上任务跑起来最怕的不是代码逻辑出错而是机器宕机、网络抖动、磁盘满这类基础设施问题。Flink CDC的恢复能力都建立在Checkpoint上。对于同步延迟要求高的任务我通常把Checkpoint间隔设定在10到30秒之间如果间隔太长恢复时binlog位点落后太多ES数据会有一段明显的空洞窗口。Checkpoint存储路径一定要配置否则Flink会用默认的JobManager内存来存状态一旦job重启状态全丢。我常用的配置state.backend: filesystem state.checkpoints.dir: hdfs://nameservice/flink-checkpoints execution.checkpointing.interval: 10000 execution.checkpointing.tolerable-failed-checkpoints: 3任务因为逻辑错误反复失败时Flink默认只会在Checkpoint失败次数达到阈值后才停止。如果阈值设置过大任务会在失败与恢复之间来回震荡导致MySQL侧积压大量binlog。所以这里的可容忍失败次数不是越大越好我一般取2~3真的出问题了宁可重启也要及时发现问题。恢复时用--fromSavepoint / --allowNonRestoredState从指定目录拉起。如果Status中显示Error先去看TaskManager的日志看是不是上游MySQL连接被长时间闲置断掉了。Flink CDC内部其实有自动重连机制但如果你的MySQL设置了wait_timeout小于任务的空闲时间还是会出现连接被服务端关闭的情况。解决方法是在MySQL连接参数里增加connectTimeout60000或者在JDBC URL里设置了后依然手动在Flink配置里把heartbeat.interval调小。5.2 常见故障从重复数据到类型转换失败重复数据问题由于ES的更新操作是幂等的相同ID会覆盖重复数据一般不影响最终状态。但如果你做的是append-only同步每次变更都应当新增文档那就要确保ES Sink里没有设置document-id或者业务主键生成规则能区分版本。否则在Exactly-once与At-least-once的混用场景下重复写入会污染索引。类型转换失败ES对日期格式非常敏感。MySQL里2024-01-15 10:31:00Flink CDC默认会解析为带T的国际格式ES默认date类型也接受这个格式但如果你在SQL里显式转换成了字符串再塞进ES的date字段结果报错。我的处理方式是在ES mapping里把date字段的format设置成yyyy-MM-dd HH:mm:ss||strict_date_optional_time||epoch_millis并且不要在Flink SQL里对时间做多余的类型转换。脏数据导致任务失败当MySQL表里有非法字符或者超长字符串时ES的bulk写入可能返回Mapping Exception这时Flink任务会持续失败。处理办法是开启Sink的容错参数比如sink.bulk-flush.backoff.strategy CONSTANT或者干脆在代码里捕获写入异常并记录到错误日志而不是让它拖垮整个任务。我在生产里用过一个简单方案在SinkProcessor里catch异常写日志任务继续跑晚一点统一处理错误表。5.3 背压与性能调优的方向流量一旦大起来细心的人会发现Flink UI里subtask的BackPressured时间比例开始变高。背压的直接原因是下游的ES写入比不过上游的binlog解析速度。这个问题的瓶颈大多数情况下在Elasticsearch端的bulk参数配置。如果bulk queue太小一次性POST的文档数量太大ES返回429或超时就会导致Flink Sink阻塞。这里我常用的几个方向按优先级排ES的bulk批量大小保持默认5mb或1000条但调整bulk-flush.interval从1秒改成100毫秒让数据更频繁地flush出去。增加ES写入并发度即调大Flink Sink算子的并行度同时确保ES索引的分片数不成为瓶颈。索引分片太多会导致每个请求都分散到多个分片反而变慢。在Source端使用scan.incremental.snapshot.chunk.size字段控制全量阶段每次读取的行数。这个值不是越大越好一般8000左右比较均衡。如果业务表数据量到几千万建议允许Flink自动分配chunk也可以手动设置同时读取的表数量。如果MySQL的binlog增长速度很快比如有频繁批量update造成大量日志建议在源表上做降级策略比如在SQL里先做WHERE status IN (...) AND update_time ...之类的过滤减少无谓的行进入Sink。另外让我印象很深的一个调优点ES的Sink在Flink 1.15之后可以设置SinkFunction的setBulkFlushMaxSizeMb和setBulkFlushIntervalMillis。我当时只改了这两个参数同步吞吐就从3000条每秒涨到了8000条每秒效果立竿见影。这个方法值得优先尝试。6. 迭代到生产环境的几个细节优化6.1 多表同步与路由策略现实中很少只同步一张表。订单与订单明细、用户与商品分类经常要同时进ES。Flink CDC是支持在同一个作业里监听多个表的你可以在databaseList和tableList里用正则匹配。但要注意如果多表进入同一个ES索引必须在Sink里通过document-id.key-delimiter把不同表的主键拼接成唯一ID避免冲突。我当时的做法是在DataStream里针对每张表分别定义Source再用union操作把多流合并流里的数据附带一个tableName字段Sink函数里根据tableName路由到不同的ES索引或者不同的type。这样虽然需要多写点Java代码但调试直观也能够分别控制不同表中字段映射逻辑。如果全用SQL也能通过CREATE TABLE ... LIKE实现多索引但一个作业内联合多表明细的SQL会让任务日志变得很乱维护起来体验一般。6.2 数据一致性对比工具与自检脚本同步链路不是布完就万事大吉数据一致性的验证要放在任务上线前做一次上线后定期抽查。我习惯用这样的方式从MySQL查某一天全量数据的主键集合再在ES里聚合出同一天文档的ID集合两个集合做set差集就能快速找出缺失或多余的数据。一个简单的自检脚本思路伪代码# 从MySQL取主键集合 mysql_pks query_db(select id from orders where order_time 2025-01-01) # 从ES取文档ID集合 es_ids query_es(select _id from orders_index where order_time 2025-01-01) # 差集 missing mysql_pks - es_ids extra es_ids - mysql_pks if missing: print(缺失数据:, missing[:100]) if extra: print(多余数据:, extra[:100])这个脚本看着简单但真能救命。有一次我改了ES Sink的主键分隔符忘了重新跑全量结果线上任务运行了一天ES里多了几万条冗余文档就是靠这个脚本发现的。对关键表我建议每12小时跑一次这样的校验比看Flink UI更直观。另外建议在MySQL和ES之间建一张元数据表记录每次同步的binlog位点与时间。这样出现数据对不上的时候能很快定位到是全量阶段丢失还是增量阶段乱序。6.3 后续扩展从单一ES到更丰富的下游当同步链路跑稳了之后很容易会想既然Flink CDC已经能听binlog那我能不能不止写ES同时把数据也放到Kafka做异步消息、写一份到ODPS做离线分析理论上完全可以只要在一个作业里定义多个Sink让Source流向两条支线即可。这个扩展思路有一个典型风险如果写ES和写Kafka在两个Sink里这两个Sink的Checkpoint是共享的一个写入失败会把整个作业拖住。所以我的经验是如果只是想复制数据到Kafka可以单独开一个Flink作业专门负责Source到Kafka另开一个作业负责Source到ES。解耦后ES的慢查询不会再阻塞Kafka的实时链路两个作业可以独立重启运维成本其实更低。当你要同时同步多套环境测试库、预发库、生产库考虑用Flink SQL里的catalog去管理多数据源。比如创建一个mysql_catalog再用USE CATALOG mysql_catalog切换写同步任务时只需要维护一段SQL逻辑。这套方式非常适合团队内部数据平台化的场景。到这一步MySQL到ES的同步就不再是一个“从零搭管道”的事了而是一个需要日常维护、校验、调优的数据基础设施。我在实际运维里最深的体会是Flink CDC本身很强大但真正决定同步链路稳不稳的往往是MySQL的binlog保留策略、ES的mapping设计、以及失败之后的恢复策略这些看起来琐碎的小事。如果你正在做类似的同步建议先把这些前提都捋清楚再往下铺任务能省掉后续大量不必要的折腾。
