先交代背景前两篇我们把ElasticSearch集群的线上配置和日常运行讲完了数据能写进去了但真正要支撑搜索业务还得把业务库里的数据稳定地同步到ES里。这篇系列三就专门讲线上环境里怎么用LogStash把RDS以MySQL为例的数据同步到ElasticSearch覆盖全量同步、增量同步、定时调度、参数调优和线上踩坑记录。文章适合谁看如果你手头是千万到上亿行的业务表搜索、聚合查询需要走ES实时性要求是分钟级而不是秒级那LogStash这套SQL游标增量拉取幂等写入的方案足够稳定而且排查问题非常方便。如果业务要求秒级同步那得考虑基于binlog的Canal或Flink CDC不在本篇讨论范围。以下所有配置均来自我线上跑了一年多的实际项目可以直接抄。1. 同步方案怎么选为什么最后定的是LogStash1.1 先想清楚同步到底在解决什么问题把RDS数据同步到ES表面上是数据复制本质是两类诉求。第一类是检索MySQL的LIKE %关键词%对中文分词不友好数据量一大就全表扫描CPU打满也未必扫完。第二类是查询分析多条件组合过滤、聚合统计、时序分析这些正是ES的强项。但ES不是业务库的替代品它就是个查询副本所以同步方案的核心要求其实是三个数据要全、更新要及时、写入要够快。我刚接这个需求时脑子里冒出来的方案至少四五个但我给自己定了条原则不在检索场景里引入一套复杂到没人敢动的数据管道。很多团队一上来就上CDC全家桶组件一多、链路一长出问题连数据到底在哪一段丢的都不知道。所以当时我的判断是量级和实时性没到必须上流式框架的时候优先选最成熟、最容易排查的同步工具。1.2 主流方案横向对比这里把当时对比过的几个代表方案摆出来可以直接拿去当选型参考。方案实时性增量机制维护成本适合场景LogStash JDBC输入分钟级SQL游标定时轮询低单节点配置驱动千万到亿级、分钟级延迟可接受Canal秒级MySQL binlog订阅高需额外维护Server和消费端秒级实时、需要捕获删除操作DataX离线全量为主增量较弱中依赖外部调度大批量离线全量迁移自研脚本取决于实现自己写高分页、重试、并发全要自己扛定制化极强且有人长期维护最后选择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.x8.x配8.x大版本必须一致小版本尽量一致。我线上用的是7.10.2配套ES 7.10.2JDBC输入插件和ES输出插件的版本都随主版本走。这里要特别提醒ES 8.x默认开启了安全认证LogStash输出端要额外配置user、passwordhosts还要换成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环境变量要指向JDK8或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?serverTimezoneAsia/ShanghaiuseSSLfalsecharacterEncodingutf8rewriteBatchedStatementstrue 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里的serverTimezoneAsia/Shanghai一定要加它决定了MySQL驱动把DATETIME转成Java时间对象时的时区基准漏掉的话时间字段很容易差8小时。useSSLfalse内网传输没问题省去证书配置。characterEncodingutf8防止中文乱码。rewriteBatchedStatementstrue开启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的写入速率。数据量大时全量同步可能要跑几十分钟甚至几小时期间不要动游标文件。全量跑完后验证数据对不对按以下顺序来用/_count接口看索引总文档数curl http://es-node:9200/article_v1/_countMySQL里执行SELECT COUNT(*) FROM article对比总数。抽查几条记录GET /article_v1/_doc/具体id核对关键字段值。验证日期字段搜索近几天的数据看时间偏移是否正常。如果总数对不上最常见的原因是同步过程中有业务数据在持续写入导致两边永远差一点这不代表同步有问题启动增量后会自动追上。如果差很多优先查日志里的报错而不是急着改配置重跑。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] DescriptionLogstash sync article Afternetwork.target [Service] Typesimple Userlogstash ExecStart/opt/logstash/bin/logstash -f /opt/logstash/conf/article_sync.conf Restartalways RestartSec10 StandardOutputappend:/var/log/logstash/article_sync.log StandardErrorappend:/var/log/logstash/article_sync.log [Install] WantedBymulti-user.targetRestartalways保证进程异常退出后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连接串有没有serverTimezoneAsia/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 connectionsES日志里出现rejected execution。前者通常是LogStash的JDBC连接没有及时释放在网络闪断时会累积后者是ES端bulk队列积压说明写入速率超过集群处理能力。处理思路不一样连接数问题靠重启LogStash进程快速恢复然后检查MySQL连接串里有没有配autoReconnecttrueES 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:59numeric型的写数字就行。重启后第一轮增量就会从指定点重新扫描逐轮追上最新状态。这个技巧比重新全量同步效率高得多也避开了全量期间业务还在写数据的对账难题。用过几次之后我现在每次动游标文件都先备份原文件改坏了还能恢复。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的Restartalways保证进程挂了能拉起但拉起后如果配置错误反复崩溃还是要靠日志监控。用定时任务检查进程存活挂了就告警。第二层是数据新鲜度。这个最实用在ES里维护一条带固定ID的心跳文档LogStash的增量SQL里加上查询心跳表的逻辑每轮同步自动更新这条文档的update_time。外部监控定时去读这条文档发现update_time超过5分钟没更新说明同步链路断了立刻告警。这个方法不依赖任何外部监控组件一个curl请求就能检查。第三层是数据一致性对账。每天凌晨业务低峰期对MySQL和ES做一次总量对比。不需要精确到每行看总数差就行。偏差超过阈值就触发告警然后再决定是定点重放还是全量重建。这里要提醒对账SQL本身会消耗数据库资源务必安排在低峰期并加上合理的查询条件避免全表扫描。到这里LogStash同步RDS到ES的完整方案就讲完了。我个人实际运营下来最大的体会是这套方案不炫技但胜在稳定和可控出任何问题都能在半小时内定位到具体环节因为链路只有SQL查出来→LogStash处理→ES写进去三段每一段都有日志可查。最后再多说一句很多团队把同步方案做得极其复杂反而忽略了数据本身的可观测性。先把监控和对账做好再用最简单的工具跑通比一上来就追求秒级实时要稳妥得多。数据链路稳了业务才敢放心把搜索流量切到ES上。