1. 先弄清楚Logstash到底解决什么问题

1.1 手打bulk API的无奈

做ElasticSearch实战系列的读者，多半已经遇到过同一个问题：ES本身是个存储和检索的引擎，但数据怎么进去，始终是个绕不开的坎。网上有很多文章教你用curl直接调_bulk接口，一次塞几条JSON进去做Demo。可一旦你面对的是真实环境——服务器上每天滚动增长的Nginx访问日志、某个业务库里的订单表、一堆格式各异的文本文件，靠手工拼JSON批量写入根本不现实。你得自己写脚本做解析、字段映射、类型转换、日期格式化，还得处理增量、去重、异常重试。

所以Logstash在这个体系里的定位非常明确：它是ES生态里的数据接入管道，负责把来源五花八门的数据，经过统一的采集、解析、过滤、转换之后，稳定地写入Elasticsearch或其他存储端。我之前在某公司做日志平台改造，几十台应用服务器的日志全走Filebeat采集，统一汇到Logstash做字段切分和清洗，再送到ES集群。这套链路扛住了日均上亿条日志的写入，可以负责任地说，Logstash是这套架构里最不起眼、但绝对关键的一环。

1.2 三段式管道，本质是一条数据加工流水线

Logstash的核心模型一句话就能讲完：一条管道由input、filter、output三段组成，数据从输入插件进入，经过多个过滤插件依次处理，最后从输出插件离开。

这个设计特别像工厂里的流水线。原料从入口进来，流水线上有不同的工位，有的工位负责拆解零件，有的负责清洗，有的负责打标签，最后包装出货。Logstash里的grok干的就是拆解这活儿，mutate负责改格式，date专门修正时间字段，geoip给数据加地理位置信息，output则把成品发货到Elasticsearch、文件、消息队列等目的地。

一个配置示例就能看出这三段的写法：

input { file { path => ["/var/log/nginx/access.log"] start_position => "end" } } filter { grok { match => { "message" => "%{COMBINEDAPACHELOG}" } } } output { elasticsearch { hosts => ["http://localhost:9200"] index => "nginx-access-%{+YYYY.MM.dd}" } }

这个文件放在config/目录下，启动时指定-f参数加载即可。注意这里的语法是Ruby风格，字符串用双引号，注释用#，插值用%{}，别用JSON的思维写配置，否则会踩很多低级错误。

1.3 什么场景适合Logstash，什么场景别用

很多人一上来就陷入工具选择的误区。这里把我的实际经验说透。

Logstash适合的场景有三类：第一类是日志文件采集，配合Filebeat做轻量收集端，Logstash做集中处理端，这是最常见也最稳的组合；第二类是数据库同步，通过JDBC插件定期拉取业务表数据进ES，适合做搜索或报表的辅助数据源；第三类是需要复杂数据清洗的场合，比如多条数据关联、字段拆分、自定义正则解析、IP地理解析等。

反过来说，有些场景完全没必要上Logstash。如果数据本身就是规整的JSON结构，量也不大，直接用Filebeat或应用程序里的ES客户端写入反而更轻；如果数据量大到每秒几十万条，Logstash会成为瓶颈，这时候通常会在前面加一层消息队列，或者直接用Spark/Flink这类流处理框架。Logstash不是万能的，我见过有人拿它做超大流量的实时数仓底座，结果性能调到头也顶不住，最后换了方案。工具选型要在动手前想清楚。

2. 安装与初体验：先让管道转起来

2.1 版本与JDK：先对齐全家桶版本

安装Logstash之前，我强烈建议先确认一件事：Logstash主版本要和Elasticsearch主版本保持一致。7.x的Logstash就配7.x的ES，8.x配8.x，别混搭。虽然跨版本有时也能跑，但Plugin之间的兼容性、索引模板格式、许可证逻辑都可能出问题。我用7.x系列做过生产实践，稳定成熟，教程也多，适合新手入门。

