如果你对 Logstash 还没有了解的话，请参阅我之前的文章 “Logstash：Data转换，分析，提取，丰富及核心操作”。在今天的文章中，我们将介绍 Logstash 中的 mutate 过滤器插件。

在数据管道中使用 Logstash 的好处之一是能够根据系统和组织的需求将数据转换为所需的格式。在 Logstash 中有多种转换数据的方法，其中一种是使用 mutate 过滤器插件。

这个 Logstash 过滤器插件允许您将字段强制为特定的数据类型，并添加，复制和更新特定的字段以使其在整个环境中兼容。这是一个使用过滤器重命名 IP 字段 HOST_IP 的简单示例。

mutate {rename => { “IP” => “HOST_IP” }
}

在本文中，我将通过三个示例说明如何设置和使用 mutate 过滤器，这些示例说明了可以执行的字段更改的类型。

基础知识

在之前的文章 “Logstash：Data转换，分析，提取，丰富及核心操作” 中，我介绍了如何获得 Logstash 的 filter :

./bin/logstash-plugin list  | grep -i mutate

我们在 Logstash 的安装目录中打入如上的命令。查询结果显示：

logstash-filter-mutate

mutate 过滤器及其不同的配置选项在 Logstash 配置文件的过滤器（filter）部分中定义。可用的配置选项将在本文后面介绍。不过，在深入探讨这些内容之前，让我们简要了解一下 Logstash 配置文件的布局。

通常，Logstash 配置文件包含三个主要部分：

输入（input）–这是识别要处理的数据源的地方。
过滤器（filter）-可以在此处转换和处理传入事件日志的字段。
输出（output）–这是将解析的数据转发到的位置。

可以在链接找到有关格式化 Logstash 配置文件的更多信息。

按照目前的官方文档中所描述的，配置文件中的 mutate 按以下顺序执行：

coerce
rename
update
replace
convert
gsub
uppercase
capitalize
lowercase
strip
remove
split
join
merge
copy

但是你可以通过使用不同的 mutate 块来控制这个顺序，比如：

filter {mutate {split => ["hostname", "."]add_field => { "shortHostname" => "%{hostname[0]}" }}mutate {rename => ["shortHostname", "hostname" ]}
}

在上面，按理 rename 的优先级是比较高的，如果我们把它放入到第一个 mutate 块中，那么它就会被先执行，但是在它执行时，hostname 可能还没有存在。在这种情况下，我们可以使用另外一个 mutate 的块来重新定义这种执行的顺序。

在下面的示例中，我们将标签（Apache Web服务器）添加到传入的 apache 访问日志中，条件是源路径包含术语 “apache”。请注意在 Logstash 配置文件的 filter 部分中添加的 mutate 过滤器：

input {file {path => "/var/log/apache/apache_access.log"start_position => "beginning"sincedb_path => "NULL" }
}
filter {if [ source ] =~ /apache/ {mutate {add_tag => [ "Apache Web Server" ] }}
}output {elasticsearch { hosts => ["localhost:9200"] }
}

Mutate 过滤器配置选项

mutate 过滤器可以使用许多配置选项，例如copy，rename，replace，join, uppercase 及 lowercase。

下表概述了它们：

Mutate 过滤器配置选项
配置选项	用途
add_field	向事件添加新字段
remove_field	从事件中删除任意字段
add_tag	向事件添加任意标签
remove_tag	从事件中删除标签（如果存在）
convert	将字段值转换为另一种数据类型
id	向现场事件添加唯一的ID
lowercase	将字符串字段转换为其小写形式
replace	用新值替换字段
strip	删除开头和结尾的空格
uppercase	将字符串字段转换为等效的大写字母
update	用新值更新现有字段
rename	重命名事件中的字段
gsub	用于查找和替换字符串中的替换
merge	合并数组或 hash 事件

用例

简单的有条件的删除

我们可以仿照我之前的教程 “导入 zipcode CSV 文件和 Geo Search 体验）”。我们使用如下的 Logstash 的配置文件：

logstash_zipcodes.conf

input {file {path => "/Users/liuxg/data/zipcodes/zipcodes.csv"start_position => "beginning"sincedb_path => "/dev/null"}
}filter {csv {separator => ","columns => ["Id", "Code", "AreaCode", "Name", "ShortName", "Longitude", "Latitude", "Sort", "Memo", "Disabled"]}mutate {convert => {"Longitude" => "float"}convert => {"Latitude" => "float"}add_field => ["location", "%{Latitude},%{Longitude}"]rename => ["Code", "zipcode"]}mutate {remove_field => [ "Disabled", "Memo" ]}
}output {stdout {codec => rubydebug}elasticsearch {index => "zipcodes"hosts => ["http://localhost:9200"]}
}

