企业官网建设流程全解析

1. 为什么还在用 Logstash 做 MySQL 到 Elasticsearch 的同步

先把结论摆在前面：MySQL 8 到 Elasticsearch 的数据同步，Logstash 依然是目前最稳、最通用、最容易落地的方案之一。不是因为它最新，而是因为它足够成熟、插件生态完整、容错机制清晰，出了问题你知道去哪里查。

我做过好几个搜索类项目，数据源清一色是 MySQL，目标端是 Elasticsearch，中间同步层换过几种方案，最后还是回到 Logstash。原因很直接：业务表结构经常变，字段增减频繁，用代码写同步逻辑意味着每次改表都要改代码、重新发版；而 Logstash 的配置文件改几行就能适配，运维成本低得多。

这个方案适合谁？如果你手上有 MySQL 8 的业务库，需要把数据同步到 Elasticsearch 做全文检索、聚合分析或者日志查询，又不想引入 Kafka、Flink 这类重型组件，那 Logstash 的 JDBC input 插件配合 Elasticsearch output 插件就是最直接的路径。整条链路是：Logstash 定时轮询 MySQL，按条件拉取增量数据，经过 filter 清洗转换，批量写入 Elasticsearch。

听起来简单，但真正跑起来，坑集中在几个地方：MySQL 8 的驱动兼容性、增量字段的选择、时区处理、批量写入的性能调优、以及删除和更新怎么同步。下面我按实际搭建的顺序，把这些一个个拆开讲。

2. 整体方案设计与核心思路拆解

2.1 为什么选 Logstash 而不是其他同步方式

同步 MySQL 到 Elasticsearch，市面上常见的有几条路：自己写代码用定时任务拉、用 Canal 解析 binlog、用 Logstash 的 JDBC 插件、或者用 DataX 这类数据集成工具。我选 Logstash 主要基于三个判断。

第一，对业务库侵入小。Logstash 走的是标准 SQL 查询，不需要开启 binlog 的特殊权限，也不需要装额外的中间件。对于很多中小团队来说，DBA 不一定愿意给你开 binlog 订阅权限，但给你一个只读账号是没问题的。

第二，配置即代码，改起来快。字段映射、类型转换、条件过滤，全在配置文件里完成。业务加了一个字段，我在 filter 里加一段 mutate 就行，不用重新编译打包。

第三，批量写入和重试机制内置。Logstash 的 Elasticsearch output 自带批量提交、失败重试、死信队列，这些如果自己写代码，光处理网络抖动和部分失败就够头疼的。

当然它也有短板：实时性依赖轮询间隔，最快也就秒级，做不到 binlog 那种毫秒级；删除操作不好同步，因为 JDBC 查询只能查到存在的记录，被删掉的行你查不到。这两个问题后面我会给出实际的应对办法。

2.2 增量同步的核心：追踪字段怎么选

整个方案的心脏是增量字段。Logstash 的 JDBC input 插件支持tracking_column参数，它会记录上一次查询到的最大值，下次从这个值往后查。选哪个字段当追踪列，直接决定了同步的准确性和性能。

常见的候选有三个：自增主键id、更新时间update_time、以及时间戳字段。我的经验是优先用update_time，但必须配合id做兜底。为什么？因为自增 id 只能捕获新增，捕获不了更新；而update_time能同时覆盖新增和更新，前提是每次数据变更都要刷新这个字段。

这里有个细节很多人踩坑：update_time如果是datetime类型，精度只到秒，同一秒内有多条变更时，tracking_column只记录一个值，可能漏数据。解决办法是把字段改成datetime(3)或者timestamp(3)，精确到毫秒。如果表结构改不了，那就用id和update_time组合，或者干脆用id做追踪列，接受更新延迟。

提示：追踪列的值必须是单调递增的。如果业务里update_time会被手动改小，或者存在时区回拨，增量同步就会丢数据。上线前一定要确认这个字段的写入逻辑。

2.3 全量与增量的衔接策略

第一次同步肯定是全量，之后才是增量。Logstash 的 JDBC 插件用sql_last_value这个变量来区分：第一次运行时它等于tracking_column的初始值（默认 1970-01-01），之后每次等于上次查到的最大值。

我的做法是首次全量单独跑一次，把sql_last_value的初始值设成业务最早的时间，让它把历史数据全部拉一遍。全量跑完之后，插件会把最后的值写进一个元数据文件（last_run_metadata_path），下次启动就自动从那个点开始增量。这个元数据文件一定要持久化，别放在临时目录，否则重启后又会全量拉一遍。

3. 环境准备与关键配置细节

3.1 MySQL 8 驱动兼容性这个坑

MySQL 8 和 5.7 最大的区别是默认认证插件从mysql_native_password换成了caching_sha2_password。Logstash 自带的 JDBC 驱动版本如果太老，连 MySQL 8 会直接报认证失败。我实测下来，驱动版本至少要用 8.0.28 以上，最好用 8.0.33。

驱动 jar 包要放到 Logstash 的指定目录，通常是logstash-core/lib/jars/下面，或者你在配置文件里用jdbc_driver_library显式指定路径。我倾向于显式指定，这样升级驱动不影响其他管道。

