0%

Logstash过滤器

mutate插件可以对事件中的数据进行修改,包括rename、update、replace、convert、split、gsub、uppercase、lowercase、strip、remove_field、join、merge等功能。

一、概念

      mutate插件可以对事件中的数据进行修改,包括rename、update、replace、convert、split、gsub、uppercase、lowercase、strip、remove_field、join、merge等功能。

  1. rename:对于已经存在的字段,重命名其字段名称。
1
2
3
4
5
filter {
mutate {
rename => ["old_name", "new_name"]
}
}
  1. update:更新字段内容(如果字段不存在不会新建)。
1
2
3
4
5
filter {
mutate {
update => {"old_data" => "new_data"}
}
}
  1. replace:与update功能相同,区别在于如果字段不存在则会新建字段。
1
2
3
4
5
filter {
mutate {
replace => {"message" => "%{source_host}: new_host" }
}
}
  1. convert:数据类型转换。
1
2
3
4
5
filter {
mutate {
convert => ["request_time", "float"]
}
}
  1. gsub:通过正则表达式实现文本替换的功能。
1
2
3
4
5
6
7
8
filter {
mutate {
gsub => [
"fieldname", "/", "_",
"fieldname2", "[\\?#-]", "."
]
}
}
  1. uppercase/lowercase:大小写转换。
1
2
3
4
5
filter {
mutate {
uppercase => [ "fieldname" ]
}
}
  1. split:将提取到的某个字段按照某个字符分割。
1
2
3
4
5
filter {
mutate {
split => ["message", "|"]
}
}
  1. strip:去除首尾的空白字符。
1
2
3
4
5
filter {
mutate {
strip => ["field1", "field2"]
}
}
  1. remove_field:删除字段。
1
2
3
4
5
filter {
mutate {
remove_field => [ "foo_%{somefield}" ]
}
}
  1. join:将类型为array的字段中的元素使用指定字符为分隔符聚合成一个字符串。
1
2
3
4
5
6
7
8
filter {
mutate {
split => ["message", "|"]
}
mutate {
join => ["message", ","]
}
}
  1. merge:合并字段。
1
2
3
4
5
filter {
mutate {
merge => [ "dest_field", "added_field" ]
}
}

二、使用

  1. 下载测试数据
  2. 解压至/Users/your_name/elk/ml-25m/movies.csv
  3. 启动Elasticsearch实例
  4. 修改Logstash配置logstash.conf
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
input {
#beats {
# port => 9011
#}
file {
path => ["/Users/your_name/elk/ml-25m/movies.csv"]
start_position => "beginning"
sincedb_path => null
}
}

filter {
csv {
separator => ","
columns => ["movieId","title","genre"]
}
mutate {
split => { "genre" => "|" }
# remove_field => ["path", "host","@timestamp","message"]
}
mutate {
convert => {
"year" => "integer"
}
strip => ["title"]
#remove_field => ["path", "host","@timestamp","message","content"]
}
}

output {
elasticsearch {
hosts => ["http://localhost:9200"]
index => "movies"
}
stdout {}
}
  1. 启动Logstash实例

  2. 查询

    • curl -XGET "localhost:9200/movies/_search?pretty" -H "content-type:application/json" -d '{"_source":["movieId","title"],"query":{"match":{"title":"liu*"}}}'

    • curl -XGET "localhost:9200/_search?pretty" -H "content-type:application/json" -d '{"_source":["movieId","title"],"query":{"match":{"title":"liu*"}}}'

三、参考

  1. 参考一
  2. 参考二