线上业务的数据量一旦上来搜索和查询的痛点就会集中爆发。我这个系列写到第三篇前两篇折腾的是ElasticSearch集群本身的配置和调优这一篇要解决的是上游数据怎么稳定地流进索引——也就是用LogStash把RDS里的数据同步到ElasticSearch。虽然LogStash常被当成日志收集器看但在我这边的线上工程里它更多是扮演数据库和搜索引擎之间的搬运工定时从RDS读增量数据做轻量清洗再写入ES索引。这篇文章适合正在搭同步链路、被增量更新搞到头大、或者刚准备把自研同步脚本替换掉的读者。我会把配置逐段拆开讲也会把那些文档里不会写、但线上跑几天就一定会遇到的坑摊开说清楚。前两篇已经解决了ES的骨架问题这篇就是给骨架供血的部分。1. 同步方案选型为什么是LogStash而不是更实时的Canal1.1 常见候选方案的横向对比在正式碰配置之前我建议先想明白一个问题同步MySQL到ElasticSearch市面上方案不少凭什么选LogStash我把当时评估的几个候选摆出来逐个说过。第一个候选是自研定时脚本。听起来最简单写一个Java或者Python程序每分钟查一次MySQL把变更数据写进ES。但这种方案有几个隐性成本自己要处理连接池、重试、幂等写入、同步状态记录还要保证脚本进程不挂。数据量小的时候无所谓数据量一上来脚本会因为各种边界情况变得非常脆弱。尤其是断点续传如果程序在查出数据之后、写入ES之前崩了重启之后怎么知道上次同步到哪了这些逻辑写起来远比想象中复杂。第二个候选是监听binlog的实时同步组件比如Canal这类方案。它走的是MySQL主从复制协议能拿到几乎实时的数据变更事件实时性是最好的。但代价是要额外部署一个独立的同步服务还要在RDS上开启binlog并申请相关权限。在托管数据库环境里开binlog很多时候意味着要走变更流程线上管控比较严的团队这一关就很难过。而且binlog方案一旦出现位点偏移、解析报错排查复杂度也比LogStash高一截。第三个候选就是LogStash。它算准实时方案默认情况下可以做到分钟级延迟对于绝大多数搜索场景完全够用。它最大的优势是同属于Elastic生态输入输出插件都是现成的jdbc_input负责读数据库elasticsearch_output负责写索引配置是声明式的不需要写业务代码。团队只要有人会写配置文件就能维护整条链路。1.2 线上工程的实际取舍逻辑我把三个方案放一起列个表方便对照着看对比项自研脚本binlog监听组件LogStash实时性取决于轮询频率秒级分钟级可调部署复杂度低高需独立服务低和LogStash共用环境维护成本高需自研断点逻辑中高解析复杂低配置驱动对RDS依赖需要读权限需要开binlog只需要读权限生态契合度一般一般原生支持ES我们最终选LogStash核心原因有三个。第一搜索场景能容忍分钟级延迟不需要秒级实时第二托管RDS开binlog的权限流程太长而LogStash只要求一个SELECT权限的账号第三团队本来就维护着ES集群LogStash作为同一生态的工具监控、排错、升级路径都很清晰。这里我想强调一个观点方案选型不是选最强的而是选跟团队运维能力最匹配的。一个需要专门写代码维护的同步模块远不如一份稳定的配置文件来得省心。2. 环境准备期最容易翻车的三件事2.1 JDK版本与LogStash版本强绑定LogStash不同版本对JDK的要求差异很大。以我们现在用的8.x为例它要求JDK 17起步而老一点的7.xJDK 8或11就能跑。这个坑我在测试环境踩过一次机器上装了多个JDK系统里JAVA_HOME还指向老版本LogStash启动脚本检测完版本直接报错退出。报错信息要么是Unsupported Java version要么是启动到一半抛ClassFormatError。排查手段很简单先在命令行跑一下java -version确认当前shell拿到的JDK版本再看LogStash安装目录下的bin/logstash脚本里有没有显式设置JAVA_HOME。最稳妥的办法是把JAVA_HOME写死在启动脚本里不要让系统环境变量替你做决定。另外注意LogStash自带的JDK检测只看JAVA_HOME不看PATH所以很多人改了PATH发现没用就是因为这个。2.2 MySQL驱动JAR的获取方式LogStash的jdbc_input插件不会自带MySQL驱动这一点和直觉相反——很多人以为装上插件就能连数据库了结果启动日志里出现java.lang.ClassNotFoundException: com.mysql.cj.jdbc.Driver人直接懵掉。正确做法是单独下载MySQL Connector/J驱动包放在一个固定的目录比如/opt/logstash/lib/下然后在管道配置里通过jdbc_driver_library参数指定绝对路径。这里有一个版本细节如果是MySQL 8.x驱动类名是com.mysql.cj.jdbc.Driver不再是老版本的com.mysql.jdbc.Driver。很多从老教程抄配置的人第一反应是驱动下载错了其实只是类名没跟着版本升级。另外驱动版本尽量选和RDS实例版本匹配或更高的版本。老驱动连新数据库可能出现认证插件不兼容、时区报错等问题。如果RDS是5.7用8.x驱动连完全没有问题因为驱动向后兼容做得很好。2.3 RDS白名单、账号权限与连接串线上RDS几乎都有IP白名单限制LogStash所在机器的公网或者内网IP必须先加进白名单否则连接串再对也是一直超时。之后是数据库账号我强烈建议单独创建同步专用账号不要用主账号。给最小权限就够了通常只需要SELECT如果某些表要做全量比对再加一个对staging表的读权限。例如CREATE USER logstash_sync% IDENTIFIED BY 这里写一个强密码; GRANT SELECT ON biz_db.* TO logstash_sync%; FLUSH PRIVILEGES;创建完账号连接串还有一处经常被忽略时区参数。MySQL 8.x的连接串必须显式设置serverTimezone否则驱动会拿本地时区去解析DATETIME后面同步到ES的时间很可能差8小时。我当时在连接串里写的是jdbc:mysql://rm-bp1x3x****.mysql.example.com:3306/biz_db?serverTimezoneAsia/ShanghaiuseSSLfalsecharacterEncodingutf8useSSLfalse是内网环境的标准配置省去SSL握手开销characterEncodingutf8确保中文不会乱码。这两项看似基础缺一个都会让你在数据同步之后花很长时间定位为什么数据看起来不对劲。3. 核心管道拆解从JDBC读MySQL到写入ES的每个参数3.1 input段JDBC轮询参数的含义环境准备好之后就看核心配置了。LogStash的管道配置文件通常是/etc/logstash/conf.d/*.conf这种形式一个管道负责一条同步链路。以文章表articles为例我给出一个精简但完整的input段input { jdbc { jdbc_driver_library /opt/logstash/lib/mysql-connector-j-8.2.0.jar jdbc_driver_class com.mysql.cj.jdbc.Driver jdbc_connection_string jdbc:mysql://rm-bp1x3x****.mysql.example.com:3306/biz_db?serverTimezoneAsia/ShanghaiuseSSLfalsecharacterEncodingutf8 jdbc_user logstash_sync jdbc_password 你的密码 schedule */5 * * * * statement SELECT id, title, content, category_id, status, updated_at FROM articles WHERE updated_at :sql_last_value ORDER BY updated_at ASC use_column_value true tracking_column updated_at tracking_column_type timestamp last_run_metadata_path /opt/logstash/data/.last_run_articles clean_run false } }这里面的参数每一个都有讲究。schedule用的是cron表达式*/5 * * * *表示每5分钟跑一次。这种轮询方式决定了同步是准实时的如果你把表达式改成* * * * *可以做到每秒钟轮询一次但会给RDS增加不必要的压力一般不建议。尤其线上RDS还有主从延迟轮询太频繁容易读到滞后数据。statement是你真正要执行的SQL。WHERE updated_at :sql_last_value是增量同步的核心——:sql_last_value这个变量就是上一次同步结束时LogStash记住的位置。ORDER BY updated_at ASC也很关键能让每次查询结果按时间顺序排列防止因数据乱序导致ES里同一文档被频繁覆盖。还有一点增量SQL里的updated_at列必须走索引否则扫描全表的查询在数据量大之后会直接拖垮RDS。这个我可以明确说线上每次轮询都是真实执行的SQL别看是增量查询计划不走索引照样能把数据库打满。use_column_value true表示启用列值追踪再配合tracking_column指定追踪列。这里有个版本差异LogStash 6.x之前用的是use_incremental_last_run参数7.x之后改成了use_column_value。如果你查老资料看到的是前者说明教程已经过时了。tracking_column_type指定追踪列的类型有两个可选值numeric和timestamp。MySQL的updated_at是DATETIME就填timestamp。last_run_metadata_path是LogStash保存同步状态的文件路径。这个文件里存着上次的sql_last_value值进程重启后不会从头扫。clean_run false的意思是不要每次启动都清空这个状态如果你设成true每次重启LogStash都会全量跑一遍线上一般没人这么干。3.2 filter段清洗要克制别在这里堆逻辑filter段是很多人喜欢放飞自我的地方。我的建议是filter要克制尽量只做轻量处理复杂的业务逻辑不要塞进LogStash。因为每个filter事件都会消耗JVM内存线上数据量一大花里胡哨的filter往往是OOM的元凶之一。我通常只做三件事去掉不需要的字段、把时间字符串规整成ES能正确解析的格式、以及必要的类型转换。示例filter { mutate { remove_field [version, category_name_raw] } date { match [published_at, yyyy-MM-dd HH:mm:ss] target timestamp } }这里说明一下MySQL的DATETIME经过JDBC读取后LogStash一般会转成字符串。如果不做date处理ES会把所有时间字段当字符串后面想按时间范围查询就得用range条件强转非常难受。所以在filter里指定原始格式让它解析进timestamp是最省事的做法。类型转换也要注意。MySQL的decimal类型在JDBC驱动读出来之后有可能被映射成浮点数而浮点数会损失精度。对金额、价格这类字段我习惯在filter里先转成字符串保证写入ES后依然是完整精度。另外如果源表里存的是JSON字符串ES不会自动帮你解析成嵌套结构需要额外加json过滤器比如json { source extra_info target extra_info_parsed }要记住LogStash的定位是管道不是业务引擎。filter里塞的规则越多线上排查问题就越困难。一个事件进了管道你很难直观判断到底是哪个filter改坏了数据。保持管道简单数据问题更容易定位。3.3 output段document_id与模板覆盖的坑output段是整个管道的出口直接决定数据怎么落进ES。我的配置output { elasticsearch { hosts [http://10.0.0.5:9200, http://10.0.0.6:9200] index articles_%{YYYY.MM.dd} document_id %{id} manage_template false user elastic_user password es_password } }document_id必须设置一般取业务主键。这个参数的价值在于幂等写入如果运行过程中同一条数据被同步了两次ES会按相同ID覆盖不会产生重复文档。很多人第一次配LogStash不写document_id结果索引里的文档数比业务表的行数还多就是漏了这行配置。manage_template false这一步我要强调一下。默认情况下LogStash会给索引自动应用一套内置模板但它的模板对中文分词、日期格式的预设不一定合适。线上通常已经用ES的模板API预先建好了业务索引模板比如给标题字段配IK分词给时间字段指定date类型。这时候LogStash再插手管模板反而可能把字段类型覆盖成它认为的类型导致业务查询出问题。稳妥的做法是模板在ES侧管理LogStash只管写入。如果确实想用LogStash生成索引就把template_overwrite打开但那样要非常小心mapping冲突。关于多ES节点的hosts配置这里不需要写全所有节点写两三个即可。ES节点会通过集群状态把请求转发到正确分片不会因为只写了两个节点就出现单点瓶颈。4. 增量同步的边界追踪列、时间窗口与数据一致性取舍4.1 追踪列不是万能的一定会漏的几种情况用updated_at做增量追踪几乎是LogStash同步MySQL的标准姿势但它有一个天然盲区它只能感知到那一行在查询时刻的字段值感知不到数据库内部发生的所有变更。具体来说至少有三类数据变更这种方案同步不到。第一类物理删除。业务系统如果直接DELETE掉一行LogStash的轮询SQL根本看不到这行数据ES里也就还留着那条已删除的文档。第二类软删除且不刷新时间戳。有些业务开发习惯只改status字段但忘记顺带更新updated_at那么这批数据就没进入下一轮查询。第三类历史数据回刷。运营或数据修复脚本偶尔会把老行直接改成历史时间哪怕它今天改动了updated_at可能是几天前甚至去年的值增量查询同样抓不到。这个问题的本质是以查询条件判断增量的方案本质上依赖业务表里必须维护一个靠谱的更新时间戳。所以我在上线前会和业务方做一次约定所有要同步的表必须满足两个条件一是要有稳定主键二是每次业务更新必须刷新updated_at。这个约定写进研发规范里能规避掉90%的增量漏同步问题。对于物理删除这种硬伤更实际的做法是看业务容忍度如果删除场景不频繁可以让ES侧保留文档并标记状态或者接受一段时间的脏数据。4.2 全量增量混合一周一次重建兜底既然增量追踪有盲区线上工程就不会只依赖增量。我目前采用的策略是增量同步常驻 每周一次全量重建。全量重建的做法不是直接删索引再导数据那样查询侧会有一段时间完全空索引。更稳妥的是建一个新的日期后缀索引比如articles_20240915同步全量数据进去等数据对齐后通过ES的别名机制把查询切到新索引然后再删掉旧索引。整个过程对业务查询几乎无感。在LogStash这边全量重建通常用单独一个管道配置关键设置是last_run_metadata_path指向一个临时文件并且启动前需要清空这个状态文件或者直接把clean_run设为true强制从最早位置开始。全量SQL就没有sql_last_value条件了直接SELECT id, ... FROM articles数据量大时分批跑每次加一个主键范围过滤避免一次性取全部记录把内存打爆。这种混合策略的代价是每周多跑一次全量但对数据一致性提供了一个兜底。增量同步好比日常蓄水全量重建是定期清淤两者配合搜索索引里基本不会出现长期残留的脏数据。4.3 时区与字段映射的隐性坑前面在连接串部分提过时区这里再展开讲一次因为它通常不是启动时报错而是莫名其妙晚8小时才被发现。MySQL的DATETIME不带时区信息JDBC驱动会拿serverTimezone指定的时区去理解它。如果你不指定驱动往往取JVM默认时区而LogStash所在服务器的系统时区不一定和业务时区一致。我们当时就遇到过一次业务同学说凌晨发布的文章ES里搜不到最后发现是同步过来的时间整体被写成了UTC时间比本地时间少了8小时按业务时间检索自然就错位了。排查时要注意一个细节在LogStash的filter里date插件解析出来的时间会转成UTC存进ES而ES的timestamp字段本身就是UTC存储。此刻你看到的差8小时不一定真是错了可能是显示层没有做时区转换。正确的判断方法是看业务时间字段本身的值是否正确而不是看timestamp默认显示。如果业务字段本身被驱动解析错了再调ES侧的时区设置也没用只能在连接串层面根治。字段映射上还有一个容易踩的坑MySQL的JSON列或者TEXT列很大时如果ES侧没有预定义mapping首次写入会自动推断字段类型长文本可能被映射成带keyword子字段的text类型导致后续聚合排序行为和你预期不一致。这也就是为什么我前面强调预建模板索引的mapping应该在ES侧定义好LogStash只负责把数据放进去不要让自动推断替你长期决定字段语义。5. 上线之后的事多管道管理、内存与同步健康观测5.1 多表同步用pipelines.yml做管道隔离线上不可能只同步一张表。商品是一张表文章是一张表用户数据可能又是另一张表。如果把这些同步逻辑全部塞进一个管道出问题时很难定位而且一张表SQL卡住整条管道的其他同步也会跟着排队。LogStash的多管道配置在config/pipelines.yml里每个管道独立指定配置文件和资源参数。我们线上的大概长这样- pipeline.id: article_sync path.config: /etc/logstash/conf.d/article_sync.conf pipeline.workers: 2 - pipeline.id: product_sync path.config: /etc/logstash/conf.d/product_sync.conf pipeline.workers: 2 - pipeline.id: user_sync path.config: /etc/logstash/conf.d/user_sync.conf pipeline.workers: 1管道隔离的好处是明显的某个管道配置写错了只会导致它自己不断重启其他管道不受影响某个同步表数据量特别大可以单独给它分配更多workers。但这也有代价每增加一个管道就多一份JVM线程和资源消耗不是所有表都有资格独立开管道。我们内部的做法是核心业务表单独管道小表可以合并进一个批次轮询管道而不是每张表都开一个。5.2 JVM堆与批量参数避免线上OOMLogStash默认的JVM堆只有1G这在本地玩一玩没问题线上跑真实业务量就太紧张了。我们第一次上线时就吃过这个亏同步全量历史数据时LogStash进程直接因为java.lang.OutOfMemoryError: Java heap space挂掉日志里一片红色。调整方式是改LogStash安装目录下的config/jvm.options-Xms4g -Xmx4g具体给多少要结合机器内存。一般来说LogStash堆不要超过机器物理内存的50%因为它还有堆外内存和线程开销。如果机器是8G内存给4G比较合理如果机器16G给6到8G也可以。改完之后要重启进程才能生效这步很容易漏掉——配置改了进程没重启等于白改。批量写入参数同样重要。默认情况下LogStash的pipeline.batch.size是125pipeline.batch.delay是50毫秒。同步大数据量时这个默认值偏保守写入ES的效率不够高。我们线上通常把batch.size调到1000batch.delay保持默认。这里要给一个提醒batch.size不能无脑调大ES端每个bulk队列有自己的处理上限如果LogStash一次性丢过去的数据太多ES会返回RejectedExecutionException或者429 too many requests反而造成写入堵塞。建议从500开始逐步上调观察ES写入响应时间没有明显恶化再继续加。5.3 三个观测点LogStash还是ES还是RDS数据同步是三条链路的事RDS产生数据、LogStash搬运、ES接收。出问题时第一步不是看配置而是先通过观测手段确认问题在哪一段。我把线上常用的观测手段列出来按排查顺序排。先说LogStash自身的观测。LogStash会监听一个本地HTTP端口默认9600通过API可以看到运行状态和事件处理量curl http://127.0.0.1:9600/_node/stats重点关注events下面的几个字段in代表累计进入管道的事件数out代表累计输出成功的事件数filtered代表经过过滤的事件数。如果in在持续增长但out长时间不动说明数据卡在输出端——大概率是ES写入被拒或者网络抖动。如果in一直不变说明LogStash根本没从RDS读到数据问题就在数据库那一段。再说ES侧观测。最直观的是看索引文档数是否随时间增长curl http://10.0.0.5:9200/_cat/indices/articles*?vsindex如果索引文档数连续几个周期都没有变化而业务库明明有新数据那问题多半出在增量条件上。这时候可以手动执行一下LogStash的statement对应的SQL看查询结果是否真的返回了数据。最后再看RDS侧慢查询日志LogStash的轮询SQL有没有变成慢SQL可以直接判断是不是索引失效了。我总结一下典型的排查链路供参考发现ES里数据少了先看LogStash的events.out是否增长再开ES的索引文档数最后回到RDS执行一次原始SQL。一般来说三步走完就能定位到是查询条件写错、输出被拒还是网络问题。别一上来就怀疑配置配置不会自己变化但数据量增长后之前能跑的查询可能会因为索引失效而变慢这种问题不观测很难发现。这个系列到这里ES的集群、数据管道、同步链路就完整了。如果上面的配置一次性都配好线上跑起来会发现LogStash其实相当安静——它会持续工作日志里不报错指标正常增长。最需要警惕的反而是看起来一切正常但数据就是不对的时刻增量时间字段没更新、模板被覆盖、时区错位这些都不在报错日志里只能靠对同步机制的理解去觉察。按我现在的习惯任何一张新表接入同步我都会先在测试环境跑一遍全量加增量重点核对三个东西文档数是否和源表一致、最新一条改动多久能出现在ES里、业务时间字段是否准确。这三关过了再上生产我才有底。