# 08_Elastic Stack 从入门到实践（八）—1

08_Elastic Stack 从入门到实践（八）—1

一、Logstash入门之简介以及部署安装

1、Elastic Stack 技术栈示意图

2、Logstash 简介

Logstash 是开源的服务器端数据处理管道，能够同时从多个来源采集数据,转换数据，然后将数据发送到您最喜欢的“存储库”中。(存储库当然是Elasticsearch.)

3、Logstash 用途

4、Logstash 部署安装示意图

5、Logstash 部署安装

1）下载Logstash （如：logstash-6.5.4.tar.gz），并上传到虚拟机服务器。

https://www.elastic.co/downloads/past-releases#logstash

https://artifacts.elastic.co/downloads/logstash/logstash-6.5.4.tar.gz

2）部署安装（解压即安装）

# 检查jdk环境，要求jdk1.8+
java -version

# 切换目录
$ cd /dzs168/

# 解压安装包
tar -xvf logstash-6.5.4.tar.gz

# 切换目录
$ cd /dzs168/logstash-6.5.4/

# 第一个1ogstash示例
bin/logstash -e 'input { stdin {} } output { stdout {} }'

3）执行效果如下：可以看出logstash 启动很慢，现在已经被 beat 代替。

二、Logstash入门之配置详解

1、Logstash 的配置有三部分，如下

input{
            #输入
stdin {
             ...}#标准输入
}
filter {
             #过滤，对数据进行分割、截取等处理
  ...
}
output{
             #输出
stdout {
             ...}#标准输出
}

2、Logstash输入：

1）采集各种样式、大小和来源的数据，数据往往以各种各样的形式，或分散或集中地存在于很多系统中。

2）Logstash支持各种输入选择，可以在同一时间从众多常用来源捕捉事件。能够以连续的流式传输方式，轻松地从您的日志、指标、Web应用、数据存储以及各种AWS服务采集数据。

3、Logstash过滤：

1）实时解析和转换数据

2）数据从源传输到存储库的过程中，Logstash过滤器能够解析各个事件，识别已命名的字段以构建结构，并将它们转换成通用格式，以便更轻松、更快速地分析和实现商业价值。

4、Logstash输出：

Logstash 提供众多输出选择，您可以将数据发送到您要指定的地方，并且能够灵活地解锁众多下游用例。

三、Logstash入门之读取自定义日志以及解析处理

1、Logstash读取自定义日志

前面我们通过Filebeat读取了nginx的日志，如果是自定义结构的日志，就需要读取处理后才能使用，所以，这个时候就需要使用Logstash了，因为Logstash有着强大的处理能力，可以应对各种各样的场景。

2、Logstash日志结构，如下

2025-03-15 21:21:21|ERROR|读取数据出错|参数:id=1002

可以看到，日志中的内容是使用“|”进行分割的，使用，我们在处理的时候，也需要对数据做分割处理。

3、Logstash 编写配置文件

# 切换到 logstash 安装目录
$ cd /dzs168/logstash-6.5.4/

$  vim dzs168-pipeline.conf

input {
            
  file {
            
    path => "/dzs168/logstash/1ogs/app.log"
    start_position => "beginning"
  }
}
filter {
            
  mutate {
            
    split => {
            "message"=>"|"}
  }
}
output {
            
  stdout {
             codec => rubydebug }
}

4、启动 logstash

# 切换到 logstash 安装目录
$ cd /dzs168/logstash-6.5.4/

# 创建新目录
$  mkdir /dzs168/logstash/1ogs/ -p

# 启动 logstash 
$  ./bin/logstash -f dzs168-pipeline.conf

5、创建 logstash 日志

# 打开另一个终端窗口，创建日志

# 切换到 logstash 日志目录
$ cd /dzs168/logstash/1ogs/

# 创建日志
$  echo "2025-03-15 21:21:21|ERROR|读取数据出错|参数:id=1002" >> app.log

# 查看 logstash 日志记录，输出结果如下（可以看到数据已经被分割）：

{
            
       "message" => [
        [0] "2025-03-15 21:21:21",
        [1] "ERROR",
        [2] "读取数据出错",
        [3] "参数:id=1002"
    ],
      "@version" => "1",
    "@timestamp" => 2025-05-12T15:48:28.676Z,
          "host" => "work1",
          "path" => "/dzs168/logstash/1ogs/app.log"
}

6、修改Logstash 配置文件，输出日志到 ElasticSearch

# 切换到 logstash 安装目录
$ cd /dzs168/logstash-6.5.4/

$  vim dzs168-pipeline.conf

input {
            
  file {
            
    path => "/dzs168/logstash/1ogs/app.log"
    start_position => "beginning"
  }
}
filter {
            
  mutate {
            
    split => {
            "message"=>"|"}
  }
}
#output {
            
#  stdout { codec => rubydebug }
#}
output {
            
  elasticsearch {
            
    hosts => [ "192.168.130.250:9200","192.168.130.71:9200","192.168.130.207:9200"]
  }
}

7、重新启动 logstash

# 切换到 logstash 安装目录
$ cd /dzs168/logstash-6.5.4/

# 启动 logstash 
$  ./bin/logstash -f dzs168-pipeline.conf

8、打开另一个终端窗口，创建 logstash 日志

# 打开另一个终端窗口，创建日志

# 切换到 logstash 日志目录
$ cd /dzs168/logstash/1ogs/

# 创建日志
$  echo "2025-03-16 21:21:21|ERROR|读取数据出错|参数:id=1003" >> app.log
$  echo "2025-03-17 21:21:21|ERROR|读取数据出错|参数:id=1004" >> app.log