企业官网建设流程全解析

1. 为什么数据要进Elasticsearch之前，先过一遍Logstash

这个实战系列写到第六篇，前面我们解决了ES集群部署、索引映射、中文分词这些事，但真正把项目跑起来才发现，最大的坑卡在数据入口：日志文件散落在一堆服务器上，MySQL里的业务数据想同步进来，各种格式千奇百怪。有人选择自己写Python脚本清洗，结果日志格式一改，脚本就得跟着改，越堆越难维护。

Logstash就是来解决这个问题的。它是一个开源的服务端数据处理管道，核心思想非常朴素：从不同数据源读取数据，在管道里做清洗、转换、补全，再写入目标存储，最常用的目标就是Elasticsearch。你千万不能把它当成一个简单的“搬运工”，它的价值不在于搬数据，而在于搬之前把数据收拾成ES能高效索引的样子。

到底什么场景适合用Logstash？我这边实际工作总结下来，三个条件至少占两个，就值得引入：数据量大、格式乱、来源多。比如同时接应用日志、数据库表、消息队列里的JSON，还要求统一清洗后写入ES。反过来，如果只是单机小项目，一天几百条数据，前端直接调ES写入接口就完事了，硬上一套Logstash反而增加维护成本。

适合谁看这篇？我默认你有ES基础，知道索引、文档、映射这些概念，但是被“数据怎么进去、进去之后怎么变成想要的格式”折磨过。《快速入门》的意思不是说只讲安装，而是把从零到能跑通一条数据管道的完整路径走一遍，不是让你看一堆概念，而是可以照着动手。

2. 装一个能跑的Logstash：版本、目录结构和第一个Hello World

2.1 版本对应关系是第一个坑

Logstash是用Java写的，对JDK版本很敏感，这一点怎么强调都不过分。ES和Logstash的版本最好保持大版本一致，7.x配JDK8，8.x配JDK17。这俩版本对应的JDK不能乱换，别在一台机器上为了省事直接复用别的Java应用的JDK，启动时经常会遇到类库版本冲突或者UnsupportedClassVersionError，排查半天最后发现是指定错了JAVA_HOME。

我这边实际习惯是：单独给Logstash解压一套JDK，在启动脚本里显式指定JAVA_HOME，不污染同一台机器上其他Java应用。tar包安装比rpm包灵活，实验环境尤其推荐，换版本直接解压换目录就行，不用跟系统服务纠缠。

wget https://artifacts.elastic.co/downloads/logstash/logstash-8.12.2-linux-x86_64.tar.gz tar -zxvf logstash-8.12.2-linux-x86_64.tar.gz cd logstash-8.12.2 bin/logstash --version

第一次跑推荐用最简单的方式验证安装成功：

bin/logstash -e 'input { stdin {} } output { stdout {} }'

启动后输入一行文本，会原样打印出来。这一步能过滤掉90%的JDK问题，如果这个都跑不通，先别急着配别的。

2.2 解压后先搞清楚这几个目录

Logstash解压后，你需要记住几个关键位置：bin目录是启动脚本所在，config目录里有logstash.yml和pipelines.yml，logstash.yml控制的是进程级别的参数，pipelines.yml则用于声明同时跑哪些管道。每个管道对应一个或多个配置文件，新手最容易懵的地方就在这儿：把管道配置直接塞进logstash.yml，或者反过来在pipelines.yml里写逻辑。

实际项目里我的布局是：

# config/pipelines.yml - pipeline.id: file_to_es path.config: "/opt/logstash/pipeline/file.conf" - pipeline.id: mysql_to_es path.config: "/opt/logstash/pipeline/mysql.conf"

每个conf文件独立，改一个重启一个，互不影响。实验阶段可以不加pipelines.yml，直接用 -f 指定单个配置文件：

bin/logstash -f /opt/logstash/pipeline/file.conf

注意，用 -f 的时候logstash.yml里关于管道的配置一般不生效，别两边写重复了。我见过有人在这两个配置里定义了两个不同的管道路径，启动后怎么都不跑自己想要的配置，其实就是配置优先级没搞清楚。

2.3 stdout输出是调试神器

初学阶段建议把output先写成stdout而不是ES，配合codec rubydebug，能看清经过filter处理后的文档长什么样：

output { stdout { codec => rubydebug } }