在上面，我们使用了：

 mutate {remove_field => [ "Disabled", ”Memo“ ]}

它将会把 Disabled 及 Memo 这两个字段删除掉。在上面，我们也同时展示了 convert， add_field 及 rename 配置选项。

当我们执行像在文章 “导入 zipcode CSV 文件和 Geo Search 体验）”描述的那样：

sudo ./bin/logstash -f ~/data/zipcodes/logstash_zipcodes.conf

请注意根据自己的环境修改上面的配置文件的路径，那么在屏幕上，我们可以看到：

{"AreaCode" => "522730","ShortName" => "麻芝乡","Longitude" => 107.035835,"location" => "26.457184,107.035835","host" => "liuxg","message" => "35373,522730201,522730,麻芝乡,麻芝乡,107.035835,26.457184,7,,false","Name" => "麻芝乡","Latitude" => 26.457184,"zipcode" => "522730201","Id" => "35373","path" => "/Users/liuxg/data/zipcodes/zipcodes.csv","Sort" => "7","@timestamp" => 2020-06-01T03:38:38.308Z,"@version" => "1"
}

在上面的输出中，我们没有看到字段 “Disabled” 及 “Memo” 字段。它们都被删除了。

除了上面，我们无条件的删除之外，我们也可以进行有条件的删除，比如：

logstash_zipcodes.conf

input {file {path => "/Users/liuxg/data/zipcodes/zipcodes.csv"start_position => "beginning"sincedb_path => "/dev/null"}
}filter {csv {separator => ","columns => ["Id", "Code", "AreaCode", "Name", "ShortName", "Longitude", "Latitude", "Sort", "Memo", "Disabled"]}mutate {convert => {"Longitude" => "float"}convert => {"Latitude" => "float"}add_field => ["location", "%{Latitude},%{Longitude}"]rename => ["Code", "zipcode"]}if[AreaCode] == "522730" {mutate {remove_field => [ "Disabled", "Memo" ]}}
}output {stdout {codec => rubydebug}elasticsearch {index => "zipcodes"hosts => ["http://localhost:9200"]}
}

在上面，我们只针对 AreaCode 为 522730 地区的一些相关字段进行删除，也许我们认为这些是一些敏感的地区，并不需要有些字段被人搜索到。

合并字段

在此示例中，我们将使用 mutate 过滤器通过 merge 选项合并 “AreaCode” 和 “ ShortName” 两个字段。

合并两者后，“ShortName”字段将具有数组格式的合并数据。另外，为了使内容更清楚，我们还将重命名该字段，如下面的代码所示：

logstash_zipcodes.conf

input {file {path => "/Users/liuxg/data/zipcodes/zipcodes.csv"start_position => "beginning"sincedb_path => "/dev/null"}
}filter {csv {separator => ","columns => ["Id", "Code", "AreaCode", "Name", "ShortName", "Longitude", "Latitude", "Sort", "Memo", "Disabled"]}mutate {convert => {"Longitude" => "float"}convert => {"Latitude" => "float"}add_field => ["location", "%{Latitude},%{Longitude}"]rename => ["Code", "zipcode"]}mutate {merge =>  { "ShortName" => "AreaCode" }}mutate {rename => [ "ShortName", "Name-Area" ]}
}output {stdout {codec => rubydebug}elasticsearch {index => "zipcodes"hosts => ["http://localhost:9200"]}
}

我们重新使用上面的配置进行运行 Logstash，我们可以看到：

{"Memo" => nil,"@version" => "1","@timestamp" => 2020-06-01T04:01:13.353Z,"zipcode" => "210114001","path" => "/Users/liuxg/data/zipcodes/zipcodes.csv","message" => "5748,210114001,210114,迎宾路街道,迎宾路街道,123.310829,41.795834,13,,false","Latitude" => 41.795834,"AreaCode" => "210114","host" => "liuxg","Id" => "5748","Name-Area" => [[0] "迎宾路街道",[1] "210114"],"Sort" => "13","Disabled" => "false","Name" => "迎宾路街道","Longitude" => 123.310829,"location" => "41.795834,123.310829"
}

我们可以看到之前的 AreaCode 及 ShortName 两个字段都不见了。取而代之的是一个崭新的字段 Name-Area，它含有 AreaCode 及 ShortName 两个字段的内容。这个有点类似于我们之前介绍的文章 “如何使用 Elasticsearch 中的 copy_to 来提高搜索效率”。我们把不同的字段集中到一个字段，这样可以更方便地进行查询。