另一个关键点是JDK。Logstash本身是Java写的，启动前必须装JDK。7.13之前官方默认支持JDK 8，7.13之后开始支持JDK 11，8.x则要求JDK 17。我第一次装的时候JDK版本没对上，启动直接报Unsupported class file major version，后来统一用JDK 11才消停。如果机器上有多套JDK，记得在启动脚本或环境变量里明确指定JAVA_HOME，不然容易撞车。

2.2 目录结构与基础配置

下载好对应版本的压缩包后解压，目录不算复杂，但有几个关键目录得心里有数：

• bin/：启动脚本，logstash和logstash.bat都在这
• config/：全局配置文件的默认位置，启动时如果不指定-f，会默认加载logstash.yml
• pipeline/：官方推荐的管道配置目录，可以在logstash.yml里配置
• data/：程序运行时的数据目录，比如文件读取的sincedb就存在这里
• logs/：Logstash自己的日志，排错时第一个看这里

配置分两层：一层是logstash.yml，管的是进程级参数，比如堆内存、管道并发数、队列类型；另一层是pipeline.conf文件，管的是数据流逻辑。新手经常把两者混为一谈，在pipeline里写pipeline.workers之类的全局参数，结果压根不生效。其实这类参数要放在logstash.yml里。

2.3 第一次启动：用stdin和stdout验证管道

别一上来就跑复杂配置，先验证环境是否通。打开终端，进到解压目录下执行：

bin/logstash -e 'input { stdin { } } output { stdout { codec => rubydebug } }'

-e参数的意思是直接在命令行指定配置，不带也行，适合做冒烟测试。启动后命令会卡住，光标停在空白处，这时输入一行任意文字，比如hello logstash，回车，你就能看到stdout插件把这条数据处理后打印出来，包含@timestamp、@version、host等自动补充的字段。这说明JVM没毛病、插件加载正常、管道链路是通的。

注意：启动时JVM初始化需要一点时间，别刚敲完命令发现没反应就以为卡死。看日志里出现Pipeline started就说明管道已就绪。

3. 核心插件逐个讲：input / filter / output

3.1 input：数据从哪来

input插件决定数据入口，常用的有这么几种。

stdin：标准输入，适合测试和写Demo，上面已经演示过了。

file：读取文件，最常见的用法之一。核心参数有path（文件路径，支持通配符）、start_position（第一次启动时从beginning还是end开始读）、stat_interval（检查文件更新的频率）。这里有个最容易误解的点：start_position只在首次读取、且没有sincedb记录时才生效。服务器上日志每天滚动，Logstash靠sincedb文件记录每个文件的读取偏移量，不会重复读，也不会漏读。如果你想强制重新读一遍日志，需要删除data/plugins/inputs/file/下的sincedb_*文件再重启。

beats：接收Filebeat发来的数据，走的是elasticsearch官方协议，端口默认5044。生产环境里Filebeat负责在应用服务器上采集文本、监听文件变化、压缩传输，Logstash负责统一接收和处理，这样能把采集端压到最低，也不会每台机器都塞一个Java进程。这个组合我强烈推荐给有多台服务器日志采集需求的场景。

jdbc：定时从数据库拉数据，常用于同步业务表。通过jdbc_connection_string、jdbc_user、jdbc_password配置连接信息，配合jdbc_driver_library和jdbc_driver_class加载驱动，最关键的是schedule参数控制定时规则，比如"*/1 * * * *"表示每分钟跑一次。增量同步的细节我在后面专门讲。

3.2 filter：解析、清洗、丰富

filter是整个Logstash的灵魂，也是新手和老手拉开差距的地方。核心插件有这么几个。

grok：日志解析之王。它的核心工作是拿正则表达式和预定义模式去切分原始字符串，把message里的一整条文本拆成一系列命名捕获字段。比如Nginx日志里的一行：

192.168.1.7 - - [18/Sep/2024:14:32:11 +0800] "GET /api/user/list?page=1 HTTP/1.1" 200 1024 "-" "Mozilla/5.0"