每个事件会以类似JSON的缩进格式打印，字段名、类型、新增字段一目了然。我习惯所有新管道先这么跑通，确认字段没问题后再把output换成Elasticsearch。有人说这样会暴露日志内容，没关系，本地调试的时候无所谓，生产环境别开就行。

3. 两种最常见的接入场景：读日志文件和同步MySQL业务表

3.1 file input读日志：不是tail那么简单

生产环境最常见的需求是把应用日志送进ES。用file input时，最核心的概念是sincedb。Logstash会记录每个文件已经读到的位置，存在一个独立文件里，下次启动时接着读，不会把整份日志再灌一遍。这个机制很省心，但有一个坑：开发环境反复测试时，想重新读日志，需要删除sincedb文件，不然看不到预期效果。

input { file { path => ["/data/logs/app/*.log"] start_position => "beginning" sincedb_path => "/tmp/app_logs_sincedb" codec => multiline { pattern => "^\\s" what => "previous" } } }

日志文件里最烦的是Java异常堆栈，一行看似独立的日志实际上是一个多行事件的整体。上面配置的意思是：如果某一行以空白字符开头，说明它是上一行的延续，把它拼回去。这个参数用语言讲挺简单，实际调试时还是要靠output打印观察，尤其是堆栈贴了很多层的时候。

start_position这个参数只对第一次读取生效，后续都是sincedb说了算。所以测试环境“重新读日志”的准确做法是：改一个不同的sincedb_path，或者删除旧文件再重启Logstash。我踩过最大的坑就是删了日志文件但忘删sincedb，结果Logstash以为文件没变过，无论如何不读新日志。

3.2 jdbc input同步MySQL：增量不是想起来就做的

除了日志，很多业务数据在MySQL里，想让ES能用上分词和聚合，就得定时同步。logstash-input-jdbc是经典插件，它之所以应用广泛，是因为它维护了“上次读到哪儿”这个状态，你不需要自己写定时任务和游标。

input { jdbc { jdbc_driver_library => "/opt/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/app_db?useSSL=false&serverTimezone=Asia/Shanghai" jdbc_user => "es_reader" jdbc_password => "xxxxxx" schedule => "*/5 * * * * *" statement => "SELECT id, title, content, update_time FROM article WHERE update_time > :sql_last_value ORDER BY update_time ASC" use_column_value => true tracking_column => "update_time" tracking_column_type => "timestamp" last_run_metadata_path => "/tmp/.mysql_last_run" } }

这里最关键的就是:sql_last_value。Logstash每次同步完会把tracking_column的最后值存到last_run_metadata_path指定的文件里，下次执行SQL时替换进去。实现增量同步的一个可靠套路：SQL里永远用大于号，排序，并确保tracking_column有索引。不要为了图省事用“等于”或“大于等于”，边界条件一多就乱套。

实际踩坑：serverTimezone不写，日期时间字段经常差8小时；tracking_column的字段类型是date的话，SQL参数绑定的时间格式可能和MySQL表里不同，容易出现“这次同步了上次已经同步过的数据”或“永远漏掉更新数据”。一般我会在MySQL侧把update_time处理成TIMESTAMP并加索引，减少边界问题。

3.3 output到ES：document_id决定幂等性

写完input和filter，数据要落到ES。最容易被忽略的选项是document_id。如果不设置，每条日志被当成新文档写入，重复运行或者数据源回放时会产生重复数据；设置了document_id，主键相同就覆盖，整体操作具备幂等性。

output { elasticsearch { hosts => ["http://es-node01:9200", "http://es-node02:9200"] index => "app-log-%{+YYYY.MM.dd}" user => "${ES_USER}" password => "${ES_PASSWORD}" document_id => "%{id}" manage_template => false template_name => "app_log_template" } }

%{+YYYY.MM.dd}是日期占位符，按天分索引时很常用；${ES_USER}是读环境变量，不要在配置文件里硬编码密码，不然代码仓库一传就泄露了。manage_template设为false，意味着你需要手动在ES里建好索引模板，字段类型由你自己控制；如果为true，Logstash默认模板可能把数字类型猜成long或text，后面查询时就会因为类型不对而头疼。

4. 清洗的舞台：filter段里最常用的三兄弟

4.1 mutate：类型转换和字段整理