添加空格

接下来，我们将使用 mutate 过滤器在传入事件的 “message” 字段中添加空格。当前，“message”字段的值中没有空格。我们将使用 mutate 过滤器的 “GSUB” 选项，如以下代码所示：

logstash_zipcodes.conf

input {file {path => "/Users/liuxg/data/zipcodes/zipcodes.csv"start_position => "beginning"sincedb_path => "/dev/null"}
}filter {csv {separator => ","columns => ["Id", "Code", "AreaCode", "Name", "ShortName", "Longitude", "Latitude", "Sort", "Memo", "Disabled"]}mutate {convert => {"Longitude" => "float"}convert => {"Latitude" => "float"}add_field => ["location", "%{Latitude},%{Longitude}"]rename => ["Code", "zipcode"]}mutate {merge =>  { "ShortName" => "AreaCode" }}mutate {rename => [ "ShortName", "Name-Area" ]}mutate {gsub => [ "message", "," , ", " ]}
}output {stdout {codec => rubydebug}elasticsearch {index => "zipcodes"hosts => ["http://localhost:9200"]}
}

重新运行 Logstash，我们可以看到：

总结

本文已经说明了 mutate 过滤器如何在数据集中创建新字段以及替换和重命名现有字段。 Logstash 中还有许多其他重要的过滤器插件，它们在解析或创建可视化效果时也很有用。

Logstash：使用 mutate 过滤器相关推荐

Logstash~filter.json过滤器使用教程（附带示例）
一.介绍用来解析json格式数据的过滤器,默认会解析置顶字段的json数据并将其放置在Logstash事件的最顶级中,可以配置target配置项选择存放结果位置如果解析的数据包含一个@timest ...
9款日志管理工具大比拼，选型必备！
点击关注公众号,利用碎片时间学习简介对于日志管理当前网络上提供了大量的日志工具,今天就给大家分析总结一下这些常用工具的特点,希望对你们在选型时有所帮助,如果有用记得一键三连. 1.Filebeat ...
这 9款日志管理工具大比拼，选型必备！
大家好,我是不才陈某~ 对于日志管理当前网络上提供了大量的日志工具,今天就给大家分析总结一下这些常用工具的特点,希望对你们在选型时有所帮助,如果有用记得一键三连. 1.Filebeat Filebea ...
9款日志采集和管理工具对比，选型必备！
对于日志管理当前网络上提供了大量的日志工具,今天就给大家分析总结一下这些常用工具的特点,希望对你们在选型时有所帮助. 一.Filebeat Filebeat是用于转发和集中日志数据的轻量级传送程序.作 ...
9 款主流日志管理工具技术选型，yyds！
点击关注公众号:互联网架构师,后台回复 2T获取2TB学习资源! 上一篇:Alibaba开源内网高并发编程手册.pdf 简介对于日志管理当前网络上提供了大量的日志工具,今天就给大家分析总结一下这些常 ...
Logstash：在实施之前测试 Logstash 管道/过滤器
检测解析的日志是否包含单个或多个警告消息,然后添加一个字段来说明这两种情况.在很多的情形下,我们在测试 Logstash 的过滤器时,并不急于把实际的 input 的数据接入到过滤器中来进行测试.我们 ...
5W字穿透 ELK（史上最全）：elasticsearch +logstash+kibana
本文 5w 字,帮忙大家绞杀式.穿透式掌握 elk 的原理和实操文章很长,建议收藏起来慢慢读! 总目录博客园版为您奉上更多の珍贵的学习资源有关本文的脚本和代码,可以来尼恩发起的J ...
大数据——Logstash（日志收集）
Logstash(日志收集) ELK Stack Logstash简介安装Logstash Logstash如何工作 Logstash配置语法(一) Logstash配置语法(二) 输入插件编解码 ...
Logstash的filter插件介绍
一官网说明过滤器插件对事件执行中介处理.通常根据事件的特征有条件地应用过滤器. 以下过滤器插件在下面可用. Plugin Description Github repository aggrega ...
Logstash学习
写在前面:本文是学习总结内容,侵犯到别的文章,在这里说声抱歉~~ 一.基础知识 1.Hello World 运行在终端中,像下面这样运行命令来启动 Logstash 进程: bin/logstash ...

Logstash：使用 mutate 过滤器

基础知识

Mutate 过滤器配置选项

用例

简单的有条件的删除

合并字段

添加空格

总结

Logstash：使用 mutate 过滤器相关推荐

最新文章

热门文章