网站导航
> 技术文档 > 【Elasticsearch系列廿】Logstash 学习_elasticsearch logstash

【Elasticsearch系列廿】Logstash 学习_elasticsearch logstash

网友: 技术文档 2025-07-30 12:46:30

💝💝💝欢迎来到我的博客，很高兴能够在这里和您见面！希望您在这里可以感受到一份轻松愉快的氛围，不仅可以获得有趣的内容和知识，也可以畅所欲言、分享您的想法和见解。
img

  • 推荐:kwan 的首页,持续学习,不断总结,共同进步,活到老学到老
  • 导航
    • 檀越剑指大厂系列:全面总结 java 核心技术,jvm,并发编程 redis,kafka,Spring,微服务等
    • 常用开发工具系列:常用的开发工具,IDEA,Mac,Alfred,Git,typora 等
    • 数据库系列:详细总结了常用数据库 mysql 技术点,以及工作中遇到的 mysql 问题等
    • 新空间代码工作室:提供各种软件服务,承接各种毕业设计,毕业论文等
    • 懒人运维系列:总结好用的命令,解放双手不香吗?能用一个命令完成绝不用两个操作
    • 数据结构与算法系列:总结数据结构和算法,不同类型针对性训练,提升编程思维,剑指大厂

非常期待和您一起在这个小小的网络世界里共同探索、学习和成长。💝💝💝 ✨✨ 欢迎订阅本专栏 ✨✨

博客目录

      • 1.什么是 Logstash
      • 2.配置文件
      • 3.启动操作
      • 4.input 原理
      • 5.Grok 正则捕获
      • 6.Mutate 插件
      • 7.output
      • 8.综合案例

1.什么是 Logstash

下载地址

输入插件地址

过滤器插件

输出插件

logstash 是一个数据抽取工具，将数据从一个地方转移到另一个地方。如 hadoop 生态圈的 sqoop 等。

logstash 之所以功能强大和流行，还与其丰富的过滤器插件是分不开的，过滤器提供的并不单单是过滤的功能，还可以对进入过滤器的原始数据进行复杂的逻辑处理，甚至添加独特的事件到后续流程中。
Logstash 配置文件有如下三部分组成，其中 input、output 部分是必须配置，filter 部分是可选配置，而 filter 就是过滤器插件，可以在这部分实现各种日志过滤功能。

2.配置文件