连接串也有讲究。MySQL 8 的连接 URL 建议加上这几个参数：

jdbc:mysql://数据库地址:3306/库名?useSSL=false&serverTimezone=Asia/Shanghai&characterEncoding=utf8&useCursorFetch=true&rewriteBatchedStatements=true

逐个解释：serverTimezone必须设，否则时间字段会差 8 小时；useCursorFetch=true配合useCursorFetch让大结果集分批取，避免全量时内存爆掉；rewriteBatchedStatements=true是写入优化的关键，后面性能部分细说。

3.2 账号权限最小化

给 Logstash 用的 MySQL 账号，只需要SELECT权限就够了，不要图省事给 root。如果涉及多张表，就针对具体库授权。这样即使配置文件泄露，风险也可控。

CREATE USER 'sync_reader'@'%' IDENTIFIED BY '强密码'; GRANT SELECT ON 业务库.* TO 'sync_reader'@'%'; FLUSH PRIVILEGES;

3.3 Elasticsearch 端的准备

Elasticsearch 这边要提前建好索引和 mapping。不要依赖动态 mapping，因为 MySQL 的字段类型和 ES 的类型不是一一对应的。比如 MySQL 的decimal到 ES 应该是scaled_float或者double，datetime到 ES 应该是date并指定格式。如果让 ES 自动推断，很可能把本该是数字的字段识别成 text，后面聚合就废了。

我的习惯是先手动建索引，把关键字段的类型、分词器、是否索引都定好。比如商品名称用text加ik分词器，商品 ID 用keyword，价格用scaled_float并设scaling_factor为 100。建好之后再让 Logstash 往里写，类型就稳了。

4. 完整实操流程与配置实现

4.1 管道配置文件逐段拆解

下面是一份我实际在用的配置骨架，以一张商品表为例。先看 input 部分：

input { jdbc { jdbc_driver_library => "/path/to/mysql-connector-j-8.0.33.jar" jdbc_driver_class => "com.mysql.cj.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/shop?useSSL=false&serverTimezone=Asia/Shanghai&characterEncoding=utf8&useCursorFetch=true&rewriteBatchedStatements=true" jdbc_user => "sync_reader" jdbc_password => "强密码" schedule => "*/30 * * * * *" statement => "SELECT id, name, price, stock, category_id, update_time FROM product WHERE update_time > :sql_last_value AND update_time <= NOW() ORDER BY update_time ASC" use_column_value => true tracking_column => "update_time" tracking_column_type => "timestamp" last_run_metadata_path => "/data/logstash/meta/product_last_value" clean_run => false jdbc_paging_enabled => true jdbc_page_size => 5000 } }

几个关键点：schedule用的是 cron 表达式，*/30 * * * * *表示每 30 秒跑一次，六个字段分别对应秒分时日月周。use_column_value => true配合tracking_column才能用字段值做追踪，否则默认用时间戳。clean_run => false保证重启后从元数据文件继续，而不是从头来。

statement里的update_time <= NOW()是个保险，防止查询期间有新数据写入导致边界混乱。ORDER BY update_time ASC保证按顺序处理，配合追踪列不会跳数据。

4.2 filter 段的清洗与类型转换

从 MySQL 出来的数据，字段类型和 ES 期望的往往对不上，filter 段就是干这个的：

filter { mutate { convert => { "price" => "float" "stock" => "integer" "category_id" => "integer" } remove_field => ["@version", "@timestamp"] } date { match => ["update_time", "yyyy-MM-dd HH:mm:ss.SSS", "yyyy-MM-dd HH:mm:ss"] target => "update_time" timezone => "Asia/Shanghai" } if [name] { mutate { strip => ["name"] } } }

convert把字符串转成数字类型，这一步不做的话，ES 里存的就是字符串，范围查询和排序都会出问题。date插件把时间字符串转成 ES 认识的日期格式，timezone一定要指定，否则又差 8 小时。remove_field去掉 Logstash 自动加的元字段，保持文档干净。

注意：date插件的match数组里可以写多个格式，它会依次尝试。MySQL 的datetime(3)出来是带毫秒的，datetime不带，两个格式都写上就不会解析失败。

4.3 output 段的批量写入配置

output { elasticsearch { hosts => ["http://es-node1:9200", "http://es-node2:9200"] index => "product_index" document_id => "%{id}" document_type => "_doc" manage_template => false flush_size => 2000 idle_flush_time => 5 retry_on_conflict => 3 } stdout { codec => rubydebug } }

document_id => "%{id}"用 MySQL 主键作为 ES 文档 ID，这样同一条记录更新时会覆盖而不是新增，天然实现 upsert。flush_size是攒够多少条就批量提交，idle_flush_time是即使没攒够也每隔几秒提交一次，两个配合平衡吞吐和延迟。retry_on_conflict处理并发写冲突。

调试阶段把stdout打开，能看到每条文档的实际结构，确认字段和类型都对。上线后关掉，否则日志量太大。

4.4 启动与验证

