先交代背景：前两篇我们把ElasticSearch集群的线上配置和日常运行讲完了，数据能写进去了，但真正要支撑搜索业务，还得把业务库里的数据稳定地同步到ES里。这篇系列三就专门讲线上环境里怎么用LogStash把RDS（以MySQL为例）的数据同步到ElasticSearch，覆盖全量同步、增量同步、定时调度、参数调优和线上踩坑记录。

文章适合谁看？如果你手头是千万到上亿行的业务表，搜索、聚合查询需要走ES，实时性要求是分钟级而不是秒级，那LogStash这套"SQL游标增量拉取+幂等写入"的方案足够稳定，而且排查问题非常方便。如果业务要求秒级同步，那得考虑基于binlog的Canal或Flink CDC，不在本篇讨论范围。以下所有配置均来自我线上跑了一年多的实际项目，可以直接抄。

1. 同步方案怎么选：为什么最后定的是LogStash

1.1 先想清楚同步到底在解决什么问题

把RDS数据同步到ES，表面上是"数据复制"，本质是两类诉求。第一类是检索：MySQL的LIKE '%关键词%'对中文分词不友好，数据量一大就全表扫描，CPU打满也未必扫完。第二类是查询分析：多条件组合过滤、聚合统计、时序分析，这些正是ES的强项。但ES不是业务库的替代品，它就是个"查询副本"，所以同步方案的核心要求其实是三个：数据要全、更新要及时、写入要够快。

我刚接这个需求时，脑子里冒出来的方案至少四五个，但我给自己定了条原则：不在检索场景里引入一套复杂到没人敢动的数据管道。很多团队一上来就上CDC全家桶，组件一多、链路一长，出问题连数据到底在哪一段丢的都不知道。所以当时我的判断是，量级和实时性没到必须上流式框架的时候，优先选最成熟、最容易排查的同步工具。

1.2 主流方案横向对比

这里把当时对比过的几个代表方案摆出来，可以直接拿去当选型参考。

最后选择LogStash的理由很朴素：配置驱动、不用写代码、一个进程就能跑、日志和报错直观。Canal虽然能秒级同步，但为了秒级延迟要付出Server高可用、binlog位点管理、消费端幂等等一堆成本，对多数检索类业务来说是过度设计。DataX做增量不顺手，每次调度还要依赖外部任务系统。自研脚本看着灵活，实际把分页查询、断点续传、连接回收、异常重试都写一遍，工程量比想象中大得多。

1.3 LogStash的能力边界要心里有数

LogStash的本质是"数据加工管道"，input-filter-output三段式设计非常清晰，但这不代表它能干所有事。用久了你会发现几个明显的边界。第一，它无法感知MySQL里的物理删除，因为JDBC输入只能从表里读到"还在的数据"，删掉的行走的是另一条路，后面我会讲怎么处理。第二，它不是真正的实时，cron的最小粒度是1分钟，即便你写成*/1 * * * * *，一轮查询加写入的耗时也会让延迟进一步放大。第三，filter阶段做复杂清洗会拖慢吞吐，能用SQL完成的事别挪到filter里做。

有边界不可怕，怕的是方案选错。正确姿势是：常态增量用LogStash跑，发现数据差异大时用"全量重建+别名切换"兜底，删除场景业务侧用逻辑删除标记。这套组合我在线上稳定跑了一年多，基本没出过需要半夜爬起来修数据的问题。

2. 部署准备与版本配套

2.1 版本选型：LogStash和ES的版本别差太远

LogStash 7.x要配ES 7.x，8.x配8.x，大版本必须一致，小版本尽量一致。我线上用的是7.10.2配套ES 7.10.2，JDBC输入插件和ES输出插件的版本都随主版本走。这里要特别提醒：ES 8.x默认开启了安全认证，LogStash输出端要额外配置user、password，hosts还要换成https，和7.x的配置方式有差异，网上大量老教程都是7.x的写法，直接照搬会在8.x上报出认证失败。

注意：下载LogStash时看清配套的ES大版本。曾经见过有人用LogStash 7.x往ES 6.x写数据，报action/metadata line is malformed之类的错，排查半天最后发现是版本协议不兼容。

