企业官网建设流程全解析

1. 实时同步的需求从哪来：MySQL 与 ES 的天然错位

我平时接到最多的数据需求，十有八九长这样："订单表里的状态能不能实时同步到 ES？我们搜索那边查不到最新数据。" 说白了，业务库 MySQL 负责扛事务，搜索和查询分析交给 Elasticsearch，两边的数据模型和用途完全不同。你不可能让所有查询都打到 MySQL 上，否则线上业务一到大促就顶不住；也不可能让 ES 直接承载写入，它就是为查询而生的。那中间的桥梁就得靠实时同步链路。

Flink CDC 就是干这个的。CDC 全称 Change Data Capture，翻译过来是变更数据捕获，它不靠业务方显式写双写逻辑，而是直接解析 MySQL 的 binlog，把 insert、update、delete 一条条捞出来，再以流的形式往下游推。搭配 Flink SQL，你可以像写普通 SQL 一样定义一张 MySQL 源表和一张 ES 目标表，一句 insert into 搞定同步。这在几年前想都不敢想，当时我们同步一份数据到 ES，要么写定时任务全量刷，要么自己解析 binlog 再手动推，成本和复杂度都很高。

这整套思路适合谁来参考？主要是三类人：一是数据开发，想把实时数仓的链路落地；二是后端工程师，需要在业务系统里快速搭建搜索数据的同步管道；三是运维或者平台工程师，要评估 Flink CDC 方案能不能扛住生产流量。读完这篇文章，你至少能独立搭出一条 MySQL 到 ES 的实时同步链路，并且知道哪些配置是坑、哪些参数在生产环境必须改。

2. 为什么选 Flink CDC：方案对比里的取舍逻辑

2.1 常见同步方案逐个拆

做 MySQL 到 ES 的同步，市面上不缺方案，关键是你愿不愿意为它付出额外的运维成本。

• 定时全量同步：最粗糙，crontab 跑一个脚本，把 MySQL 数据查出来灌进 ES。数据量小的时候能凑合用，数据量一大就会出现明显的时间差，业务上"刚下单搜不到"的体验就会很差，而且每晚全量对 MySQL 和 ES 都是压力。
• Canal 同步：Canal 伪装成 MySQL 从库，接收 binlog 后推给 MQ，再由消费者写入 ES。这套链路成熟，但你需要部署 Canal、部署 Kafka 或 RocketMQ、写消费者，组件多，任何一个环节出问题都要排查半天。
• 基于 Debezium + Kafka Connect：思路与 Canal 类似，Debezium 负责监听 binlog，Kafka Connect 负责分发，最后还要自己写 sink。配置多、版本兼容问题也不少。
• Flink CDC：核心差异在于，它把"解析 binlog、转换、写入 ES"压进了一个 Flink 任务里，依赖只有 Flink 集群和连接器 jar。使用门槛一下子降下来了。

2.2 为什么 Flink CDC 是最优解

从实际维护角度讲，Flink CDC 的优势有三个。

第一，Flink SQL 的 DSL 让同步任务像写 SQL 一样简单。不需要写一堆 Java 代码，一张 CREATE TABLE 定义源，一张 CREATE TABLE 定义目标，再写一句 INSERT INTO，任务就跑起来了。对于大多数后端开发来说，这比研究 Canal 的 adapter 和 Debezium 的 JSON 格式要友好得多。

第二，Flink 的 Checkpoint 机制天然保障了"至少一次"和"精确一次"语义。同步任务最怕丢数据，binlog 消费到一半挂了，重启后从哪接着读？Flink CDC 会把 binlog 位点保存在状态里，配合 Checkpoint 定期持久化，挂了重启能续上，不会从头扫全量。这一点在做生产链路时非常关键。

第三，Flink CDC 内置了增量快照框架，对于已经存在大量历史数据的表，第一次启动会自动做一次全量快照，快照完成后再无缝切换为 binlog 增量。整个过程不需要你手动停机、导出、导入，所谓"先全量后增量"，它帮你包办了。

注意：选型的时候别只盯着 demo 跑通的那一下。生产上你要考虑组件数、故障恢复、监控告警、版本升级路径，Flink CDC 在组件数量和运维复杂度上是明显占优的。

3. 前置准备：版本选型与环境搭建

3.1 版本匹配是第一个大坑

