做实时数据同步这些年绕不开一个经典场景把 MySQL 里的业务数据搬到 Elasticsearch给搜索、筛选、报表加速。我之前维护过一个库存查询系统几十张表每天晚上全量灌进 ES白天业务库一更新ES 就对不上最后只能重刷越刷越被动。后来切到 FlinkCDC 这套方案从 Flink SQL 到 DataStream API 都实际跑过今天把“FlinkCDC 实战将 MySQL 数据同步至 ES”这条链路的选型、依赖、代码、踩坑完整拆开讲一遍给准备做实时同步的同学一个可直接抄作业的参考。这套方案解决的核心问题很简单让 MySQL 的增删改自动变成 ES 索引的增删改延迟做到秒级并且支持历史数据全量回刷不用再手工写定时 Job。适合正在选型实时同步方案、或者已经决定用 Flink 但不知道怎么接 MySQL 和 ES 的人尤其是团队里已经引入 Flink 的成本比维护一套独立的同步集群低很多。下面从方案选型开始一步步展开。1. 为什么选择 Flink CDC 做 MySQL 到 ES 的同步1.1 先跟 Canal、DataX、自研定时任务对比做 MySQL 到 ES 的同步能选的路不少但每一条都有明显的取舍。自研定时任务本质上就是定期全量查一遍 MySQL然后重建索引。问题很直接数据量一大全量重建耗时很长期间 ES 查询的是一份旧数据或半成品数据。而且业务库白天有高并发写入你再用定时全量扫描去读会把 MySQL 的 IO 打高影响线上业务。只适合数据量小、实时性要求不高的场景。DataX 等离线同步工具可以做全量和部分增量但它的定位是批式数据搬迁不是持续监听 binlog。要模拟实时的效果得每隔几分钟调度一次中间窗口的数据延迟不可控而且同步状态管理很别扭。CanalCanal 本身很棒它能稳定地从 MySQL binlog 解析出变更事件。但 Canal 只是一个数据管道真正要同步到 ES你还要自己写一个消费端处理连接状态、位点保存、重启恢复、并发写入 ES 这些问题。说白了你拿到的是半成品后面一大半工程化工作要自己做。Flink CDC直接封装了 binlog 解析把整张表当成一个无界流。历史数据全量扫描和实时增量变更由框架自动衔接checkpoint 里保存 binlog 位点任务重启后能接续不用自己管理状态。更重要的是它和 Flink 生态天然打通下游接 ES、Kafka、Hudi 都是现成的 connector。所以我的结论很明确如果业务数据在 MySQL目标端要实时同步到 ES而且团队已经用或准备用 Flink直接选 Flink CDC 是最省心的路径。1.2 实时链路里的三种典型架构实际落地时不是所有场景都适合让 Flink CDC 直接连 ES要看下游消费的形态。方式一Flink CDC - Elasticsearch。最简单直接建两个表、一条 INSERT INTO同步任务就完成了。适合单系统内的核心表同步不涉及多团队多系统消费。缺点是如果后面要再加一个数据消费方比如还要同步到 ClickHouse你得再开一个 CDC 任务多读一份 binlog对源库和网络都是额外负担。方式二Flink CDC - Kafka - Flink - Elasticsearch。binlog 变更先进 KafkaKafka 作为数据总线下游可以同时挂 ES、数仓、其他业务系统。这种架构解耦更彻底但多引入 Kafka 一套组件同步链路延迟会高一点点运维成本也上升。适合数据需要被多份消费的团队。方式三Flink CDC Flink SQL 关联维表 - Elasticsearch。比如订单表同步到 ES 时还需要把商品名称、用户等级这些维度字段拼接进宽表。直接用 Flink SQL 做 join中间不需要额外落 Kafka。这种方式能实现比较复杂的宽表构建。本篇以方式一为主线讲解因为最核心的 CDC 机制是一样的。如果你的场景需要走 Kafka把中间环节换成 Kafka sink 和 Kafka source 即可Flink CDC 的配置完全复用。1.3 全量增量协调Flink CDC 的核心机制Flink CDC 连接器内部是基于 Debezium 实现的但它在传统 Debezium 基础上做了一个很重要的升级增量快照框架。它把同步过程拆成两段全量阶段任务启动后先把表中的存量数据按主键分片读取并行度可以很高。全量读不代表源库持续锁表框架会利用一致性快照获取一个 binlog 位点然后一边读存量数据一边把这段时间内新产生的 binlog 变更缓存下来。增量阶段存量数据读完之后框架自动切换到位点之后的 binlog 消费保证全量和增量之间不丢数据。这个切换对开发者完全透明。没有 checkpoint 的 Flink CDC 是不完整的。binlog 当前消费位点、全量阶段读到哪里这些状态都存在 Flink 的状态后端里。只有开启 checkpoint任务重启后才知道从哪里续读否则重启只能重新全量扫一遍大表会很痛苦。后面我会专门讲 checkpoint 怎么配。2. 环境准备与依赖版本2.1 版本搭配表版本选型是 Flink CDC 最容易翻车的地方没有之一。Flink CDC 的版本和 Flink 主版本不是一一严格对应关系而是有一个兼容区间。我用下来相对稳的组合如下仅供参考具体以官方文档为准Flink 版本Flink CDC 版本Elasticsearch 连接器说明1.15.x2.3.xflink-sql-connector-elasticsearch7:1.15.x经典组合社区资料最多适合稳定项目1.16.x2.3.x 或 2.4.x1.16.x增量快照和并行度控制更好1.17.x2.4.x1.17.x我近期主力稳定性不错1.183.0需分别下载 pipeline connector3.x 架构变化较大生产迁移要谨慎评估这里重点说明一下Flink CDC 2.x 的连接器 groupId 是com.ververica到 3.x 之后组织迁移到org.apache.flink并且新增了 YAML pipeline 模式可以把整条链路用一份 pipeline 配置文件描述少写很多 Java 代码。但 3.x 的生态迁移还在推进中如果你的项目是现有 Flink SQL/DataStream 任务不建议为了追新盲目升级。生产环境优先选你熟悉的 Flink 版本配套的 CDC 版本。ES 端我用的是 7.x对应 Flink 官方提供的elasticsearch-7connector。如果你们已经上 ES 8需要额外确认连接器兼容性官方 Flink SQL 连接器对 ES8 的支持一直不太痛快。2.2 用 Docker Compose 拉起 MySQL、ES 和 Kafka本地调试时我最常用 Docker Compose 一次性拉起整套环境。MySQL 必须开启 binlog并且格式是 ROWFlink CDC 才能拿到完整的前后镜像。version: 3.8 services: mysql: image: mysql:8.0 environment: MYSQL_ROOT_PASSWORD: root123 MYSQL_DATABASE: testdb ports: - 3306:3306 command: [ --server-id1, --log-binmysql-bin, --binlog-formatROW, --binlog-row-imageFULL, --gtid-modeON, --enforce-gtid-consistencyON ] elasticsearch: image: docker.elastic.co/elasticsearch/elasticsearch:7.17.9 environment: discovery.type: single-node ES_JAVA_OPTS: -Xms512m -Xmx512m ports: - 9200:9200 kibana: image: docker.elastic.co/kibana/kibana:7.17.9 ports: - 5601:5601 environment: ELASTICSEARCH_HOSTS: http://elasticsearch:9200 depends_on: - elasticsearchMySQL 参数里binlog_row_imageFULL很重要。如果设置成 MINIMALupdate 事件只包含被修改的列ES 那边的 upsert 映射会变得很麻烦。server-id1只是给当前容器实例用但你跑 Flink CDC 时连接器会模拟一个从库连上来这个 server-id 必须和 MySQL 实例自己的 server-id 以及线上已有的从库区分开否则会被 MySQL 判定为 server-id 冲突直接断连。如果走 Kafka 链路再补一个 Kafka 服务即可注意版本和 Flink kafka connector 的兼容性本地测试用 3.x 问题不大。2.3 pom.xml 依赖清单我用的是 Flink 1.17 Flink CDC 2.4 ES7pom 里核心依赖如下dependencies dependency groupIdcom.ververica/groupId artifactIdflink-sql-connector-mysql-cdc/artifactId version2.4.2/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-sql-connector-elasticsearch7/artifactId version1.17.1/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version1.17.1/version /dependency /dependencies注意这两个 connector 都是带依赖的 fat jar里面已经集成了 MySQL 驱动、Debezium、ES REST 客户端不需要额外引入。如果你在 IDEA 里写完代码直接mvn exec:java跑建议把 scope 调成 compile如果提交到 Flink 集群记得把依赖打进发布包或者放在集群 lib 目录。3. 核心实现同步任务怎么写3.1 最快的落地方式Flink SQL 一把梭如果同步逻辑不复杂用 Flink SQL 是最快的。核心思路是建一张 MySQL 源表一张 ES 目标表然后INSERT INTO ... SELECT。先建 MySQL CDC 源表CREATE TABLE mysql_user ( id INT, name STRING, age INT, email STRING, create_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname localhost, port 3306, username root, password root123, database-name testdb, table-name user, server-id 5400-5404, scan.startup.mode initial );这里有几个关键点scan.startup.mode默认就是 initial表示全量加增量。如果只想消费新变更可以配 latest-offset想从某个时间点消费配 timestamp。一般做数据同步都建议 initial先把历史数据补齐。server-id我习惯给一个范围比如 5400-5404不是单值。因为源表在 Flink 里可以并行读取每个并行分片需要一个独立的 server-id。你的任务并行度如果是 3给至少 3 个 id。如果多个 CDC 任务同时连同一个 MySQL 实例每个任务都要换个不重叠的范围。PRIMARY KEY 声明是必须的不仅是为了 CDC 分片也关系到下游 ES 的文档 ID 映射。MySQL 表如果没有主键后面会遇到很坑的问题我放到常见问题里细说。再建 ES 目标表CREATE TABLE es_user ( id INT, name STRING, age INT, email STRING, create_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector elasticsearch-7, hosts http://localhost:9200, index user, sink.bulk-flush.max-actions 500, sink.bulk-flush.max-size 10mb, sink.bulk-flush.interval 2000 );然后执行同步INSERT INTO es_user SELECT id, name, age, email, create_time FROM mysql_user;这个任务跑起来后MySQL 里新增数据、修改数据、删除数据都会对应转换到 ES 的 index 操作。你可以在 Flink UI 上看到 source 和 sink 的实时 record 数。3.2 必须注意的 ES 文档 ID 问题ES sink 最容易被忽略的是文档 ID。Flink ES connector 默认情况下生成的文档 ID 是随机的也就是说即使你同步的是同一行数据更新时也可能产生一份新文档而不是更新旧文档。删除操作更没法保证删到正确的文档。解决办法是在 ES 目标表的 DDL 里声明主键PRIMARY KEY (id) NOT ENFORCED。连接器会自动用主键值构造 ES 的_id。MySQL 表的主键是 idES 文档_id也会是 id 的值这样 update 落到同一条文档delete 也能正确触发。如果你在 DataStream API 里自己写 ES sink可以自定义IndexRequest的 id原理是一样的。但用 Flink SQL 时忘写主键往往是“同步后 ES 数据越来越多一份数据好几个副本”的元凶。3.3 复杂场景用 DataStream API 兜底Flink SQL 做不了复杂转换时就要回到 DataStream API。典型场景包括需要把 Debezium JSON 里的 before、after、op 做细粒度控制需要根据 binlog 事件类型决定 ES 的 index、update、delete 动作需要关联多张表后输出。核心代码如下MySqlSourceString source MySqlSource.Stringbuilder() .hostname(localhost) .port(3306) .databaseList(testdb) .tableList(testdb.user) .username(root) .password(root123) .serverTimeZone(Asia/Shanghai) .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); DataStreamSourceString stream env.fromSource( source, WatermarkStrategy.noWatermarks(), mysql-cdc-source );JsonDebeziumDeserializationSchema输出的是 Debezium 风格的 JSON结构大致是{ before: null, after: { id: 1, name: 张三, age: 20 }, op: c }op字段对应事件类型c 是 insertu 是 updated 是 deleter 是快照阶段读取的存量记录。DataStream 里要做的是解析出op然后决定发IndexRequest还是DeleteRequest。同步 SQL 模式不用关心这些但 DataStream 模式下这块是绕不开的。stream.map(new DebeziumToDocRequest()) .sinkTo(new ElasticsearchSink(...));ElasticsearchSink需要实现ElasticsearchSinkFunction里面拿到DocRequest后根据 action 类型构造请求。这里不再贴完整代码核心思想是不要一股脑把每条记录都当成 insert 写入delete 事件必须显式转成DeleteRequest否则 ES 里的死数据会一直留着。3.4 提交任务前先配好 checkpoint不管用 SQL 还是 DataStream提交之前一定要确认 checkpoint 已经打开。Flink SQL 客户端跑任务时默认可能没开DataStream 里要显式配置env.enableCheckpointing(5000); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000); env.getCheckpointConfig().setCheckpointStorage(hdfs://namenode:8020/flink-cdc-checkpoint); env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);checkpoint 的核心作用有两个保存 binlog 消费位点使任务重启后能从最近一次成功的位点续读而不是重新全量扫表。给 ES sink 提供一致性的基础。ES 连接器本身做不到精确一次但至少能保证不会因为任务重启而漏掉太多数据。本地测试时可以直接把 checkpoint 存储配到本地文件系统如下env.getCheckpointConfig().setCheckpointStorage(file:///tmp/flink-cdc-checkpoint);如果 checkpoint 配得太频繁比如 1 秒一次MySQL binlog 消费较慢时容易堆积瓶颈期我一般从 5 秒开始调压缩后单表同步基本够用。这个参数没有最优只有最适。4. 实操验证从全量到增量的完整链路4.1 造数据并启动任务先把 Docker Compose 里的 MySQL 和 ES 拉起来在 MySQL 中创建 user 表并插入几条数据CREATE TABLE user ( id INT PRIMARY KEY, name VARCHAR(50), age INT, email VARCHAR(100), create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); INSERT INTO user VALUES (1, 张三, 20, zhangsanexample.com, NOW()), (2, 李四, 25, lisiexample.com, NOW());真实业务表字段更多我这里只保留最典型的结构。接着启动 Flink 任务在 Flink Web UI 里观察 source 的 records 数。任务跑起来后ES 里的 user 索引应该已经有两条文档。如果之前创建过同名的 index建议先删掉避免类型映射残留旧字段。访问 ES 验证一下curl http://localhost:9200/user/_search?pretty返回的hits.total应该是 2。此时全量阶段已经完成。4.2 验证 DML 变更同步全量同步只是第一步增量同步才是 CDC 的核心要逐个验证 insert、update、delete。先在 MySQL 里插入一条新数据INSERT INTO user VALUES (3, 王五, 30, wangwuexample.com, NOW());正常情况下一两秒后 ES 就能查到 id3 的文档。如果长时间没出现优先怀疑 checkpoint 没开或 binlog 消费位点没有推进。再执行一次更新UPDATE user SET age 31 WHERE id 3;验证 ES 中 id3 的文档 age 字段是否变为 31。这里必须确认 ES 目标表 DDL 里有PRIMARY KEY (id) NOT ENFORCED。如果没有主键你看到的很可能不是 age 更新而是新增了一条 id 随机、age31 的文档。最后验证删除DELETE FROM user WHERE id 3;同样去 ES 里查 id3应该已经不存在。如果删除没生效大概率是因为源表 DDL 或目标表 DDL 没声明主键导致 delete 事件无法定位文档。4.3 大表全量同步的调优思路实际业务很容易遇到几千万行甚至上亿行的大表全量阶段可能是整个同步链路中最危险的一段。默认情况下 Flink CDC 会把表按主键分片每个分片读取一批数据。分片大小由scan.incremental.snapshot.chunk.size控制默认是 8096。对大表这个参数很值得调。如果分片太大单个 chunk 读取时间过长内存和网络压力大还容易超时。如果表主键不是均匀分布的比如自增主键中间有大量空洞默认分片方式会读到很多空数据效率反而低。可以适当减少 chunk size比如 2000 到 3000配合提高并行度让全量阶段更轻快。全量阶段还需要关注源库的 IO。Flink CDC 虽然做了一致性快照但毕竟是高强度扫描线上表建议低峰期触发首次全量或者先同步到测试环境验证性能再在正式环境跑。5. 常见问题与排查实录5.1 无主键表同步失败Flink CDC 官方要求源表必须有主键或者唯一键主要原因是框架需要用它来做全量分片和增量事件的定位。没有主键的表会报类似“Cannot find a suitable chunk key”的错误。实际处理不只有加主键这一条路以下几种方案都可以改表加主键最彻底但线上大表加锁风险高需要 DBA 配合。用唯一键代替MySQL CDC 配置里可以指定scan.incremental.snapshot.chunk.key-column如果你有唯一性较强的业务字段可以顶一下。DataStream 模式自定义分片键复杂但对不可改表的情况最通用。还有一个典型的误判MySQL 表明明有主键但主键是复合主键且包含VARCHAR类型且字段值非常长。这种情况分片时也会出问题最好给表加一个自增的代理主键或者用单独的唯一数字列。5.2 binlog 格式、时区、日期类型问题同步结果和源库对不上最常见的原因有三个binlog_format 不是 ROW。MySQL 只有 ROW 格式才能从 binlog 完整解析出 before 和 afterSTATEMENT 格式下 Flink CDC 拿不到准确变更数据。用SHOW VARIABLES LIKE binlog_format检查。时区不一致。MySQL 的 TIMESTAMP 存储的是 UTC 时间展示时按 session 时区转换。Flink CDC 读取时如果没指定正确时区ES 里看到的时间可能差 8 小时。建议在 source 连接参数里明确server-time-zone或者设置connectionTimeZone不要依赖默认值。WITH ( connector mysql-cdc, server-time-zone Asia/Shanghai )DATETIME 和 TIMESTAMP 的精度问题。MySQL 的 DATETIME 默认到秒而 Flink 的 TIMESTAMP(3) 是毫秒级如果原表用了ON UPDATE CURRENT_TIMESTAMP同步到 ES 后的数值可能和预期略有差异。不是大问题但做数据对比时要知道有这个精度差。5.3 checkpoint 失败导致一直重新全量这是一个非常隐蔽的坑。任务在增量阶段正常跑某天一次网络抖动导致 checkpoint 连续失败如果配置了重启策略任务会尝试从最近一次成功的 checkpoint 恢复。但如果你的 checkpoint 持久化目录有问题或者setMinPauseBetweenCheckpoints配置不当状态恢复一直失败Flink CDC 就会退回到重新全量扫描的路径。大表在深夜被打回全量资源开销会非常夸张。我给几个建议一定要配独立的 checkpoint 存储目录不要用本地临时路径跑生产任务。配置setTolerableCheckpointFailureNumber(3)允许偶发失败而不是立刻触发大重启。开启setExternalizedCheckpointsRetention(RETAIN_ON_CANCELLATION)任务主动停止时保留状态文件后续可以从保存点恢复。如果任务已经反复全量多次最直接的止损办法是停掉任务用 savepoint 恢复而不是任它继续循环。5.4 分库分表多张表如何同步一个索引业务上经常有 user_0、user_1 这种分表结构同步到 ES 时希望合并到一个 user 索引。Flink CDC 的配置是支持正则匹配的WITH ( connector mysql-cdc, database-name testdb, table-name user_[0-9], server-id 5400-5404 )source 表 DDL 里字段结构要保持一致然后 sink 都写到同一个 index。这里要注意一点多张表的文档 ID 可能会冲突。如果每张分表的主键都是从 1 开始自增直接同步到同一个 ES 索引会导致文档互相覆盖。解决办法有两种要么同步时给主键加表前缀把文档 ID 改成user_0_1这种形式要么在 ES 索引设计阶段就用复合业务键作为唯一 ID。我踩过这个坑之后现在一律要求分库分表场景先校准文档 ID 策略再上线。5.5 写入 ES 的字段类型不对怎么办ES 的动态映射很聪明但也很笨。比如 MySQL 的VARCHAR字段ES 会自动映射成text然后额外生成一个keyword子字段。如果你在 ES 里做精确筛选会发现字段带上.keyword才有效不带就查不到非常反直觉。更麻烦的是MySQL 的表结构后续加了字段ES 索引里没有对应 mapping同步虽然是成功的但新字段可能被动态映射成一种奇怪的类型。我建议正式环境提前把 ES 的 index template 建好把关键字段的type、分词器、format 都固定下来。索引模板的作用就是让你在 Flink 任务启动前已经决定了 ES 端的数据形状而不是等任务跑起来让 ES 猜。5.6 其他高频问题速查表现象常见原因处理方式任务启动报 server id 冲突多个任务使用相同 server-id 范围每个任务独立分配不重叠的范围ES 里出现重复文档目标表 DDL 没声明主键加PRIMARY KEY NOT ENFORCEDupdate 变成新增文档文档 ID 未映射到业务主键用主键构造 ES _iddelete 不生效主键缺失无法定位文档补齐主键后重启任务全量阶段 OOM分片设置过大或并行度不合理减小 chunk size调大并行度同步后时间字段少 8 小时时区配置不一致设置 server-time-zone校验连接参数任务重启后重新全量checkpoint 未开启或恢复失败开启 checkpoint检查状态存储目录6. 最后一点经验这套链路跑顺之后能明显感受到实时同步和定时重建之间的差别。我个人在实际操作里的体会是Flink CDC 最大的价值不是省掉那几行同步代码而是把“数据变更”这件事从业务系统里解耦出来让下游系统可以按自己的节奏消费同一份变更流。建议团队第一次落地时先挑一张核心小表跑通全流程验证主键映射、删除语义、时区表现再逐步扩大到全量表。别一上来就追求所有表都实时同步链路越多源库压力、checkpoint 开销、ES 写入放大这些成本都会线性增长先保住最核心的几张表稳定运行比铺开一堆时好时坏的任务要实在得多。