ES动态映射有个特点：它会把看起来像数字的字段映射为long，把看起来像日期的字段映射为date，但实际情况全然不可控。比如订单号以0开头，传到ES被当成long，前面的0就丢了，只能靠字符串字段保存。用mutate可以显式处理这些类型问题。

filter { mutate { convert => { "status" => "integer" } rename => { "ts" => "@timestamp" } copy => { "hostname" => "host" } remove_field => ["sensitive_data", "raw_message"] gsub => ["message", "\\s+", "_"] } }

convert的可用目标类型包括string、integer、float、boolean等。注意它在字段值为空或格式不对时不会报错，而是保留原值，所以不要写完convert就认为万事大吉，需要抽几行输出确认。gsub是正则替换，实际中常用于把日志中敏感信息替换成星号，或者把特殊字符统一化。

4.2 grok：把非结构化日志变成结构化字段

应用日志经常是这种格式：

2024-06-17T10:23:45+08:00 INFO 192.168.1.10 "GET /api/v1/order/123" 200 45ms

想把它拆成level、clientip、url、response_code、duration_ms字段，靠的就是grok。grok本质上是一堆命名正则表达式，内置了IP、NUMBER、HTTPDATE等常见模式。写法是%{SYNTAX:semantic}，冒号前面是模式名，后面是字段名。

filter { grok { match => { "message" => "%{TIMESTAMP_ISO8601:ts} %{LOGLEVEL:level} %{IP:clientip} \"%{WORD:method} %{URIPATHPARAM:url}\" %{INT:status} %{INT:duration}ms" } } }

grok写完后，第一眼看事件里有没有_grokparsefailure这个标签。只要出现这个字段，说明没匹配上。最常见的错误是表达式和实际日志格式稍有出入，比如多了中括号、引号、空格。建议用在线调试工具或直接在grok里加add_tag调试，不要靠猜。

用正则时，“贪婪匹配”害死人。比如想匹配日志里某个JSON字段，直接用.*可能吃掉后面所有内容。所以能用内置模式就尽量用，不能用时限定字符范围而不是任意字符。实际项目里，我见过有人一个grok表达式写了200多字符，跑起来性能很差，后来拆成多个小表达式，吞吐翻了一倍。

4.3 date：让时间字段成为真正的date类型

日志里的时间是字符串，但ES要排序、要按时间聚合，必须是date类型。date过滤器就是干这个的。

filter { date { match => ["ts", "ISO8601"] target => "@timestamp" timezone => "Asia/Shanghai" } }

match里的格式支持Java时间格式字符串和几个内置命名，如ISO8601、UNIX、UNIX_MS。如果日志时间是标准的yyyy-MM-dd HH:mm:ss，要写"yyyy-MM-dd HH:mm:ss"。这个步骤的实质是把字符串转成内部的UTC时间存储，同时映射到ES的date字段。

重点提醒：ELK生态里@timestamp是默认时间字段，ES的索引模板也会默认按这个字段建date映射。如果你同时从日志里解析出另一个ts字段，记得在filter最后把原始字符串ts去掉，不然会出现两套时间字段，显示很乱。过滤器的顺序也至关重要，date要放在grok之后，因为它要解析的是grok抽取出来的ts字段。

4.4 把想要的都抽出来，再决定丢哪些

刚开始用Logstash的人容易陷入两个极端：要么只把message整段丢给ES，啥都不解析，查询时只能全文搜；要么配置里写了一大堆字段，很多根本不用于查询，白白增加索引体积。我建议的套路是：filter里先把关键字段全部解析出来，然后通过mutate把不需要的原始字段删掉，只保留必要的。这样日志保留期长，磁盘也吃得住。解析字段这件事，宁可前期多花半小时调试，也别上线之后天天做关联查询。

5. 跑得慢、丢数据、时间不对：排错思路和性能调优

5.1 先学会看Logstash自己的日志

Logstash管道配置出错时，启动时会直接失败，但运行中的报错都写在logstash-plain.log里。很多人接入后发现ES里没数据，第一反应是盯ES，其实应该先看Logstash日志。每次启动时注意看“Pipelines running”和下面的pipeline id，看到这些字样才说明管道加载成功。

调试期最有效的输出手段就是stdout rubydebug。不过生产环境千万别一直开着，否则日志量直接爆炸。想在不干扰业务数据的情况下调试，可以另外起一个临时pipeline，只用stdin模拟数据，或者用只读模式跑一个小样本文件。