Flink CDC 的版本匹配问题，我见过太多人踩坑。下载一个 2.x 的 mysql-cdc 连接器，配上 Flink 1.14，结果启动就报各种 ClassNotFoundException。原因在于不同版本的 Flink CDC 对 Flink 的编译依赖不一样，对 Scala 版本也有要求。

我的建议是直接选一个经过验证的组合。当前主流稳定的方案是 Flink 1.17 或 1.18 搭配 Flink CDC 2.4.x，这是社区测试比较充分的版本组合。如果你用的发行版是 Flink 1.16 及以下，建议升级，不然后续很多新特性都用不上。

具体依赖这样配：

• Flink 集群版本：1.17.2
• flink-sql-connector-mysql-cdc：2.4.2
• flink-sql-connector-elasticsearch7：1.17.2（注意这个连接器版本号前面对应 Flink 版本，别搞混）

这里提醒一句，连接器的 jar 包要放到 Flink 的 lib 目录下，或者用-j参数提交时指定。如果提交 SQL 任务时报找不到连接器，大概率是 jar 包没放到位。

3.2 MySQL 侧需要开启 binlog

MySQL 默认不开启 binlog，而 Flink CDC 的一切都建立在 binlog 之上，所以你得先确认你的 MySQL 实例到底有没有开。

登录 MySQL 执行：

SHOW VARIABLES LIKE 'log_bin'; SHOW VARIABLES LIKE 'binlog_format'; SHOW VARIABLES LIKE 'binlog_row_image';

如果log_bin是 OFF，那就需要改配置文件。确认 MySQL 的 my.cnf 中存在以下配置：

[mysqld] server-id = 123456 log_bin = mysql-bin binlog_format = ROW binlog_row_image = FULL expire_logs_days = 7

为什么必须是 ROW 格式？因为 Statement 格式记录的是 SQL 语句，同步时要重新执行一遍，精度和一致性都很差；ROW 格式直接记录每一行的变更前值和变更后值，Flink CDC 才能拿到准确的数据。至于binlog_row_image = FULL，是为了让 binlog 里包含完整的行数据，否则有些情况下只有变更列的值，解析出来的数据不完整。

改完配置后重启 MySQL 服务。注意，修改 my.cnf 属于会重启实例的操作，必须提前评估对线上业务的影响。如果你用的是云数据库，通常在控制台就能开启 binlog 开关，不用动配置文件。

3.3 创建 MySQL 同步账号

Flink CDC 读取 binlog 需要一个账号，建议单独创建，不要用业务账号直接复用。

CREATE USER 'cdc_user'@'%' IDENTIFIED BY 'YourStrongPassword123'; GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'cdc_user'@'%'; FLUSH PRIVILEGES;

这里REPLICATION SLAVE和REPLICATION CLIENT是读取 binlog 的必需权限，SELECT用于全量快照阶段读取数据。如果表很多，建议授权后马上测试一下连接，避免后面跑任务才发现权限不足。

3.4 ES 索引准备

ES 侧相对简单，先分析业务字段，确定好 mapping 再建索引。因为 Flink SQL 写入 ES 时，如果索引不存在默认会自动创建，但自动创建的 mapping 往往不符合生产要求，尤其是分词器、字段类型这些，后面再改 mapping 很麻烦。

比如一个订单索引，我通常会这么设计：

PUT /orders { "mappings": { "properties": { "order_id": { "type": "keyword" }, "user_id": { "type": "keyword" }, "product_name": { "type": "text", "analyzer": "ik_max_word" }, "order_status": { "type": "keyword" }, "total_amount": { "type": "double" }, "create_time": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis" }, "update_time": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis" } } } }

这里有几个细节值得说。第一，order_id这种需要精确匹配的字段要用keyword，别用text，否则搜索时会被分词器拆得乱七八糟。第二，中文检索字段product_name配置了ik_max_word分词器，如果没有安装 IK 分词插件，先装好，否则索引创建会失败。第三，日期字段统一指定format，避免 Flink 写入时因为格式不匹配而报错。

4. 实战链路搭建：Flink SQL 完成 MySQL 到 ES 同步

4.1 建立 Flink 目录与依赖

假设你在 Linux 服务器上操作，Flink 已经解压到/opt/flink。进入 lib 目录，把两个连接器 jar 放进去：