用%{COMBINEDAPACHELOG}可以一次匹配出clientip、timestamp、request、status、bytes、user_agent等字段。grok内置了一百多种常用模式，但真实日志往往千奇百怪，需要自定义模式。自定义时写(?<field_name>正则表达式)就行，也可以把自定义模式放到patterns_dir指定的模式文件里复用。我运维过的一个内部系统日志，时间字段带毫秒和时区，标准模式匹配不上，最后自己写了三行正则才算完事。建议手边常备一个grok调试工具，边调边看匹配结果，效率高很多。

mutate：数据转换的瑞士军刀。常见的操作有rename重命名字段、copy复制字段、update更新已有字段、replace替换字段内容、remove_field删字段，以及convert做类型转换。日志解析出来的字段默认都是字符串，但ES里很多字段需要数值类型，比如status想做范围查询、response_time想取平均值，都得先转成int或float。我在实战里最常用的就是convert，正则切分后把明显是数字的字段全部转一遍。

date：专门处理时间。日志里通常自带时间戳，但Logstash生成的@timestamp是数据处理时刻，二者往往不一致，比如日志是昨天凌晨生成的，到了白天才采集进来。用date插件解析日志自带的时间字段，然后覆盖掉@timestamp，之后按时间归档索引、按时间范围查询才准确。时区问题也在这里处理，timezone => "Asia/Shanghai"是必配项，否则默认UTC会导致时间差8小时。

geoip：根据IP地址附加地理位置信息，包括经纬度、国家、城市等。配上之后，Kibana里做地图分布展示就很方便。这个插件要联网下载GeoIP数据库，网络受限的环境里需要提前准备好数据库文件，否则启动时插件会报错。

useragent：解析浏览器UA字符串，把操作系统、浏览器类型、设备型号拆成结构化字段。做访问日志分析时，这个插件配合geoip，一套下来就能得到比较完整的流量画像。

3.3 output：数据到哪去，以及多路输出

output决定数据流出去的方向。

stdout：打印到控制台。配codec => rubydebug时输出结构化的Ruby对象样式，字段层次一目了然，调试阶段基本离不开它。另外注意，stdout在sink模式并不会有太多额外开销，直接和elasticsearch输出并存即可。

elasticsearch：写入ES。关键参数是hosts、index和user/password。index支持按时间动态拼索引名，比如nginx-access-%{+YYYY.MM.dd}，这样每天自动生成一个索引，方便后面的索引生命周期管理。还可以通过template参数指定自定义索引模板，把number_of_shards、number_of_replicas、mapping等预置好。写入时如果某条数据格式对不上，默认会怎么办？这个很多人没注意，retry_on_conflict和action参数决定了冲突重试策略。生产环境我最常配的是action => "index"，让新文档直接写入，版本冲突交给ES自己处理。

kafka、redis、file：输出到消息队列或文件，适合做数据分发。比如日志平台既要把数据写入ES供检索，又要转一份到Kafka供实时计算，这时候可以在一个pipeline里写多个output，数据会同时发往所有输出端。注意输出插件之间是并行的，哪个输出慢就拖累整体管道，所以生产上多路输出时，要把慢端前置解耦，比如先写Kafka再消费进ES。

4. 实战：把Nginx日志送进Elasticsearch

理论知识再多，不如跑通一个完整例子。这一节我用一个常见的Nginx访问日志采集场景，从零写一份可用的pipeline配置，并把每一步设计思路讲清楚。

4.1 第一步，先让file插件稳定读文件

假设Nginx日志路径是/var/log/nginx/access.log，格式是默认的combined格式。先只写input和output，不接filter，把原始数据先读进ES里看看效果：

input { file { path => ["/var/log/nginx/access.log"] start_position => "end" stat_interval => 3 codec => "plain" } } output { elasticsearch { hosts => ["http://127.0.0.1:9200"] index => "nginx-original-%{+YYYY.MM.dd}" } stdout { codec => rubydebug } }

