1. 为什么还在用 Logstash 做 MySQL 到 Elasticsearch 的同步先把结论摆在前面MySQL 8 到 Elasticsearch 的数据同步方案其实不少比如 Canal、Flink CDC、DataX、Debezium 等等但 Logstash 依然是中小规模场景里最稳、最快能跑起来的那一个。原因很简单——它自带 JDBC Input 插件配置文件写几行就能跑不需要额外部署中间件不需要啃 Binlog 协议也不需要维护一套流处理集群。对于数据量在百万级到千万级、实时性要求是分钟级而不是毫秒级的业务Logstash 的性价比几乎没有对手。这个项目的核心目标很明确把 MySQL 8 里的业务数据通过 Logstash 的 JDBC Input 插件定时拉取经过字段清洗和类型转换后批量写入 Elasticsearch让搜索、聚合、报表这些场景能直接查 ES 而不用压主库。适合谁来参考我总结了三类人一是刚接触 ES 同步、想找一个能当天跑通的方案的后端开发二是手上有一堆 MySQL 报表查询拖慢主库、想迁到 ES 的运维三是做日志或商品搜索、需要把关系型数据变成文档模型的工程师。但这里有个前提必须说清楚Logstash 的 JDBC Input 是基于时间戳或自增 ID 的轮询拉取不是真正的 CDC 实时捕获。这意味着它天然有延迟也天然会漏掉物理删除。理解这一点后面所有的配置取舍才有依据。我见过太多人上来就照抄一份配置跑起来发现删掉的数据还在 ES 里然后回头骂 Logstash 不行——其实是一开始就没搞清楚它的工作模型。下面我会从整体设计、核心细节、实操落地、问题排查四个维度把这条链路完整拆一遍包括我踩过的坑和最后稳定跑了大半年的配置。2. 整体方案设计与关键取舍2.1 同步链路的分层思路一条完整的同步链路我习惯把它拆成四层来看这样出问题的时候能快速定位是哪一层的事数据源层MySQL 8负责产生数据。这里的关键是表结构设计必须有可靠的时间戳字段或自增主键否则增量同步无从谈起。采集层Logstash 的 JDBC Input负责按 SQL 拉数据。它本质是一个定时任务每隔一段时间执行一次查询。处理层Logstash 的 Filter负责字段重命名、类型转换、时间格式化、字段增删。这一层决定了 ES 里的文档长什么样。写入层Logstash 的 Elasticsearch Output负责批量 bulk 写入包括索引命名、文档 ID 指定、失败重试。这四层里最容易出问题的是采集层和处理层。采集层的坑在于增量字段选得不对处理层的坑在于类型没转干净导致 ES 映射冲突。2.2 为什么选时间戳增量而不是全量覆盖全量同步的逻辑最简单每次把整张表查出来覆盖写 ES。但它有两个致命问题。第一数据量一大单次查询会把 MySQL 内存打爆尤其是几百万行的表SELECT *直接拖垮主库。第二每次全量写入 ES网络和磁盘 IO 都是浪费明明只有几条数据变了却要重写全部文档。时间戳增量的逻辑是记录上一次同步到的最大时间下次只查比这个时间大的数据。这样每次拉取的数据量很小对 MySQL 压力也小。代价是需要一个可靠的update_time字段而且这个字段必须在每次数据变更时自动更新。MySQL 8 里可以用ON UPDATE CURRENT_TIMESTAMP来实现这是最省事的做法。注意如果业务代码里用的是手动赋值update_time一定要确认所有写路径都更新了这个字段否则会漏数据。我遇到过某张表部分更新语句忘了带update_time结果那部分数据永远同步不过去。2.3 增量字段的选择时间戳 vs 自增 ID这两种方案我都用过各有适用场景直接上对比对比维度时间戳字段自增 ID能否捕获更新能不能只能捕获新增能否捕获删除不能不能时钟回拨风险有无并发写入乱序可能漏不会漏适用场景有更新需求的表只追加的日志表时间戳方案能捕获更新这是它最大的优势但它的软肋是同一秒内的并发写入可能漏数据。假设同步游标记录的是2024-01-01 10:00:00下次查询用WHERE update_time 2024-01-01 10:00:00如果两条数据在同一秒写入一条在游标更新前被读到另一条在游标更新后才写入但时间戳相同就会被跳过。解决办法有两个一是游标回退几秒比如每次查询用update_time (last_time - 5s)用重复读取换不漏数据反正 ES 写入是幂等的二是用update_time加自增 ID 的组合游标。我一般用第一种简单可靠。2.4 文档 ID 的设计为什么必须显式指定ES 的_id如果不指定会自动生成一个随机字符串。这会导致同一个 MySQL 主键的数据每次同步都生成新文档ES 里出现大量重复。所以必须在 Output 里用document_id指定为 MySQL 的主键值。elasticsearch { hosts [http://es-node:9200] index goods document_id %{id} doc_as_upsert true }doc_as_upsert true的作用是文档存在就更新不存在就插入。这样即使重复同步同一条数据结果也是幂等的不会产生脏数据。这个参数是整条链路幂等性的基石千万别漏。3. 核心配置细节与实操要点3.1 JDBC Input 的关键参数逐个拆先看一份我实际在用的 Input 配置然后逐个参数解释input { jdbc { jdbc_driver_library /usr/share/logstash/lib/mysql-connector-j-8.0.33.jar jdbc_driver_class com.mysql.cj.jdbc.Driver jdbc_connection_string jdbc:mysql://mysql-host:3306/biz_db?useSSLfalseserverTimezoneAsia/ShanghaicharacterEncodingutf8useCursorFetchtrue jdbc_user sync_user jdbc_password your_password schedule */1 * * * * statement_filepath /usr/share/logstash/sql/goods.sql use_column_value true tracking_column update_time tracking_column_type timestamp last_run_metadata_path /usr/share/logstash/data/goods_last_run clean_run false jdbc_fetch_size 1000 jdbc_page_size 100000 } }jdbc_driver_library指向 MySQL 8 的驱动 jar 包。这里有个大坑MySQL 8 必须用mysql-connector-j-8.x版本的驱动老版本的mysql-connector-java-5.x连不上会报认证协议错误。驱动类名也从com.mysql.jdbc.Driver变成了com.mysql.cj.jdbc.Driver写错了直接启动失败。jdbc_connection_string里的参数每一个都有用。useSSLfalse是因为内网环境不需要 SSL开着反而增加握手开销。serverTimezoneAsia/Shanghai必须加否则 MySQL 8 的时区处理会让时间字段偏移 8 小时同步过去的时间全是错的。useCursorFetchtrue配合jdbc_fetch_size使用让驱动用游标分批取数据避免一次性把结果集全加载到内存。schedule是 cron 表达式*/1 * * * *表示每分钟执行一次。这个频率要根据数据量和实时性要求来定。数据量大就拉长到 5 分钟实时性要求高就缩短但别低于 1 分钟否则上一次还没跑完下一次就触发了。use_column_value true和tracking_column update_time是增量同步的核心。前者告诉 Logstash 用字段值而不是时间戳来追踪后者指定追踪字段。tracking_column_type timestamp明确类型避免被当成字符串比较。last_run_metadata_path是游标持久化文件记录上次同步到的值。这个文件一定要挂载到持久化卷上否则容器重启后游标丢失会从头全量拉一遍。3.2 SQL 语句的写法与游标回退技巧SQL 文件的内容大概长这样SELECT id, goods_name, category_id, price, stock, status, create_time, update_time FROM goods WHERE update_time :sql_last_value AND update_time NOW() ORDER BY update_time ASC:sql_last_value是 Logstash 自动替换的占位符值就是上次记录的游标。AND update_time NOW()这个条件很重要它防止查询到未来时间的数据比如业务写入了一条时间戳异常的数据保证游标单调递增。前面提到的游标回退可以这样改WHERE update_time DATE_SUB(:sql_last_value, INTERVAL 5 SECOND)这样每次会重复读取最近 5 秒的数据但因为 ES 写入是幂等的重复数据不会造成问题反而堵住了并发写入的漏数据窗口。5 秒这个值可以根据业务写入频率调整写入越密集回退窗口越大。实操心得回退窗口不是越大越好。如果回退太大每次重复拉取的数据量会显著增加尤其是高频更新的表可能每次都在重复处理大量数据。我的经验是回退窗口设为同步周期的 1/10 左右比如每分钟同步一次回退 5 到 6 秒就够了。3.3 Filter 层的字段清洗与类型转换MySQL 的数据类型和 ES 的字段类型不是一一对应的Filter 层就是做这个翻译工作。常见的转换有这么几类filter { mutate { rename { goods_name goodsName } remove_field [version, timestamp] convert { price float stock integer } } date { match [create_time, yyyy-MM-dd HH:mm:ss] target createTime timezone Asia/Shanghai } }rename把下划线命名改成驼峰这是为了配合前端或 Java 客户端的字段习惯不是必须的看团队约定。remove_field去掉 Logstash 自动加的元字段减少 ES 存储浪费。convert做类型转换。这里要特别注意MySQL 的decimal类型被 JDBC 读出来是字符串如果不转成float或doubleES 里就是text类型没法做范围查询和排序。但float有精度损失金额字段建议用double或者干脆在 ES 里存成scaled_float。date插件把字符串时间转成 ES 的日期类型。timezone必须指定否则会按 UTC 解析导致时间偏移。这个坑我在时区处理上踩过不止一次配置里所有涉及时间的地方都要显式指定时区。3.4 Output 的批量写入与性能调优output { elasticsearch { hosts [http://es-node1:9200, http://es-node2:9200] index goods document_id %{id} doc_as_upsert true manage_template true template_name goods_template template_overwrite true flush_size 2000 idle_flush_time 5 retry_on_conflict 3 } }hosts配多个节点Logstash 会自动做负载均衡和故障转移。flush_size 2000表示攒够 2000 条文档才发一次 bulk 请求这个值直接影响写入吞吐。太小则请求次数多、开销大太大则单次请求体过大、容易超时。我的经验是 1000 到 5000 之间根据文档大小调整文档大的话调小一点。idle_flush_time 5是兜底机制如果 5 秒内没攒够flush_size也会强制发送避免数据卡在缓冲区里迟迟不写。retry_on_conflict 3处理并发更新冲突。当多个同步任务同时更新同一个文档时ES 会返回版本冲突这个参数让它自动重试 3 次。4. 完整实操流程与现场记录4.1 环境准备与依赖安装第一步是准备 Logstash 环境。我用的是官方镜像但官方镜像里不带 MySQL 驱动需要自己挂载进去。目录结构大概这样/opt/logstash-sync/ ├── config/ │ └── goods.conf ├── sql/ │ └── goods.sql ├── lib/ │ └── mysql-connector-j-8.0.33.jar ├── data/ │ └── goods_last_run └── pipelines.ymlpipelines.yml用来管理多个同步任务每个任务一个配置文件- pipeline.id: goods_sync path.config: /opt/logstash-sync/config/goods.conf - pipeline.id: orders_sync path.config: /opt/logstash-sync/config/orders.conf启动命令docker run -d --name logstash-sync \ -v /opt/logstash-sync/config:/usr/share/logstash/config-pipeline \ -v /opt/logstash-sync/sql:/usr/share/logstash/sql \ -v /opt/logstash-sync/lib:/usr/share/logstash/lib \ -v /opt/logstash-sync/data:/usr/share/logstash/data \ docker.elastic.co/logstash/logstash:8.11.0 \ logstash -f /usr/share/logstash/config-pipeline/goods.conf注意data目录必须挂载否则容器重启后last_run_metadata_path丢失会触发全量同步。这个坑我踩过一次一张 300 万行的表重启后全量拉了一遍MySQL 直接告警。4.2 MySQL 侧的准备工作同步账号要单独建只给必要的读权限别用 rootCREATE USER sync_user% IDENTIFIED BY your_password; GRANT SELECT ON biz_db.goods TO sync_user%; FLUSH PRIVILEGES;表结构上update_time字段必须有索引否则每次增量查询都会全表扫描ALTER TABLE goods ADD INDEX idx_update_time (update_time);这个索引是性能的关键。没有它每次同步的WHERE update_time ?都会走全表扫描数据量一大 MySQL 就扛不住。我见过一张 500 万行的表没加这个索引同步任务一跑主库 CPU 直接飙到 90%。4.3 ES 索引模板的预定义在同步开始前最好先手动创建索引模板把字段类型定死。否则让 ES 自动推断第一次写入的文档决定了字段类型后面类型不一致就会报错。PUT _template/goods_template { index_patterns: [goods*], settings: { number_of_shards: 3, number_of_replicas: 1, refresh_interval: 5s }, mappings: { properties: { id: { type: long }, goodsName: { type: text, fields: { keyword: { type: keyword } } }, categoryId: { type: integer }, price: { type: scaled_float, scaling_factor: 100 }, stock: { type: integer }, status: { type: byte }, createTime: { type: date }, updateTime: { type: date } } } }refresh_interval设成 5s 而不是默认的 1s是因为同步场景不需要秒级可见降低刷新频率能显著提升写入吞吐。price用scaled_float加scaling_factor: 100既保证精度又节省空间比double更适合金额。4.4 启动与首次全量验证首次启动时last_run_metadata_path文件不存在Logstash 会把游标当成 1970 年触发全量同步。这时候要盯着日志看docker logs -f logstash-sync正常的话会看到类似这样的输出[INFO ][logstash.inputs.jdbc] (0.5s) SELECT ... FROM goods WHERE update_time 1970-01-01 08:00:00 [INFO ][logstash.outputs.elasticsearch] Successfully bulk indexed 2000 documents全量同步完成后去 ES 里 count 一下和 MySQL 的 count 对比curl -s http://es-node:9200/goods/_count | jq .count两边数量一致说明首次全量没问题。之后每次增量同步日志里会显示实际拉取的行数正常情况下应该是几条到几十条。5. 常见问题排查与避坑实录5.1 时间偏移 8 小时这是最高频的问题。现象是 ES 里的时间和 MySQL 里的差了 8 小时。原因通常是连接串没加serverTimezone或者 Filter 的date插件没指定timezone。两个地方都要检查缺一不可。我一般会在连接串里加serverTimezoneAsia/Shanghai同时在date插件里加timezone Asia/Shanghai双保险。5.2 增量同步漏数据前面讲过同一秒并发写入会漏。解决办法是游标回退。但还有一种漏数据的情况是update_time字段没更新。排查方法是拿一条漏掉的数据去 MySQL 里看它的update_time是不是比实际修改时间早。如果是说明业务代码的更新语句没带这个字段。5.3 ES 映射冲突报错报错信息类似mapper_parsing_exception: failed to parse field [price] of type [float]。原因是 ES 里字段类型和实际写入的数据类型不匹配。比如模板里price是float但某条数据的price是空字符串转换失败。解决办法是在 Filter 里做空值处理filter { mutate { gsub [price, ^$, 0] } }把空字符串替换成 0避免类型转换失败。5.4 同步任务卡死或越来越慢如果发现同步任务执行时间越来越长大概率是游标文件没更新每次都在重复拉取大量数据。检查last_run_metadata_path文件的内容看看游标值是不是在正常递增。另一个可能是 MySQL 的update_time索引失效了用EXPLAIN看一下查询计划。5.5 常见问题速查表问题现象可能原因排查方向解决办法时间差 8 小时时区未指定检查连接串和 date 插件两处都加 Asia/Shanghai数据漏同步并发写入或字段未更新对比 MySQL 和 ES 数据游标回退 检查 update_time映射冲突类型不一致看 ES 报错字段Filter 里做类型转换和空值处理任务越来越慢游标未更新或索引失效看游标文件和 EXPLAIN修复游标 加索引容器重启后全量data 目录未挂载检查挂载配置挂载持久化卷驱动连接失败驱动版本不对看启动日志换 mysql-connector-j-8.x5.6 几个我踩过的坑第一个坑是驱动版本。MySQL 8 默认的认证插件是caching_sha2_password老驱动不支持必须用 8.x 的驱动。这个坑卡了我半天日志里报的是认证失败但实际是驱动太老。第二个坑是clean_run参数。默认是false表示保留游标。如果设成true每次启动都会清空游标从头全量。这个参数千万别乱动我有次调试时改成true忘了改回来结果每次重启都全量同步。第三个坑是jdbc_page_size和jdbc_fetch_size的区别。前者是 JDBC 查询的分页大小后者是游标每次取的行数。两个都设太小会导致频繁的数据库往返设太大又占内存。我的经验是jdbc_fetch_size设 1000jdbc_page_size设 100000兼顾内存和效率。第四个坑是 ES 的refresh_interval。默认 1s同步场景下完全没必要改成 5s 或 30s 能明显提升写入性能。但改完之后查询会有延迟如果业务对实时性要求高就别改。6. 性能调优与扩展思路6.1 批量写入的参数调优写入性能的三个关键参数是flush_size、idle_flush_time和 ES 的refresh_interval。这三个参数是联动的。flush_size越大单次 bulk 请求越大吞吐越高但延迟也越高。idle_flush_time是兜底防止数据卡住。refresh_interval决定数据多久可见。我的调优顺序是先把refresh_interval调到 30s然后把flush_size从 500 逐步往上调每次翻倍观察 ES 的写入延迟和 Logstash 的 CPU。调到某个值后延迟开始明显上升就回退一档。一般 2000 到 5000 是比较舒服的区间。6.2 多表同步的 pipeline 拆分一个 Logstash 实例可以跑多个 pipeline每个 pipeline 同步一张表。这样好处是互不影响一张表出问题不会拖累其他表。但要注意资源隔离pipeline 之间会争抢 CPU 和内存。如果表很多建议拆到多个 Logstash 实例上。pipelines.yml里可以给每个 pipeline 单独配pipeline.workers和pipeline.batch.size- pipeline.id: goods_sync path.config: /opt/logstash-sync/config/goods.conf pipeline.workers: 2 pipeline.batch.size: 500pipeline.workers是工作线程数默认等于 CPU 核数。同步任务大多是 IO 密集型的线程数不用太多2 到 4 个就够。pipeline.batch.size是单次处理的批次大小和flush_size配合使用。6.3 后续可以扩展的方向如果数据量继续增长Logstash 的轮询模式会逐渐吃力。这时候可以考虑几个方向一是引入消息队列做缓冲Logstash 从队列消费而不是直接查 MySQL二是换成基于 Binlog 的 CDC 方案实时性更好三是把同步任务拆成多个实例按数据范围分片。但我要说的是别过早优化。我见过不少团队数据量才几十万行就上 Flink CDC 集群运维成本高得离谱。Logstash 这套方案在千万级数据量下跑得稳稳的真到了瓶颈再换也不迟。最后分享一个我一直在用的小技巧给同步任务加一个监控每次同步完成后把拉取的行数和耗时打到日志里再用日志平台做告警。如果某次同步行数突然暴涨或者耗时异常能第一时间发现。这个监控成本很低但能帮你提前发现很多问题比如某张表突然被批量更新或者游标文件损坏导致全量拉取。 
长河君 · 2026/10/15 13:18:40 · 阅读 1.2k Logstash 实现 MySQL 8 到 Elasticsearch 增量同步实战
1. 为什么还在用 Logstash 做 MySQL 到 Elasticsearch 的同步先把结论摆在前面MySQL 8 到 Elasticsearch 的数据同步方案其实不少比如 Canal、Flink CDC、DataX、Debezium 等等但 Logstash 依然是中小规模场景里最稳、最快能跑起来的那一个。原因很简单——它自带 JDBC Input 插件配置文件写几行就能跑不需要额外部署中间件不需要啃 Binlog 协议也不需要维护一套流处理集群。对于数据量在百万级到千万级、实时性要求是分钟级而不是毫秒级的业务Logstash 的性价比几乎没有对手。这个项目的核心目标很明确把 MySQL 8 里的业务数据通过 Logstash 的 JDBC Input 插件定时拉取经过字段清洗和类型转换后批量写入 Elasticsearch让搜索、聚合、报表这些场景能直接查 ES 而不用压主库。适合谁来参考我总结了三类人一是刚接触 ES 同步、想找一个能当天跑通的方案的后端开发二是手上有一堆 MySQL 报表查询拖慢主库、想迁到 ES 的运维三是做日志或商品搜索、需要把关系型数据变成文档模型的工程师。但这里有个前提必须说清楚Logstash 的 JDBC Input 是基于时间戳或自增 ID 的轮询拉取不是真正的 CDC 实时捕获。这意味着它天然有延迟也天然会漏掉物理删除。理解这一点后面所有的配置取舍才有依据。我见过太多人上来就照抄一份配置跑起来发现删掉的数据还在 ES 里然后回头骂 Logstash 不行——其实是一开始就没搞清楚它的工作模型。下面我会从整体设计、核心细节、实操落地、问题排查四个维度把这条链路完整拆一遍包括我踩过的坑和最后稳定跑了大半年的配置。2. 整体方案设计与关键取舍2.1 同步链路的分层思路一条完整的同步链路我习惯把它拆成四层来看这样出问题的时候能快速定位是哪一层的事数据源层MySQL 8负责产生数据。这里的关键是表结构设计必须有可靠的时间戳字段或自增主键否则增量同步无从谈起。采集层Logstash 的 JDBC Input负责按 SQL 拉数据。它本质是一个定时任务每隔一段时间执行一次查询。处理层Logstash 的 Filter负责字段重命名、类型转换、时间格式化、字段增删。这一层决定了 ES 里的文档长什么样。写入层Logstash 的 Elasticsearch Output负责批量 bulk 写入包括索引命名、文档 ID 指定、失败重试。这四层里最容易出问题的是采集层和处理层。采集层的坑在于增量字段选得不对处理层的坑在于类型没转干净导致 ES 映射冲突。2.2 为什么选时间戳增量而不是全量覆盖全量同步的逻辑最简单每次把整张表查出来覆盖写 ES。但它有两个致命问题。第一数据量一大单次查询会把 MySQL 内存打爆尤其是几百万行的表SELECT *直接拖垮主库。第二每次全量写入 ES网络和磁盘 IO 都是浪费明明只有几条数据变了却要重写全部文档。时间戳增量的逻辑是记录上一次同步到的最大时间下次只查比这个时间大的数据。这样每次拉取的数据量很小对 MySQL 压力也小。代价是需要一个可靠的update_time字段而且这个字段必须在每次数据变更时自动更新。MySQL 8 里可以用ON UPDATE CURRENT_TIMESTAMP来实现这是最省事的做法。注意如果业务代码里用的是手动赋值update_time一定要确认所有写路径都更新了这个字段否则会漏数据。我遇到过某张表部分更新语句忘了带update_time结果那部分数据永远同步不过去。2.3 增量字段的选择时间戳 vs 自增 ID这两种方案我都用过各有适用场景直接上对比对比维度时间戳字段自增 ID能否捕获更新能不能只能捕获新增能否捕获删除不能不能时钟回拨风险有无并发写入乱序可能漏不会漏适用场景有更新需求的表只追加的日志表时间戳方案能捕获更新这是它最大的优势但它的软肋是同一秒内的并发写入可能漏数据。假设同步游标记录的是2024-01-01 10:00:00下次查询用WHERE update_time 2024-01-01 10:00:00如果两条数据在同一秒写入一条在游标更新前被读到另一条在游标更新后才写入但时间戳相同就会被跳过。解决办法有两个一是游标回退几秒比如每次查询用update_time (last_time - 5s)用重复读取换不漏数据反正 ES 写入是幂等的二是用update_time加自增 ID 的组合游标。我一般用第一种简单可靠。2.4 文档 ID 的设计为什么必须显式指定ES 的_id如果不指定会自动生成一个随机字符串。这会导致同一个 MySQL 主键的数据每次同步都生成新文档ES 里出现大量重复。所以必须在 Output 里用document_id指定为 MySQL 的主键值。elasticsearch { hosts [http://es-node:9200] index goods document_id %{id} doc_as_upsert true }doc_as_upsert true的作用是文档存在就更新不存在就插入。这样即使重复同步同一条数据结果也是幂等的不会产生脏数据。这个参数是整条链路幂等性的基石千万别漏。3. 核心配置细节与实操要点3.1 JDBC Input 的关键参数逐个拆先看一份我实际在用的 Input 配置然后逐个参数解释input { jdbc { jdbc_driver_library /usr/share/logstash/lib/mysql-connector-j-8.0.33.jar jdbc_driver_class com.mysql.cj.jdbc.Driver jdbc_connection_string jdbc:mysql://mysql-host:3306/biz_db?useSSLfalseserverTimezoneAsia/ShanghaicharacterEncodingutf8useCursorFetchtrue jdbc_user sync_user jdbc_password your_password schedule */1 * * * * statement_filepath /usr/share/logstash/sql/goods.sql use_column_value true tracking_column update_time tracking_column_type timestamp last_run_metadata_path /usr/share/logstash/data/goods_last_run clean_run false jdbc_fetch_size 1000 jdbc_page_size 100000 } }jdbc_driver_library指向 MySQL 8 的驱动 jar 包。这里有个大坑MySQL 8 必须用mysql-connector-j-8.x版本的驱动老版本的mysql-connector-java-5.x连不上会报认证协议错误。驱动类名也从com.mysql.jdbc.Driver变成了com.mysql.cj.jdbc.Driver写错了直接启动失败。jdbc_connection_string里的参数每一个都有用。useSSLfalse是因为内网环境不需要 SSL开着反而增加握手开销。serverTimezoneAsia/Shanghai必须加否则 MySQL 8 的时区处理会让时间字段偏移 8 小时同步过去的时间全是错的。useCursorFetchtrue配合jdbc_fetch_size使用让驱动用游标分批取数据避免一次性把结果集全加载到内存。schedule是 cron 表达式*/1 * * * *表示每分钟执行一次。这个频率要根据数据量和实时性要求来定。数据量大就拉长到 5 分钟实时性要求高就缩短但别低于 1 分钟否则上一次还没跑完下一次就触发了。use_column_value true和tracking_column update_time是增量同步的核心。前者告诉 Logstash 用字段值而不是时间戳来追踪后者指定追踪字段。tracking_column_type timestamp明确类型避免被当成字符串比较。last_run_metadata_path是游标持久化文件记录上次同步到的值。这个文件一定要挂载到持久化卷上否则容器重启后游标丢失会从头全量拉一遍。3.2 SQL 语句的写法与游标回退技巧SQL 文件的内容大概长这样SELECT id, goods_name, category_id, price, stock, status, create_time, update_time FROM goods WHERE update_time :sql_last_value AND update_time NOW() ORDER BY update_time ASC:sql_last_value是 Logstash 自动替换的占位符值就是上次记录的游标。AND update_time NOW()这个条件很重要它防止查询到未来时间的数据比如业务写入了一条时间戳异常的数据保证游标单调递增。前面提到的游标回退可以这样改WHERE update_time DATE_SUB(:sql_last_value, INTERVAL 5 SECOND)这样每次会重复读取最近 5 秒的数据但因为 ES 写入是幂等的重复数据不会造成问题反而堵住了并发写入的漏数据窗口。5 秒这个值可以根据业务写入频率调整写入越密集回退窗口越大。实操心得回退窗口不是越大越好。如果回退太大每次重复拉取的数据量会显著增加尤其是高频更新的表可能每次都在重复处理大量数据。我的经验是回退窗口设为同步周期的 1/10 左右比如每分钟同步一次回退 5 到 6 秒就够了。3.3 Filter 层的字段清洗与类型转换MySQL 的数据类型和 ES 的字段类型不是一一对应的Filter 层就是做这个翻译工作。常见的转换有这么几类filter { mutate { rename { goods_name goodsName } remove_field [version, timestamp] convert { price float stock integer } } date { match [create_time, yyyy-MM-dd HH:mm:ss] target createTime timezone Asia/Shanghai } }rename把下划线命名改成驼峰这是为了配合前端或 Java 客户端的字段习惯不是必须的看团队约定。remove_field去掉 Logstash 自动加的元字段减少 ES 存储浪费。convert做类型转换。这里要特别注意MySQL 的decimal类型被 JDBC 读出来是字符串如果不转成float或doubleES 里就是text类型没法做范围查询和排序。但float有精度损失金额字段建议用double或者干脆在 ES 里存成scaled_float。date插件把字符串时间转成 ES 的日期类型。timezone必须指定否则会按 UTC 解析导致时间偏移。这个坑我在时区处理上踩过不止一次配置里所有涉及时间的地方都要显式指定时区。3.4 Output 的批量写入与性能调优output { elasticsearch { hosts [http://es-node1:9200, http://es-node2:9200] index goods document_id %{id} doc_as_upsert true manage_template true template_name goods_template template_overwrite true flush_size 2000 idle_flush_time 5 retry_on_conflict 3 } }hosts配多个节点Logstash 会自动做负载均衡和故障转移。flush_size 2000表示攒够 2000 条文档才发一次 bulk 请求这个值直接影响写入吞吐。太小则请求次数多、开销大太大则单次请求体过大、容易超时。我的经验是 1000 到 5000 之间根据文档大小调整文档大的话调小一点。idle_flush_time 5是兜底机制如果 5 秒内没攒够flush_size也会强制发送避免数据卡在缓冲区里迟迟不写。retry_on_conflict 3处理并发更新冲突。当多个同步任务同时更新同一个文档时ES 会返回版本冲突这个参数让它自动重试 3 次。4. 完整实操流程与现场记录4.1 环境准备与依赖安装第一步是准备 Logstash 环境。我用的是官方镜像但官方镜像里不带 MySQL 驱动需要自己挂载进去。目录结构大概这样/opt/logstash-sync/ ├── config/ │ └── goods.conf ├── sql/ │ └── goods.sql ├── lib/ │ └── mysql-connector-j-8.0.33.jar ├── data/ │ └── goods_last_run └── pipelines.ymlpipelines.yml用来管理多个同步任务每个任务一个配置文件- pipeline.id: goods_sync path.config: /opt/logstash-sync/config/goods.conf - pipeline.id: orders_sync path.config: /opt/logstash-sync/config/orders.conf启动命令docker run -d --name logstash-sync \ -v /opt/logstash-sync/config:/usr/share/logstash/config-pipeline \ -v /opt/logstash-sync/sql:/usr/share/logstash/sql \ -v /opt/logstash-sync/lib:/usr/share/logstash/lib \ -v /opt/logstash-sync/data:/usr/share/logstash/data \ docker.elastic.co/logstash/logstash:8.11.0 \ logstash -f /usr/share/logstash/config-pipeline/goods.conf注意data目录必须挂载否则容器重启后last_run_metadata_path丢失会触发全量同步。这个坑我踩过一次一张 300 万行的表重启后全量拉了一遍MySQL 直接告警。4.2 MySQL 侧的准备工作同步账号要单独建只给必要的读权限别用 rootCREATE USER sync_user% IDENTIFIED BY your_password; GRANT SELECT ON biz_db.goods TO sync_user%; FLUSH PRIVILEGES;表结构上update_time字段必须有索引否则每次增量查询都会全表扫描ALTER TABLE goods ADD INDEX idx_update_time (update_time);这个索引是性能的关键。没有它每次同步的WHERE update_time ?都会走全表扫描数据量一大 MySQL 就扛不住。我见过一张 500 万行的表没加这个索引同步任务一跑主库 CPU 直接飙到 90%。4.3 ES 索引模板的预定义在同步开始前最好先手动创建索引模板把字段类型定死。否则让 ES 自动推断第一次写入的文档决定了字段类型后面类型不一致就会报错。PUT _template/goods_template { index_patterns: [goods*], settings: { number_of_shards: 3, number_of_replicas: 1, refresh_interval: 5s }, mappings: { properties: { id: { type: long }, goodsName: { type: text, fields: { keyword: { type: keyword } } }, categoryId: { type: integer }, price: { type: scaled_float, scaling_factor: 100 }, stock: { type: integer }, status: { type: byte }, createTime: { type: date }, updateTime: { type: date } } } }refresh_interval设成 5s 而不是默认的 1s是因为同步场景不需要秒级可见降低刷新频率能显著提升写入吞吐。price用scaled_float加scaling_factor: 100既保证精度又节省空间比double更适合金额。4.4 启动与首次全量验证首次启动时last_run_metadata_path文件不存在Logstash 会把游标当成 1970 年触发全量同步。这时候要盯着日志看docker logs -f logstash-sync正常的话会看到类似这样的输出[INFO ][logstash.inputs.jdbc] (0.5s) SELECT ... FROM goods WHERE update_time 1970-01-01 08:00:00 [INFO ][logstash.outputs.elasticsearch] Successfully bulk indexed 2000 documents全量同步完成后去 ES 里 count 一下和 MySQL 的 count 对比curl -s http://es-node:9200/goods/_count | jq .count两边数量一致说明首次全量没问题。之后每次增量同步日志里会显示实际拉取的行数正常情况下应该是几条到几十条。5. 常见问题排查与避坑实录5.1 时间偏移 8 小时这是最高频的问题。现象是 ES 里的时间和 MySQL 里的差了 8 小时。原因通常是连接串没加serverTimezone或者 Filter 的date插件没指定timezone。两个地方都要检查缺一不可。我一般会在连接串里加serverTimezoneAsia/Shanghai同时在date插件里加timezone Asia/Shanghai双保险。5.2 增量同步漏数据前面讲过同一秒并发写入会漏。解决办法是游标回退。但还有一种漏数据的情况是update_time字段没更新。排查方法是拿一条漏掉的数据去 MySQL 里看它的update_time是不是比实际修改时间早。如果是说明业务代码的更新语句没带这个字段。5.3 ES 映射冲突报错报错信息类似mapper_parsing_exception: failed to parse field [price] of type [float]。原因是 ES 里字段类型和实际写入的数据类型不匹配。比如模板里price是float但某条数据的price是空字符串转换失败。解决办法是在 Filter 里做空值处理filter { mutate { gsub [price, ^$, 0] } }把空字符串替换成 0避免类型转换失败。5.4 同步任务卡死或越来越慢如果发现同步任务执行时间越来越长大概率是游标文件没更新每次都在重复拉取大量数据。检查last_run_metadata_path文件的内容看看游标值是不是在正常递增。另一个可能是 MySQL 的update_time索引失效了用EXPLAIN看一下查询计划。5.5 常见问题速查表问题现象可能原因排查方向解决办法时间差 8 小时时区未指定检查连接串和 date 插件两处都加 Asia/Shanghai数据漏同步并发写入或字段未更新对比 MySQL 和 ES 数据游标回退 检查 update_time映射冲突类型不一致看 ES 报错字段Filter 里做类型转换和空值处理任务越来越慢游标未更新或索引失效看游标文件和 EXPLAIN修复游标 加索引容器重启后全量data 目录未挂载检查挂载配置挂载持久化卷驱动连接失败驱动版本不对看启动日志换 mysql-connector-j-8.x5.6 几个我踩过的坑第一个坑是驱动版本。MySQL 8 默认的认证插件是caching_sha2_password老驱动不支持必须用 8.x 的驱动。这个坑卡了我半天日志里报的是认证失败但实际是驱动太老。第二个坑是clean_run参数。默认是false表示保留游标。如果设成true每次启动都会清空游标从头全量。这个参数千万别乱动我有次调试时改成true忘了改回来结果每次重启都全量同步。第三个坑是jdbc_page_size和jdbc_fetch_size的区别。前者是 JDBC 查询的分页大小后者是游标每次取的行数。两个都设太小会导致频繁的数据库往返设太大又占内存。我的经验是jdbc_fetch_size设 1000jdbc_page_size设 100000兼顾内存和效率。第四个坑是 ES 的refresh_interval。默认 1s同步场景下完全没必要改成 5s 或 30s 能明显提升写入性能。但改完之后查询会有延迟如果业务对实时性要求高就别改。6. 性能调优与扩展思路6.1 批量写入的参数调优写入性能的三个关键参数是flush_size、idle_flush_time和 ES 的refresh_interval。这三个参数是联动的。flush_size越大单次 bulk 请求越大吞吐越高但延迟也越高。idle_flush_time是兜底防止数据卡住。refresh_interval决定数据多久可见。我的调优顺序是先把refresh_interval调到 30s然后把flush_size从 500 逐步往上调每次翻倍观察 ES 的写入延迟和 Logstash 的 CPU。调到某个值后延迟开始明显上升就回退一档。一般 2000 到 5000 是比较舒服的区间。6.2 多表同步的 pipeline 拆分一个 Logstash 实例可以跑多个 pipeline每个 pipeline 同步一张表。这样好处是互不影响一张表出问题不会拖累其他表。但要注意资源隔离pipeline 之间会争抢 CPU 和内存。如果表很多建议拆到多个 Logstash 实例上。pipelines.yml里可以给每个 pipeline 单独配pipeline.workers和pipeline.batch.size- pipeline.id: goods_sync path.config: /opt/logstash-sync/config/goods.conf pipeline.workers: 2 pipeline.batch.size: 500pipeline.workers是工作线程数默认等于 CPU 核数。同步任务大多是 IO 密集型的线程数不用太多2 到 4 个就够。pipeline.batch.size是单次处理的批次大小和flush_size配合使用。6.3 后续可以扩展的方向如果数据量继续增长Logstash 的轮询模式会逐渐吃力。这时候可以考虑几个方向一是引入消息队列做缓冲Logstash 从队列消费而不是直接查 MySQL二是换成基于 Binlog 的 CDC 方案实时性更好三是把同步任务拆成多个实例按数据范围分片。但我要说的是别过早优化。我见过不少团队数据量才几十万行就上 Flink CDC 集群运维成本高得离谱。Logstash 这套方案在千万级数据量下跑得稳稳的真到了瓶颈再换也不迟。最后分享一个我一直在用的小技巧给同步任务加一个监控每次同步完成后把拉取的行数和耗时打到日志里再用日志平台做告警。如果某次同步行数突然暴涨或者耗时异常能第一时间发现。这个监控成本很低但能帮你提前发现很多问题比如某张表突然被批量更新或者游标文件损坏导致全量拉取。