cd /opt/flink/lib wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-mysql-cdc/2.4.2/flink-sql-connector-mysql-cdc-2.4.2.jar wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch7/1.17.2/flink-sql-connector-elasticsearch7-1.17.2.jar

如果你的网络环境不能直接访问 Maven 中央仓库，那就先在其他机器下载好再传到服务器，注意别下错版本。ES 连接器和 Flink 版本强绑定，1.17 的 Flink 就配 1.17.x 的连接器，不要试图用 1.16 的连接器顶上。

4.2 Flink SQL 的完整任务脚本

进入 Flink SQL 客户端之后，核心就三张表加一条插入语句。我先贴一段最精简可运行的脚本，后面再逐行解释。

-- 1. 源表：MySQL CREATE TABLE mysql_orders ( order_id BIGINT PRIMARY KEY NOT ENFORCED, user_id BIGINT, product_name STRING, order_status STRING, total_amount DOUBLE, create_time TIMESTAMP(3), update_time TIMESTAMP(3) ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = '192.168.1.100', 'port' = '3306', 'username' = 'cdc_user', 'password' = 'YourStrongPassword123', 'database-name' = 'trade_db', 'table-name' = 'orders', 'scan.startup.mode' = 'initial' ); -- 2. 目标表：Elasticsearch CREATE TABLE es_orders ( order_id BIGINT PRIMARY KEY NOT ENFORCED, user_id BIGINT, product_name STRING, order_status STRING, total_amount DOUBLE, create_time TIMESTAMP(3), update_time TIMESTAMP(3) ) WITH ( 'connector' = 'elasticsearch-7', 'hosts' = 'http://192.168.1.200:9200', 'index' = 'orders', 'sink.bulk-flush.max-actions' = '1000', 'sink.bulk-flush.max-size' = '5mb', 'sink.bulk-flush.interval' = '1000' ); -- 3. 启动同步 INSERT INTO es_orders SELECT * FROM mysql_orders;

看到这里你可能觉得太简单了，但它确实就是这么简单。关键在于你要理解每个 WITH 参数背后的含义。

scan.startup.mode = 'initial'表示任务启动时先做一次全量快照，把当前表里已有的数据全部同步过去，之后再实时消费 binlog。如果你只想从当前 binlog 位点开始，不管历史数据，可以改成latest-offset。初次跑通建议用initial，这样能顺便验证全量链路通不通。

PRIMARY KEY NOT ENFORCED这行不是可有可无。Flink 是流处理引擎，它不会去强制执行数据库主键约束，但会拿主键做 upsert 去重逻辑。ES sink 依赖主键来决定文档的_id，如果你不写主键，那 update 和 delete 就会变得很不可控。所以源表和目标表的主键定义必须一致，我这里都用order_id。

目标表里的sink.bulk-flush.*参数是 ES 批量写入的调优参数。ES 的写入走 HTTP 请求，单条插入性能很差，必须用 bulk 批量提交。max-actions=1000表示攒够 1000 条就刷一次，max-size=5mb表示数据量达到 5MB 就刷，interval=1000表示最多等 1 秒也会刷。三个条件满足任意一个即触发 flush。

注意：ES 连接器的 upsert 写入模式下，delete 是通过 index 请求加上_routing等特殊处理实现的。如果你的业务里经常有删除操作，务必测试一下从 MySQL 删除一行，ES 里对应的文档是否也真的消失了，别只看插入和更新。

4.3 提交任务到 Flink 集群

如果你用的是 SQL 客户端，直接粘贴上面的脚本回车，任务就会在 Session 模式下提交。生产环境我更推荐用flink run以 Application 模式提交，这样任务生命周期跟集群解耦，不会因为客户端关了任务就挂。

写一个简单的 SQL 文件sync_orders.sql，内容就是上面三步。然后执行：

/opt/flink/bin/flink run \ -d \ -c org.apache.flink.table.client.SqlClient \ /opt/flink/lib/flink-sql-client-1.17.2.jar \ -f /path/to/sync_orders.sql

或者，如果你们用的 Flink 版本支持flink sql命令直接提交文件，可以更简单。重点记住-d是 detach 模式，提交后终端可以退出，任务在后台跑。不加-d的话，终端一关任务就没了，这个坑我踩过一次，映像非常深刻。