这里几个参数值得留意。start_position => "end"表示只读新增内容，不会把历史日志全部导入，避免首次启动就把磁盘和网络打满；stat_interval => 3表示每3秒检测一次文件变化，兼顾实时性和IO开销；codec => "plain"表示按纯文本处理，不拆包、不解析，先把原始行保留下来。

启动后观察stdout输出，能看到自动生成的message字段就是原始日志行，另外还有path、host、@timestamp等字段。这一步的目标是把链路打通，确认文件读取没有问题。

4.2 第二步，用grok把日志切成一堆字段

数据能进去之后，接着做解析。Nginx的combined格式是现成的模式，直接匹配：

filter { grok { match => { "message" => "%{COMBINEDAPACHELOG}" } } }

匹配成功后，你会在stdout里看到clientip、ident、auth、timestamp、verb、request、httpversion、rawbytes、response、bytes、referrer、agent这些字段。举个例子，访问IP会落到clientip里，请求方法落到verb里，请求路径和参数落到request里，返回状态码落到response里。

不过COMBINEDAPACHELOG是通用模式，遇到实际格式有出入很正常。我碰到过某个内部服务在日志里额外加了$request_time字段，标准模式肯定匹配不出，这时就要自定义模式：

filter { grok { match => { "message" => "%{IP:clientip} \[%{HTTPDATE:timestamp}\] \"%{WORD:verb} %{NOTSPACE:request} HTTP/%{NUMBER:httpversion}\" %{INT:status} %{INT:bytes} %{QS:referrer} %{QS:agent} %{NUMBER:request_time}" } } }

注意\s+不要漏，\[和\]都要转义。每一段写完最好单独拿日志行去测试，全部匹配成功后才会得到干净的字段集。grok的匹配是贪婪的，多个模式都匹配时默认只取第一个成功的结果，这点可以在break_on_match参数里调整。

4.3 第三步，用mutate和date收尾

解析出字段之后，先做类型清洗，再做时间修正。Nginx的status、bytes、request_time解析出来都是字符串，得转成数值：

filter { mutate { convert => { "status" => "integer" "bytes" => "integer" "request_time" => "float" } remove_field => ["message", "path", "ident", "auth", "httpversion"] } }

bytes有可能是-，这时候convert成integer会失败并抛异常。稳妥的做法是先加一个条件判断：如果bytes == "-"，就给它赋值为0再转。这个坑我踩过好多次，日志里一个缺省值就能让整条管道报错。

接着处理时间。Nginx日志里的时间字段长这样：18/Sep/2024:14:32:11 +0800。用date解析并覆盖@timestamp：

filter { date { match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ] target => "@timestamp" timezone => "Asia/Shanghai" } }

关键点有两个。第一，match的格式要和日志里的时间格式完全对应，dd/MMM/yyyy对应18/Sep/2024，Z对应+0800时区偏移；第二，timezone一定要设置成业务所在时区。如果你的日志时间就是Asia/Shanghai，但ES里查出来总是8小时前的数据，十有八九就是这里漏配了时区。

4.4 完整配置与启动验证

把三段拼起来，一份可用的pipeline配置就是这样的：

input { file { path => ["/var/log/nginx/access.log"] start_position => "end" stat_interval => 3 } } filter { grok { match => { "message" => "%{COMBINEDAPACHELOG}" } } if [bytes] == "-" { mutate { replace => { "bytes" => "0" } } } mutate { convert => { "status" => "integer" "bytes" => "integer" "request_time" => "float" } remove_field => ["message", "path", "ident", "auth", "httpversion"] } date { match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ] target => "@timestamp" timezone => "Asia/Shanghai" } } output { elasticsearch { hosts => ["http://127.0.0.1:9200"] index => "nginx-access-%{+YYYY.MM.dd}" } stdout { codec => rubydebug } }

启动前先做配置校验：

bin/logstash -t -f /path/to/nginx-pipeline.conf

看到Configuration OK再启动。运行后打开Kibana的Index Patterns，创建nginx-access-*索引模式，就能按天看到完整的访问日志数据了。