配置写好后，用-f指定配置文件启动，加--config.test_and_exit先做语法检查：

bin/logstash -f /path/to/mysql_to_es.conf --config.test_and_exit bin/logstash -f /path/to/mysql_to_es.conf

验证分三步：先看 Logstash 日志有没有报错，再看 ES 里文档数量对不对，最后抽查几条文档的字段值。我一般用_count接口对数量，用_search抽几条看内容：

curl -s "http://es-node1:9200/product_index/_count" curl -s "http://es-node1:9200/product_index/_search?size=3&pretty"

数量对不上，先查是不是有字段解析失败被丢弃，再看追踪列的边界有没有问题。

5. 性能调优与常见问题排查

5.1 批量写入的性能参数怎么调

同步慢，八成卡在写入。几个关键参数我按重要性排：flush_size、jdbc_page_size、pipeline.workers、pipeline.batch.size。

flush_size控制 ES 批量提交的条数，默认 500，我一般调到 2000 到 5000。太大内存压力大，太小网络往返多。jdbc_page_size控制每次从 MySQL 取多少行，配合useCursorFetch用，5000 是个比较稳的值。pipeline.workers是 Logstash 的并行工作线程数，默认等于 CPU 核数，如果 ES 端扛得住可以适当调大。

还有一个容易被忽略的：MySQL 连接串里的rewriteBatchedStatements=true。这个参数对写入 MySQL 有用，但对读 MySQL 影响不大，真正影响读取的是useCursorFetch。别搞混了。

5.2 常见问题速查表

5.3 删除操作怎么同步

这是 JDBC 方案的天然短板。MySQL 里删了一行，Logstash 查不到这行，自然也不会去删 ES 里的文档。我的处理办法有两种。

一种是软删除。业务表加一个is_deleted字段，删除时置 1 而不是物理删除。Logstash 查询时把is_deleted也拉过来，在 filter 里判断，如果为 1 就在 ES 里标记删除或者直接删文档。这样追踪列update_time也会更新，增量能捕获到。

另一种是定期对账。写一个独立的脚本，定期比对 MySQL 和 ES 的 ID 集合，把 ES 里多出来的文档删掉。这个脚本不用频繁跑，一天一次就够，作为兜底。

提示：软删除方案需要在业务代码里配合，如果业务不允许改表，那就只能靠对账脚本。两种方案我都用过，软删除实时性好但侵入业务，对账脚本无侵入但有延迟，按团队情况选。

5.4 我踩过的几个真实坑

第一个坑是元数据文件权限。Logstash 进程的用户如果没有元数据目录的写权限，sql_last_value存不下来，每次重启都全量。这个错误日志里不一定明显，我是查了半天才发现目录属主不对。

第二个坑是追踪列被业务改小。有次业务做数据修复，把一批记录的update_time手动改成了更早的时间，结果这些记录再也没被同步过。后来我在对账脚本里加了一个检查，定期比对两边数量，发现偏差就告警。

第三个坑是ES 索引 mapping 冲突。一开始图省事用了动态 mapping，结果某个字段第一批数据是数字，第二批来了个字符串，ES 直接拒绝写入。后来老老实实手动建 mapping，再没出过这个问题。

6. 进阶玩法与扩展思路

6.1 多表同步怎么组织

一个 Logstash 实例可以跑多个管道，每个管道对应一张表或者一组表。我习惯按业务域拆分，比如商品一个管道、订单一个管道、用户一个管道，各自独立的追踪列和元数据文件。这样一张表出问题不影响其他表，排查也清晰。

如果表特别多，可以用jdbc插件的多语句模式，但我不推荐，因为追踪列不好统一管理。宁可多写几个配置文件，用pipelines.yml统一管理，清晰可控。

6.2 关联数据的处理

有时候 ES 里的文档需要多表关联的结果，比如商品要带上分类名称。有两种做法：一是在 SQL 里直接 join，把结果一次性查出来，简单直接；二是分管道同步，在 ES 端做关联查询，但 ES 不擅长 join，性能差。

我倾向第一种，SQL join 在 MySQL 端完成，Logstash 只负责搬运。join 的字段记得都放进SELECT，filter 里做好类型转换。如果 join 的表数据也会变，那追踪列要取所有相关表update_time的最大值，这个逻辑稍微复杂点，可以在 SQL 里用GREATEST函数处理。

6.3 监控与告警

同步链路跑起来之后，监控不能少。我关注三个指标：同步延迟（当前时间减去最后一条文档的 update_time）、文档数量偏差（MySQL 和 ES 的数量差）、Logstash 错误日志。

延迟和偏差可以用定时脚本采集，推到监控系统。Logstash 本身的日志要配置轮转，别让它把磁盘写满。这几个监控搭起来之后，同步出问题能第一时间发现，不用等业务方来投诉。

整套方案我从头搭到尾大概花了两天，其中一半时间在调时区和驱动兼容。跑稳之后基本不用管，偶尔业务加字段改一下 filter 就行。如果你也在做类似的同步，建议先把追踪列和时区这两个点确认清楚，能省掉后面一大半的排查时间。