2.2 安装LogStash与MySQL驱动准备

安装本身不复杂，解压即用。以Linux为例，把安装包解压到/opt/logstash，然后确认两件事。一是JAVA_HOME环境变量要指向JDK（8或11都行，看LogStash版本要求），二是下载对应版本的MySQL JDBC驱动jar包放到一个固定目录，比如/opt/logstash/lib/mysql-connector-java-8.0.20.jar。注意8.x驱动的类名是com.mysql.cj.jdbc.Driver，老5.x驱动才是com.mysql.jdbc.Driver，网上很多教程用后者，照抄会报ClassNotFound。

装好后先做个冒烟测试。执行bin/logstash -e 'input { stdin {} } output { stdout {} }'，启动后输入一行字符，如果能原样回显，说明装好了、JVM也正常。接着用bin/logstash --config.test_and_exit -f your.conf校验配置文件语法，这个命令不会真正启动管道，只做配置检查和插件加载验证，上线前一定要跑一遍。

2.3 JVM堆内存和系统参数

LogStash默认JVM堆只有1GB，跑JDBC同步加ES输出，强烈建议至少给2GB，最好4GB。修改config/jvm.options里的-Xms和-Xmx，两个值设成一样，避免运行期堆扩容抖动。我线上的机器和ES集群是分开部署的，这台机器内存32GB，给了LogStash 6GB堆，JDBC分页拉数和ES批量写入都够用。

另一个容易忽略的是文件描述符上限。LogStash会同时维持到MySQL的连接和到ES各节点的连接，系统默认的1024很容易被打满，表现为连接报错、日志里刷Too many open files。改/etc/security/limits.conf，把nofile调到65535，重启进程生效。这几个参数现在不配好，后面数据量起来再改就要重启中断同步，成本更高。

3. 配置文件逐参数拆解：JDBC输入与ES输出

3.1 input段：jdbc插件的核心字段

这节是全文核心，先把一份线上真实在跑的配置贴出来，然后逐段拆解。

input { jdbc { jdbc_driver_library => "/opt/logstash/lib/mysql-connector-java-8.0.20.jar" jdbc_driver_class => "com.mysql.cj.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://your-rds-host:3306/business_db?serverTimezone=Asia/Shanghai&useSSL=false&characterEncoding=utf8&rewriteBatchedStatements=true" jdbc_user => "sync_user" jdbc_password => "your_password" jdbc_paging_enabled => true jdbc_page_size => 10000 use_column_value => true tracking_column => "update_time" tracking_column_type => "timestamp" statement => "SELECT id, title, author, content, status, create_time, update_time FROM article WHERE update_time >= :sql_last_value AND update_time < NOW() ORDER BY update_time ASC, id ASC" schedule => "*/1 * * * * *" last_run_metadata_path => "/opt/logstash/data/last_run_article" clean_run => false } }

逐项说。

jdbc_connection_string里的serverTimezone=Asia/Shanghai一定要加，它决定了MySQL驱动把DATETIME转成Java时间对象时的时区基准，漏掉的话时间字段很容易差8小时。useSSL=false内网传输没问题，省去证书配置。characterEncoding=utf8防止中文乱码。rewriteBatchedStatements=true开启JDBC批处理重写，分页拉数性能提升明显，这个参数很多人忽略，但对全量同步影响很大。

jdbc_paging_enabled => true和jdbc_page_size => 10000组合使用，避免一次查询把几百万行全部装进内存。分页的原理是驱动按LIMIT分块拉取，每拉一页发给下游处理。page_size的取值取决于单行大小，如果单行包含大字段（比如正文几千字），建议降到5000，防止单页过大撑爆内存。

use_column_value => true和tracking_column => "update_time"用来指定增量游标字段。LogStash每次跑完查询都会记录这个字段的最大值，下一轮执行时以:sql_last_value的形式传回SQL里。tracking_column_type => "timestamp"表示游标是时间类型，首轮默认值是1970-01-01 00:00:00；如果表主键是自增id，也可以改成tracking_column => "id"配tracking_column_type => "numeric"，首轮默认值是0。两种游标各有适用场景，时间游标能捕获更新记录，id游标查询更高效但抓不到修改，后面会展开讲。