提醒一下：Logstash配置文件改动了，如果只是局部调整，可以用-r参数开启自动重载，配置变更后不用重启进程。但某些插件参数变化不会触发重载，需要手动重启才能生效，生产环境改配置前建议先明确哪个生效机制。

5. 进阶玩法：JDBC同步数据库与多管道配置

5.1 用JDBC把业务库增量同步进ES

日志是一类数据，业务库里的表是另一类常见数据源。比如商品表、订单表，需要把字段同步到ES里做搜索和聚合。Logstash的JDBC插件让这个活儿变得很简单。

装好logstash-input-jdbc插件之后，写一份这样的配置：

input { jdbc { jdbc_driver_library => "/opt/logstash/lib/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/business_db?useSSL=false&characterEncoding=utf8" jdbc_user => "logstash" jdbc_password => "你的密码" statement => "SELECT * FROM products WHERE update_time > :sql_last_value" schedule => "*/5 * * * * *" use_column_value => true tracking_column => "update_time" tracking_column_type => "timestamp" last_run_start_time => true } } output { elasticsearch { hosts => ["http://127.0.0.1:9200"] index => "product-search" document_id => "%{id}" } }

这里有几个机制要理解透。schedule用cron表达式控制轮询频率，*/5 * * * * *表示每5秒跑一次。sql_last_value是一个内置变量，会自动记录上次同步的最后一条记录的tracking_column值（这里是update_time），下次执行时替换到SQL里，实现增量拉取。第一次运行时，sql_last_value默认是1970-01-01，会把全表数据捞一遍。

document_id参数非常关键，它让ES里的文档ID和业务主键一一对应，重复同步时不会产生重复文档，而是覆盖更新。否则每次同步都会新增一批文档，查出来全是重复数据。

5.2 用pipeline.yml管理多个管道

如果一台Logstash同时要处理Nginx日志、业务数据库同步、安全设备日志，不可能都塞到一个pipeline里，因为不同数据源的处理逻辑差别太大。官方推荐的方式是在config/pipeline.yml里定义多个管道：

- pipeline.id: nginx-access path.config: "/opt/logstash/pipeline/nginx.conf" pipeline.workers: 2 - pipeline.id: product-sync path.config: "/opt/logstash/pipeline/product.conf" pipeline.workers: 1 - pipeline.id: syslog-input path.config: "/opt/logstash/pipeline/syslog.conf" pipeline.workers: 1

这样每个pipeline有独立的配置、独立的工作线程、独立的队列，互不干扰。某个pipeline的filter写法再乱、输出端再慢，也只是拖累自己所在的管道。管理多个管道时，注意pipeline.workers的分配，JDBC同步这种低频任务给1个worker就够，高并发的日志管道可以多给几个。

5.3 性能调优：workers、batch、队列

Logstash性能调优，很多人第一反应是调JVM堆内存，但堆内存只是一部分。核心参数在logstash.yml里：

• pipeline.workers：相当于管道的并行处理线程数，默认等于CPU核数。不是越大越好，I/O密集和CPU密集场景的最佳值不同，我一般在CPU核数的基础上上下试探。
• pipeline.batch.size：单个worker每次批量处理的event数，默认125。调大能提升吞吐，但也会增加内存消耗和单批次失败的影响范围。
• pipeline.batch.delay：批次组装的最长等待时间，默认50ms。如果单条事件处理很快，适当提高batch.size比死磕delay管用。
• queue.type：默认是memory，事件堆积时内存会涨。如果追求可靠性，改成persisted，把事件队列持久化到磁盘，Logstash重启后队列不丢。

我遇到过的一个真实调优案例：一台8核16G的Logstash，处理的是多路日志聚合，一开始默认参数，吞吐只有每秒不到3000条。后来把pipeline.workers调到6，batch.size调到500，batch.delay调到100，吞吐直接翻倍到了7000多条。不过也提醒一句，往上调参数不是无脑的，观察堆内存和GC日志，老是可以的，堆外内存爆了会频繁Full GC，反而拖垮整体。