5.2 重点调优参数

Logstash性能瓶颈多不在CPU，而在内存和下游ES写入速度。相关参数集中在logstash.yml里：

batch.size调大不一定更快，因为内部队列要攒事件，内存占用跟着涨。我实践下来，把batch.size从125调到500，大部分场景吞吐能翻倍，但堆内存也得从默认1G调到4G。具体调到多少，没有标准答案，要看事件大小和ES端bulk的性能。

JVM堆配置改的是jvm.options文件里的-Xms和-Xmx。注意别超过物理内存的一半，你还有filebeat、ES和其他Java程序在跑，得给系统留点余量。改完堆配置要重启Logstash才会生效，别只改不重启然后说没变化。

5.3 常见故障现象和根因速查

我把这几年遇到的案列整理成表，方便大家直接对号入座：

5.4 压测数据管道最简单的办法

不用专门造压力工具，生产服务器上挑一份真实的存量日志文件，配置file input从头读取，输出到测试ES索引，记录完成时间。用两份不同规模的日志跑两遍，能大致算出吞吐。再逐步调batch.size、workers，观察ES的bulk队列和CPU。这套方法比理论计算靠谱，因为我们最关心的就是真实数据格式下的表现。记住压测时用单独索引，别把测试数据混进生产索引。

6. 别把所有事情都压给Logstash：和Filebeat的分工

6.1 两者的边界在哪里

现在ELK生态里，Beats和Logstash经常被一起讨论，也容易被弄混。Filebeat是轻量级采集器，部署在每个应用服务器上，负责读日志、做简单过滤、发给Logstash或直接发给ES；Logstash是集中式处理器，负责多来源汇聚、复杂解析、富化、输出到多个目标。

如果服务器数量多到几十台，每台都装Logstash显然不现实，内存就是个大坑，用Filebeat更合理。但如果日志格式复杂，需要处理多行、做grok、用业务系统数据补全字段，这些最好集中交给Logstash，而不是在每台服务器上复制一堆复杂的filter配置。

一个实践上比较舒服的架构是：Filebeat安装到各个业务机，只做读文件和简单压缩；中心化部署一两个Logstash节点，专门做解析和清洗；Elasticsearch只承接处理后的结构化数据。如果日志量不大、服务器就三五台，也可以Filebeat直连ES，减少链路。

6.2 pipeline按业务拆分，别堆在一个大conf里

随着业务增加，把多个数据源的配置写在一个conf里会很难维护，改一处可能牵连整个管道。我的建议是pipelines.yml里按业务拆出多个pipeline，每个pipeline有自己的workers和内存配额。出问题时，单独重启某条pipeline，不会影响其他数据的采集。

每个pipeline的持久化队列也可以打开（queue.type为persisted），这样Logstash重启期间数据不会丢太多。代价是需要额外磁盘，但相比丢日志，这个代价通常值得。生产环境我一般给Logstash单独挂一块数据盘，专门放持久化队列和sincedb文件。

6.3 扩展思路：输出到多个目标

Logstash的output段支持同时配置多个目标，比如一份数据既写入ES，又写入Kafka，或者同时写ES和发送到监控系统。实际项目里我会把原始日志归档到另一个索引，处理后字段写入业务索引，两个索引的生命周期策略不同。这个玩法很适合做数据审计。

现在很多团队慢慢走向轻量采集、集中处理的架构，Logstash在链路的定位越来越清晰。它不会是配置最复杂的组件，但一定是让数据好用的关键一环。每次看到有人问“为什么我的ES里字段类型不对”，十有八九是他的数据管道少做了清洗这一步。

在维护这套数据接入链条的两三年里，我最大的感触是：Logstash的难点不在怎么装、怎么配，而在怎么让数据格式稳定下来。日志格式在业务代码里一变，ES里字段就得跟着调；调字段最有效的方法，就是一遍遍在stdout里看实际输出，直到字段干净为止。这个“脏活”绕不开，但入门的路径其实很短，只要先跑通一个文件管道，再摸透grok和date，大部分场景的套路就通了。希望这次快速入门能帮你把数据管道这扇门推开，后面真正做索引模板和生命周期管理的时候，就会顺手很多。