statement里的SQL是增量同步的精髓，两个关键点。一是:sql_last_value作为起始过滤条件，只拉上次游标之后的新增和更新。二是update_time < NOW()这个上限条件，用途有两个：防止应用写入"未来时间"的脏数据被同步进来；避免同一轮里刚被更新过的行在下一轮因为时间边界问题再次被扫到，减少无效重复。ORDER BY update_time ASC, id ASC保证分页顺序稳定，分页状态下没有稳定排序容易导致翻页错位，丢数据。

schedule => "*/1 * * * * *"是Quartz的6段cron表达式，表示每秒的0秒触发一次，实际就是每分钟跑一轮。这里有个非常实用的技巧：如果暂时不想定时，只做一次性全量同步，就把这行注释掉。LogStash会执行完statement里的查询然后自然退出，不会进入常驻等待。这个特性用来做全量初迁非常方便。

last_run_metadata_path指定游标持久化文件路径。LogStash每轮跑完会把sql_last_value写进这个文件，下次启动时读取。这个文件的妙用后面讲"定点重放"时会专门说。

clean_run => false意思是保留游标文件继续增量。如果某天发现数据差了很多，想重新全量拉一遍，把clean_run改成true并重启，LogStash会忽略游标文件从头执行statement。跑完一定记得改回false，否则下次重启还会全量。

提示：密码别明文写在conf里。用环境变量引用，input里写jdbc_password => "${JDBC_PWD}"，启动前export一下，避免密码泄露到版本库里。

3.2 filter段：不是必须，但别完全忽略

JDBC同步场景里，filter最常见的写法是这个：

filter { mutate { remove_field => ["@version", "@timestamp"] } }

这个操作的实际意义不只是"让数据看着干净"。@timestamp是LogStash的处理时间，不是业务时间，如果后面在Kibana做时序分析、按天分片，默认会拿@timestamp当时间字段，很容易得出错误结论。所以我倾向于把业务时间字段（比如update_time、create_time）显式保留，在ES mapping里定义成date类型，然后去掉@timestamp，避免混淆。

filter里还能做数据脱敏（手机号、身份证掩码）、字段重命名、类型转换（mutate { convert => [...] }）。但记住一个原则：filter越重，同步速率越低。能用SQL完成的数据处理尽量在SQL里做。比如字段拼接、大小写转换、NULL值兜底，都直接写进SELECT语句，比在filter里跑Ruby正则快一个量级。

3.3 output段：ES输出与幂等写入

output { elasticsearch { hosts => ["http://10.0.1.11:9200", "http://10.0.1.12:9200"] index => "article_v1" document_id => "%{id}" action => "index" } # stdout { codec => json_lines } # 排障时取消注释 }

hosts填多个节点就行，其中一个不可用时LogStash会自动切到其他节点，少写一个不影响写入。index是目标索引名，注意ES 7.x之后不支持一个索引下多个type，这里直接指索引。document_id和action => "index"是幂等写入的核心：document_id指向业务主键，同一行数据重复同步只会覆盖更新，不会产生重复文档。如果表有联合主键，就用字段拼接生成唯一ID，比如"%{user_id}_%{article_id}"。

关于批量大小要专门说一下。新版LogStash的ES输出插件里已经不推荐配置bulk和flush_size了，批量写入节奏由Logstash管道的pipeline.batch.size控制，在config/logstash.yml里配置：

pipeline: batch: size: 1000 delay: 50

我线上设为1000，单条文档几十KB以内这个值比较均衡。设太大虽然吞吐高，但单批内存峰值和ES端的bulk队列压力同时上涨；设太小则写入QPS上不去。增量场景每分钟一般也就几十到几千条，这个批量值完全够用。pipeline.batch.delay是凑批等待时间，单位毫秒，50是默认值，没有特殊需求不用动。

ES 8.x开了安全认证的话，output段要改成：

output { elasticsearch { hosts => ["https://es-node:9200"] user => "elastic" password => "你的密码" ssl_certificate_verification => false index => "article_v1" document_id => "%{id}" } }

4. 从全量到增量的完整落地流程

4.1 第一步：先建索引mapping再同步

很多人上来就启动LogStash让ES自动推断mapping，结果数字变成了text、日期格式解析失败、keyword被拆词，后期改mapping只能重建索引。正确做法是同步前先手动建好mapping。

以文章表为例，一份简单的mapping长这样：

PUT /article_v1 { "mappings": { "properties": { "id": { "type": "keyword" }, "title": { "type": "text", "analyzer": "ik_max_word" }, "author": { "type": "keyword" }, "content": { "type": "text", "analyzer": "ik_max_word" }, "status": { "type": "integer" }, "create_time": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss" }, "update_time": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss" } } } }