提交成功后，去 Flink Web UI 看任务状态。正常情况下，任务会先进入 RUNNING，然后你观察 Elasticsearch 里的索引，会发现历史数据慢慢写入。等全量阶段完成，再回 MySQL 手动 update 一行数据，几秒之内 ES 里的对应文档就应该更新了。

4.4 验证同步效果

验证的常规操作是：

1. 在 MySQL 中执行SELECT COUNT(*) FROM orders;，再看 ES 中_count是否一致。
2. 在 MySQL 中 update 一条记录，比如把某个订单的order_status从pending改成paid，再到 ES 里GET /orders/_doc/{order_id}看字段是否变化。
3. 在 MySQL 中 delete 一条记录，到 ES 里确认对应文档消失。

这里要特别说明一下 delete 的验证逻辑。如果你不配置删除相关参数，Flink CDC 的默认行为是把 delete 事件转成对 ES 的 delete 请求，前提是主键定义正确。如果主键没定义好，删除可能变成一次带空字段的更新，ES 里删不掉数据。

我自己的经验是，验证时最好写一个小脚本遍历对比两边的主键集合，而不是只测几条数据。数据量大的时候，主键集合比对是发现增量丢失最快的方法。

5. 深入问题排查：我遇到过的 5 个真实故障

5.1 MySQL 没有开启 binlog 导致任务启动即失败

症状：Flink 任务提交后，很快进入 FAILED 状态，日志里报错信息类似The MySQL server is not configured with binlog_format = ROW。

原因：前面提到过，MySQL 8.0 默认启用 binlog 的场景有限，很多自建的 5.7 实例根本没有开 binlog。

排查思路：确认 MySQL 配置，SHOW VARIABLES LIKE 'log_bin'返回ON，并且binlog_format是ROW。注意有些云数据库在控制台显示开启，但实际可能是 STATEMENT 格式，一定要查实际值。

5.2 server-id 冲突导致同步中断

症状：任务运行一段时间后，偶尔报A slave with the same server_id as this slave was connected to the master。

原因：Flink CDC 的每个并行实例会注册成一个 MySQL 从库，默认生成的 server-id 可能是 0，这种情况下多个并行任务同时连接同一台 MySQL，就会发生 server-id 冲突。

解决：在源表 WITH 参数中显式指定 server-id 范围：

'server-id' = '5400-5404'

这里的5400-5404表示给自己预留 5 个 server-id，对应源表并行度最大为 5。如果你给单表设置了并行度 4，这里至少要有 4 个编号。注意，同一个 Flink 集群里多个 CDC 任务如果连的是同一台 MySQL，各自的 server-id 范围不能重叠，否则照样冲突。

5.3 Checkpoint 失败但任务不自动恢复

症状：任务显示 RUNNING，但 ES 数据迟迟不更新，查看 Web UI 发现 Checkpoint 一直在失败。

原因：Flink 默认的 Checkpoint 目录是临时目录，重启后状态丢失。或磁盘空间不足导致 Checkpoint 无法持久化。

解决：一定要配置 Checkpoint 的持久化目录和间隔。在 Flink 配置里加上：

execution.checkpointing.interval: 60s state.checkpoints.dir: file:///data/flink/checkpoints state.backend: rocksdb

顺便说一句，我后来把代码里的 Checkpoint 失败也配了告警，因为"任务还活着但状态已经不可恢复"这种情况最坑，比任务直接挂掉还难发现。

5.4 大表全量同步慢到怀疑人生

症状：一张几亿行的历史表，Flink CDC 的initial模式跑了一两个小时还没进入增量阶段。

原因：Flink CDC 对单表的全量快照，在 2.x 版本里默认只能单并行度执行，一张大表的所有数据都由同一个 subtask 读取，速度自然上不去。

解决思路：如果不要求一次性全量，可以先把数据导入完成再开启 CDC；或者给大表做分区，拆成多张表分别同步。另一个经验是把 MySQL 源表的server-id范围拉大，让 snapshot 阶段可以使用多个连接并行读取表分片。在 Flink CDC 2.3 之后的版本里，如果你的表有主键，快照阶段可以适当提高源表并行度，但需要注意不能高于 server-id 的个数。

5.5 时区问题：ES 里的时间比 MySQL 慢了 8 小时

症状：同步的数据在 ES 里查询，发现时间字段比 MySQL 里存储的值少 8 小时。