input { #输入插件}filter { #过滤匹配插件}output { #输出插件}

标准输入stdin:

input{ stdin{ }}output { stdout{ codec=>rubydebug }}

读取 TCP 网络数据:

input { tcp { port => \"1234\" }}filter { grok { match => { \"message\" => \"%{SYSLOGLINE}\" } }}output { stdout{ codec=>rubydebug }}

3.启动操作

#windows启动logstash.bat -e \'input{stdin{}} output{stdout{}}\'#脚本启动logstash.bat -f ../config/test1.conf

4.input 原理

logstash 使用一个名为 filewatch 的 ruby gem 库来监听文件变化,并通过一个叫.sincedb 的数据库文件来记录被监听的日志文件的读取进度（时间戳），这个 sincedb 数据文件的默认路径在 /plugins/inputs/file 下面，文件名类似于.sincedb_123456，而表示 logstash 插件存储目录，默认是 LOGSTASH_HOME/data。

input { file { path => [\"/var/*/*\"] start_position => \"beginning\" }}output { stdout{ codec=>rubydebug }}

默认情况下，logstash 会从文件的结束位置开始读取数据，也就是说 logstash 进程会以类似 tail -f 命令的形式逐行获取数据。

5.Grok 正则捕获

grok 是一个十分强大的 logstash filter 插件，他可以通过正则解析任意文本，将非结构化日志数据弄成结构化和方便查询的结构。他是目前 logstash 中解析非结构化日志数据最好的方式。

Grok 的语法规则是：

%{语法: 语义}

例如输入的内容为：

172.16.213.132 [07/Feb/2019:16:24:19 +0800] \"GET / HTTP/1.1\" 403 5039

  • %{IP:clientip}匹配模式将获得的结果为：clientip: 172.16.213.132

  • %{HTTPDATE:timestamp}匹配模式将获得的结果为：timestamp: 07/Feb/2018:16:24:19 +0800

  • %{QS:referrer}匹配模式将获得的结果为：referrer: “GET / HTTP/1.1”

下面是一个组合匹配模式，它可以获取上面输入的所有内容：

%{IP:clientip}\\ \\[%{HTTPDATE:timestamp}\\]\\ %{QS:referrer}\\ %{NUMBER:response}\\ %{NUMBER:bytes}

通过上面这个组合匹配模式，我们将输入的内容分成了五个部分，即五个字段，将输入内容分割为不同的数据字段，这对于日后解析和查询日志数据非常有用，这正是使用 grok 的目的。

例子：

input{ stdin{}}filter{ grok{ match => [\"message\",\"%{IP:clientip}\\ \\[%{HTTPDATE:timestamp}\\]\\ %{QS:referrer}\\ %{NUMBER:response}\\ %{NUMBER:bytes}\"] }}output{ stdout{ codec => \"rubydebug\" }}

输入内容：

172.16.213.132 [07/Feb/2019:16:24:19 +0800] \"GET / HTTP/1.1\" 403 5039

时间处理(Date):

date 插件是对于排序事件和回填旧数据尤其重要，它可以用来转换日志记录中的时间字段，变成 LogStash::Timestamp 对象，然后转存到@timestamp 字段里，下面是 date 插件的一个配置示例（这里仅仅列出 filter 部分）：

filter { grok { match => [\"message\", \"%{HTTPDATE:timestamp}\"] } date { match => [\"timestamp\", \"dd/MMM/yyyy:HH:mm:ss Z\"] }}

GeoIP 地址查询归类:

filter { geoip { source => \"ip_field\" }}

kibana测试Grok:

【Elasticsearch系列廿】Logstash 学习_elasticsearch logstash

6.Mutate 插件

数据修改(Mutate):

正则表达式替换匹配字段

gsub 可以通过正则表达式替换字段中匹配到的值，只对字符串字段有效，下面是一个关于 mutate 插件中 gsub 的示例（仅列出 filter 部分）：

filter { mutate { gsub => [\"filed_name_1\", \"/\" , \"_\"] }}

这个示例表示将 filedname_1 字段中所有\"/“字符替换为”\"。

分隔符分割字符串为数组:

split 可以通过指定的分隔符分割字段中的字符串为数组，下面是一个关于 mutate 插件中 split 的示例（仅列出 filter 部分）：

filter { mutate { split => [\"filed_name_2\", \"|\"] }}

这个示例表示将 filed_name_2 字段以\"|\"为区间分隔为数组。

重命名字段:

rename 可以实现重命名某个字段的功能，下面是一个关于 mutate 插件中 rename 的示例（仅列出 filter 部分）：

filter { mutate { rename => { \"old_field\" => \"new_field\" } }}

这个示例表示将字段 old_field 重命名为 new_field。

删除字段:

remove_field 可以实现删除某个字段的功能，下面是一个关于 mutate 插件中 remove_field 的示例（仅列出 filter 部分）：

filter { mutate { remove_field => [\"timestamp\"] }}

这个示例表示将字段 timestamp 删除。

综合例子:

input { stdin {}}filter { grok { match => { \"message\" => \"%{IP:clientip}\\ \\[%{HTTPDATE:timestamp}\\]\\ %{QS:referrer}\\ %{NUMBER:response}\\ %{NUMBER:bytes}\" } remove_field => [ \"message\" ] }date { match => [\"timestamp\", \"dd/MMM/yyyy:HH:mm:ss Z\"] }mutate { convert => [ \"response\",\"float\" ]  rename => { \"response\" => \"response_new\" }  gsub => [\"referrer\",\"\\\"\",\"\"]  split => [\"clientip\", \".\"] }}output { stdout { codec => \"rubydebug\" }

7.output

output 是 Logstash 的最后阶段，一个事件可以经过多个输出，而一旦所有输出处理完成，整个事件就执行完成。 一些常用的输出包括：

  • file： 表示将日志数据写入磁盘上的文件。
  • elasticsearch：表示将日志数据发送给 Elasticsearch。Elasticsearch 可以高效方便和易于查询的保存数据。

输出到标准输出(stdout):

output { stdout { codec => rubydebug }}

保存为文件file:

output { file { path => \"/data/log/%{+yyyy-MM-dd}/%{host}_%{+HH}.log\" }}

输出到 elasticsearch:

output { elasticsearch { host => [\"192.168.1.1:9200\",\"172.16.213.77:9200\"] index => \"logstash-%{+YYYY.MM.dd}\" }}
  • host：是一个数组类型的值，后面跟的值是 elasticsearch 节点的地址与端口，默认端口是 9200。可添加多个地址。
  • index：写入 elasticsearch 的索引的名称，这里可以使用变量。Logstash 提供了%{+YYYY.MM.dd}这种写法。在语法解析的时候，看到以+ 号开头的，就会自动认为后面是时间格式，尝试用时间格式来解析后续字符串。这种以天为单位分割的写法，可以很容易的删除老的数据或者搜索指定时间范围内的数据。此外，注意索引名中不能有大写字母。
  • manage_template:用来设置是否开启 logstash 自动管理模板功能，如果设置为 false 将关闭自动管理模板功能。如果我们自定义了模板，那么应该设置为 false。
  • template_name:这个配置项用来设置在 Elasticsearch 中模板的名称。

8.综合案例

input { file { path => [\"D:/ES/logstash-7.3.0/nginx.log\"] start_position => \"beginning\" }}filter { grok { match => { \"message\" => \"%{IP:clientip}\\ \\[%{HTTPDATE:timestamp}\\]\\ %{QS:referrer}\\ %{NUMBER:response}\\ %{NUMBER:bytes}\" } remove_field => [ \"message\" ] }date { match => [\"timestamp\", \"dd/MMM/yyyy:HH:mm:ss Z\"] }mutate {  rename => { \"response\" => \"response_new\" }  convert => [ \"response\",\"float\" ]  gsub => [\"referrer\",\"\\\"\",\"\"]  remove_field => [\"timestamp\"]  split => [\"clientip\", \".\"] }}output { stdout { codec => \"rubydebug\" }elasticsearch { host => [\"localhost:9200\"] index => \"logstash-%{+YYYY.MM.dd}\"}}

觉得有用的话点个赞 👍🏻 呗。
❤️❤️❤️本人水平有限，如有纰漏，欢迎各位大佬评论批评指正！😄😄😄

💘💘💘如果觉得这篇文对你有帮助的话，也请给个点赞、收藏下吧，非常感谢!👍 👍 👍

🔥🔥🔥Stay Hungry Stay Foolish 道阻且长,行则将至,让我们一起加油吧！🌙🌙🌙

img

植物大战僵尸JS版游戏

网络标签：

相关问题