几个注意点。ID字段用keyword不用long，因为ES里文档ID本质是字符串，keyword类型和document_id => "%{id}"配合最顺畅。标题和正文用text类型并指定分词器，检索需求如果有中文分词就用IK分词器；如果只做精确匹配和过滤，字段类型统一用keyword更省空间。日期字段必须显式声明format，且要和MySQL里格式化出来的字符串格式完全一致，否则日期解析会报错。

4.2 第二步：全量同步要注意的几个验证动作

建好mapping后，开始第一次全量同步。配置上做三件事：把schedule注释掉、clean_run改成true、打开output里的stdout注释。然后启动：

bin/logstash -f /opt/logstash/conf/article_sync.conf --config.test_and_exit

配置文件校验通过后，正式启动。头一次跑全量，日志里会看到Paging statement之类的JDBC分页输出，每拉完一页会往ES写一批。全量同步期间建议盯着两个指标：日志里有没有报错，以及ES的写入速率。数据量大时全量同步可能要跑几十分钟甚至几小时，期间不要动游标文件。

全量跑完后，验证数据对不对，按以下顺序来：

1. 用/_count接口看索引总文档数：curl http://es-node:9200/article_v1/_count
2. MySQL里执行SELECT COUNT(*) FROM article对比总数。
3. 抽查几条记录：GET /article_v1/_doc/具体id，核对关键字段值。
4. 验证日期字段：搜索近几天的数据，看时间偏移是否正常。

如果总数对不上，最常见的原因是同步过程中有业务数据在持续写入，导致两边永远差一点，这不代表同步有问题，启动增量后会自动追上。如果差很多，优先查日志里的报错，而不是急着改配置重跑。

4.3 第三步：从全量切到增量

全量验证通过后，开始增量改造，就三步。第一，把schedule => "*/1 * * * * *"加回来。第二，clean_run改回false，让LogStash从游标文件继续。第三，确认last_run_metadata_path文件里有全量同步结束时的游标值。这三步做完，重启进程，增量就生效了。

验证增量是否正常，最直接的办法是到MySQL里update一条已有记录，等1分钟以内，然后到ES里查这条记录，看字段是否更新。注意验证时要改一条update_time在游标值之后的记录，否则同步不到。再验证新增，往MySQL插入一条新数据，同样等一分钟后到ES里确认能搜到。

提示：全量同步结束后，游标文件里保存的是全量过程中扫到的最大update_time。如果全量期间MySQL还在不断写入，这些新数据可能已经同步过，也可能因为时间边界刚好卡在游标附近而漏掉。稳妥做法是在全量完成、切增量前，观察下一轮增量日志里有没有把游标附近的数据补进来。正常情况下，第一轮增量会扫到游标之后所有"还没被扫过"的数据。

4.4 用systemd守护进程，别用nohup裸跑

增量同步上线后，进程必须被守护。我吃过nohup的亏，机器重启后忘了手动拉起，第二天才发现数据同步停了十几个小时。现在统一用systemd管理，unit文件如下：

[Unit] Description=Logstash sync article After=network.target [Service] Type=simple User=logstash ExecStart=/opt/logstash/bin/logstash -f /opt/logstash/conf/article_sync.conf Restart=always RestartSec=10 StandardOutput=append:/var/log/logstash/article_sync.log StandardError=append:/var/log/logstash/article_sync.log [Install] WantedBy=multi-user.target