原因：MySQL 的timestamp类型存储的是 UTC 时间，展示时依赖会话时区；而 ES 对date类型写入时，如果时间字符串没有时区信息，默认按 UTC 处理。

解决思路：尽量用datetime而不是timestamp来存业务时间，或者在 Flink 的 DDL 里把字段转为 string 再指定时区：

DATE_FORMAT(create_time, 'yyyy-MM-dd HH:mm:ss') AS create_time_str

然后再用带时区的格式写入，让 ES 正确解析。这个问题在跨时区部署时很容易出现，但只要统一规则，处理起来不复杂。

6. 优化与进阶：从 demo 到生产级同步任务

6.1 Flink SQL 任务的调优参数

生产环境的任务绝对不能直接拿 demo 脚本跑。在当前脚本基础上，下面几个参数建议优先加上。

第一，并行度。ES sink 的并行度如果太低，写入很容易成为瓶颈。我常用SET 'parallelism.default' = 4;，然后结合 bulk-flush 参数观察写入速率。并行度也不宜设太高，否则 ES 集群压力会突然增大。

第二，Checkpoint 间隔。默认不配置时，Flink 的 Checkpoint 间隔通常是无限大，也就是不触发。这对于 CDC 任务是不能接受的，因为 CDC 需要把 binlog 位点持久化。建议生产环境execution.checkpointing.interval设为 30 到 60 秒，既能及时保存位点，又不会因为太频繁而影响性能。

第三，状态 TTL。如果你只需要同步数据，不需要做复杂的 join 或聚合，状态一般不长，可以设置table.exec.state.ttl，防止状态无限膨胀。比如：

SET 'table.exec.state.ttl' = '1h';

6.2 常见问题速查表

这个表格是我做技术支持时最常被问到的问题集合，基本上照着排查能解决 80% 的故障。

6.3 数据校验与监控告警是生产底线

最后想多说一句，实时同步做出来只是第一步，能证明数据一直是对的才是关键。我见过不少团队同步链路搭完就再也不管了，直到业务方反馈搜索数据不对才想起来查，这时候往往已经丢了几个小时甚至几天的数据。

我的做法是，每天凌晨对 MySQL 和 ES 的核心索引做一次全量比对，逻辑不复杂：按主键分段拉取，比对数量、校验和的哈希值。跑完只输出不一致的清单。这样如果白天链路出了问题，第二天早上一定能发现，而且能根据不一致的时间点反推故障发生的时间窗口。

监控层面，除了 Flink Web UI，最好把 Checkpoint 的成功率、任务重启次数、ES 写入延迟这几个指标接入告警。尤其是 Checkpoint 成功率，一旦超过一段时间连续失败，就要立刻介入，因为这意味着任务虽然还在跑，但已经从故障点"恢复"不了了。

提示：做数据校验时，别只比对 count，因为行数一样不代表数据对。最好对每一行的关键字段做 hash，比如把主键、状态、金额拼接后计算 MD5，再和 ES 侧同样方式计算，比对结果才可靠。

7. 整个流程做完，我的个人体会

这套 MySQL 到 ES 的 Flink CDC 同步链路，前前后后我做了不下十次，踩过的坑远比文章里写的多。要说最深的体会，其实是：Flink CDC 大大降低了实时同步的门槛，但门槛降低不等于不需要理解原理。你依然得懂 binlog 的格式、懂主键语义、懂 ES 的写入模型，否则出了问题连日志都看不懂。

另外一个让我反复吃亏的点是：demo 跑通了，并不意味着生产没问题。demo 数据量小、无并发、无故障，你能看到的效果都很好；一旦上了生产，binlog 量大、ES 写入压力大、网络抖动、磁盘满了，各种边界情况都会冒出来。所以我会建议每个准备上这个方案的团队，先拿一张真实的业务大表做一次压测，把前后的数据校验脚本准备好，再切真实流量。

如果你现在正准备做类似同步，我的建议是从简单开始：先搭一条非核心业务表，跑个一两天观察稳定性，确认没有数据丢失和延迟问题后，再逐步扩展到其他表。不要一上来就追求"几十张表一锅端"，Flink CDC 任务数量多起来之后，资源占用和故障排查的复杂度是指数增长的，稳扎稳打比追求炫技重要得多。