6. 踩坑实录：我遇到过的常见问题与排查思路

6.1 时间整整少了8小时

这是Logstash新手遇到最多的坑，我当年也中招过。现象是数据写进ES之后，Kibana里看到的时间比实际日志时间早8个小时。原因无非两处：

第一，date插件解析出时间后，@timestamp存的是UTC时间，ES的Kibana默认按浏览器时区展示，看起来就少了8小时。解决方式是给@timestamp设置正确时区，或者在Kibana里手动改时间字段的时区。但更推荐在Logstash里统一处理，让数据的时区明确一致。

第二，源日志时间本身是Asia/Shanghai，但Logstash进程所在服务器的系统时区是UTC。Logstash运行时会读取系统时区做一些隐式处理，所以如果服务端时区错乱，各种诡异时间问题都会冒出来。最省事的做法是把Logstash所在服务器系统时区也设为Asia/Shanghai。

6.2 同一个字段，索引映射说变就变

某个字段第一次写入ES时是字符串，ES自动建的mapping把字段映射成了text，后面某天数据里这个字段变成了数值，写入直接报mapper_parsing_exception。这个问题在日志场景特别常见，因为日志格式是松散可变的。

常规解法是在Logstash侧把字段类型统一：能转数值的字段全部用mutate的convert转成明确类型。如果字段值本身不稳定，还可以在ES侧预先定义索引模板，把容易变动的字段的mapping固定死。最彻底的方案是设计好索引生命周期，定期滚动索引，避免老索引的mapping约束新数据。

6.3 中文日志乱码与配置文件编码

如果采集的日志文件是GBK编码，而Logstash默认按UTF-8读，会出现一堆乱码。处理方式是在对应的input或codec上指定编码：

input { file { path => ["/var/log/legacy/app.log"] codec => plain { charset => "GBK" } } }

同理，filter里的中文匹配也有编码问题。我踩过的坑是配置文件里写中文正则或中文替换内容，但文件保存成了GBK，Logstash加载配置时解析失败。配置文件一律用UTF-8无BOM保存，这是最基本的习惯。

6.4 数据重复、管道挂了怎么办

数据重复的情况要分两类。文件采集场景里，sincedb记录没写好或文件被滚动后重新创建了，Logstash可能重复读取同一批日志。排查时先看data/plugins/inputs/file/下的sincedb文件，确认记录的inode和offset是否正常。真遇到重复数据，干脆停掉Logstash，删掉sincedb，调整start_position重新跑一次。

管道挂掉最常见的表现是queue is full或者worker busy。先检查Logstash自己的日志logs/logstash-plain.log，再看输出端的压力。ES写入慢了，管道自然积压。稳妥的做法是前面加一层Kafka削峰，Logstash从Kafka消费，无论上游流量怎么突刺，管道都稳如老狗。

6.5 排查工具：-t校验、-r重载、监控API

最后分享几个我日常排查的固定动作。

启动前用bin/logstash -t -f 配置文件做语法检查和配置校验，这个动作能挡掉90%的起步问题。改完配置懒得重启，用-r开启自动重载，当然发生产前还是要确认某些参数是否支持热加载。

运行中的状态查看，用HTTP监控API。在logstash.yml里开启：

http.host: "0.0.0.0" http.port: 9600-9700

然后访问http://localhost:9600，能看到节点信息；/_node/stats/pipelines里能看到每个管道的events.in、events.out、filtered、queue等实时指标。管道吞吐突然下降、积压加剧，这些数字会第一时间告诉你问题出在输入端、过滤端还是输出端。

从我个人的经验看，Logstash真正难的不是写配置，而是出了问题不知道怎么定位。把监控指标看明白、把常见坑提前避开，这个工具用起来会顺手很多。后面我在实战系列里还会继续深挖Kafka到Logstash再到ES的经典链路，到时候再跟大家聊削峰填谷的细节。