Restart=always保证进程异常退出后10秒自动拉起，日志统一重定向到固定文件。多个同步任务就写多个unit和多个conf文件，每个任务独立的游标文件、独立的日志，排查问题互不干扰。一个LogStash进程里也可以写多个jdbc input块，但我更推荐一个进程只跑一个同步任务，因为一旦某个input的SQL写错卡死，同一个管道里的其他任务也会被拖累。隔离进程虽然多占一点内存，但换来了故障隔离，很值。

5. 线上高频问题与排查记录

5.1 数据对不上：丢数据、重复数据、延迟越来越大

这是最常被问的一类问题。先说丢数据。LogStash增量同步丢数据，90%是SQL的游标边界写错了。用update_time >= :sql_last_value时，如果同一秒内有大量数据更新，两个轮次之间可能出现游标值恰好等于某些记录的update_time，这些记录理论上会被下一轮再扫一次，不丢；真正丢数据的是漏了ORDER BY加上游标字段不稳定。全量阶段就已经因为排序不稳定丢数据的话，增量阶段也一样丢。所以前面的SQL里强行要求ORDER BY update_time ASC, id ASC，就是为了让分页边界的顺序确定下来。

再说重复数据。如果document_id配了业务主键，重复同步只会覆盖不会新增文档，这种"伪重复"无害。真正需要警惕的是索引里文档ID设计不合理导致的重复。比如联合主键只取了其中一个字段，两个不同业务主键映射成了同一个ES文档ID，互相覆盖，表现为"数据越同步越少"。遇到这种，回头检查document_id的拼接规则。

还有同步延迟越来越大的情况。增量轮询每分钟一轮，正常情况下每轮查询只要处理"这一分钟内变化的数据"，几秒内跑完。如果发现日志里每轮执行时间从几秒涨到几十秒，甚至下一轮开始时上一轮还没结束，说明statement里的查询条件没起到过滤作用。最常见的错误是把update_time < NOW()写成update_time <= NOW()，或者游标字段没走索引。检查MySQL里update_time字段上有没有索引，没有索引的话增量查询会全表扫描，分钟级任务很快就把数据库拖垮。

5.2 时区错乱和日期类型问题

时间字段差8小时，是JDBC同步的高频问题。原因有三个层级，逐层排查。第一层，MySQL连接串有没有serverTimezone=Asia/Shanghai，没有的话驱动用JVM默认时区解析，可能偏差。第二层，MySQL服务端时区本身，执行SELECT NOW()看返回时间是否正常。第三层，ES内部统一存储UTC时间，Kibana展示时按浏览器时区换算。如果你在Kibana里看到的时间总是差8小时，检查Kibana的时区设置，而不是怀疑同步有问题。

还有个容易踩的坑是日期字段的format不匹配。MySQL的DATETIME格式化出来默认是2024-01-15 10:30:00，但如果代码里配了date_format之类的设置，出来可能是2024-01-15T10:30:00。ES mapping里声明了"format": "yyyy-MM-dd HH:mm:ss"，遇到带T的字符串就会解析失败，整条文档写入报错。排查方法很简单，打开stdout输出看原始事件里日期字段长什么样，照着改mapping里的format就行。

5.3 内存溢出、连接数打满、ES bulk拒绝

LogStash进程OOM，日志里出现OutOfMemoryError，优先查三处。一是jvm.options的堆设置，对照同步数据量，4GB以下基本都偏小；二是jdbc_page_size，单行数据大时一页10000行可能直接把堆撑爆，降到2000到5000再观察；三是pipeline.batch.size，它决定管道内待处理事件上限，设置过大同样会放大内存压力。

连接数打满的表现是MySQL日志里出现Too many connections，ES日志里出现rejected execution。前者通常是LogStash的JDBC连接没有及时释放，在网络闪断时会累积；后者是ES端bulk队列积压，说明写入速率超过集群处理能力。处理思路不一样：连接数问题靠重启LogStash进程快速恢复，然后检查MySQL连接串里有没有配autoReconnect=true；ES bulk拒绝则要先看集群的CPU和load，负载高就调大pipeline.batch.delay降低写入速率，负载不高就调大bulk队列参数。

我在线上还遇到过一个隐蔽问题：LogStash读了MySQL的SQL日志后，发现每轮查询虽然只返回几条数据，但MySQL端实际扫描的行数很大。后来定位到是游标字段update_time的索引因为数据量大了之后区分度下降，优化SQL在末尾加上AND id > 0这种无用条件并不解决，最终方案是换成双游标策略，用id做主游标、update_time做辅助过滤，这个问题才彻底解决。双游标属于进阶玩法，如果单游标已经稳定，不建议一开始就上。

6. 线上稳定运行的一些经验沉淀

6.1 游标文件的定点重放技巧

last_run_metadata_path这个游标文件平时不起眼，但关键时刻能救命。比如某天运营说"昨晚10点到11点之间的数据在ES里查询结果不对"，排查后发现是那段时间业务代码有bug，写进MySQL的数据本身有问题，现在数据在MySQL里已经修好了，但ES里还是坏数据。

最直接的办法是把游标文件改到昨晚10点之前的时间点，然后重启LogStash，让它从那个点重新拉一遍。具体操作是停掉进程、编辑游标文件的值、启动进程。游标文件保存的格式就是上次游标值的字面文本，时间型的可以手动写成一个合法的时间字符串，比如2024-06-20 21:59:59，numeric型的写数字就行。重启后第一轮增量就会从指定点重新扫描，逐轮追上最新状态。

这个技巧比重新全量同步效率高得多，也避开了"全量期间业务还在写数据"的对账难题。用过几次之后，我现在每次动游标文件都先备份原文件，改坏了还能恢复。

6.2 索引别名与定期重建兜底

LogStash的JDBC方案有一个天然缺陷：同步不了MySQL里的物理删除。如果业务经常删数据，ES里会残留大量"幽灵文档"，搜索时还能搜出来。应对方案有两种。

第一种是业务侧配合，删数据时把status字段改成已删除状态，同步和查询都过滤掉status=已删除。这是最干净的做法，业务改动也不大。第二种是定时重建索引。给索引挂一个别名，日常查询都用别名：

POST /_aliases { "actions": [ { "add": { "index": "article_v1", "alias": "article_search" } } ] }

定期（比如每周）新建一个索引article_v2，把MySQL全量导入进去，然后原子切换别名：remove旧索引、add新索引，一步完成。切换期间查询不中断，这是ES的原子别名操作带来的最大好处。切换完成后删除旧索引，释放空间。

我在线上是"增量实时同步 + 每周全量重建"双轨并行。增量保证日常数据分钟级可见，全量重建兜底解决删除残留和游标累计偏差。每周重建时选业务低峰期执行，切换别名那几秒内的并发查询也完全无感。

6.3 监控和告警，别等业务来投诉

同步链路最怕的就是静默失败——进程还在跑，但数据已经不更新了。所以监控必须做，而且要做三层。

第一层是进程层。systemd的Restart=always保证进程挂了能拉起，但拉起后如果配置错误反复崩溃，还是要靠日志监控。用定时任务检查进程存活，挂了就告警。

第二层是数据新鲜度。这个最实用：在ES里维护一条带固定ID的"心跳文档"，LogStash的增量SQL里加上查询心跳表的逻辑，每轮同步自动更新这条文档的update_time。外部监控定时去读这条文档，发现update_time超过5分钟没更新，说明同步链路断了，立刻告警。这个方法不依赖任何外部监控组件，一个curl请求就能检查。

第三层是数据一致性对账。每天凌晨业务低峰期，对MySQL和ES做一次总量对比。不需要精确到每行，看总数差就行。偏差超过阈值就触发告警，然后再决定是定点重放还是全量重建。这里要提醒，对账SQL本身会消耗数据库资源，务必安排在低峰期，并加上合理的查询条件避免全表扫描。

到这里，LogStash同步RDS到ES的完整方案就讲完了。我个人实际运营下来最大的体会是：这套方案不炫技，但胜在稳定和可控，出任何问题都能在半小时内定位到具体环节，因为链路只有"SQL查出来→LogStash处理→ES写进去"三段，每一段都有日志可查。最后再多说一句，很多团队把同步方案做得极其复杂，反而忽略了数据本身的可观测性。先把监控和对账做好，再用最简单的工具跑通，比一上来就追求秒级实时要稳妥得多。数据链路稳了，业务才敢放心把搜索流量切到ES上。