一、背景

当咱们应用logstash从内部读取到数据后，默认状况下读取到的值都是string的类型，假如咱们这个时候须要批改字段值的类型，如果从string批改成integer，或者删除字段、批改字段的名字、给字段一个默认值等操作时，这个时候咱们就能够借助 mutate filter 来实现。

二、需要

1、从文件中读取数据，文件中的数据合乎csv的格局，即默认是以,分隔。

2、对读取到的字段进行 删除字段、批改字段的值、批改字段的类型、给一个默认值、字段合并等操作。

三、实现步骤

1、装置 csv codec 插件

留神⚠️：

默认状况下,csv codec插件并没有装置，须要咱们本人手动装置一下，执行如下命令bin/logstash-plugin install logstash-codec-csv

# 进入 logstash 的装置目录cd /Users/huan/soft/elastic-stack/logstash/logstash# 监测是否装置了 csv codec 插件bin/logstash-plugin list --verbose# 装置 csv 插件bin/logstash-plugin install logstash-codec-csv

2、筹备须要读取的文件数据

3、编写 pipeline ，读取和输入数据

input {    file {        id => "mutate-id"        path => ["/Users/huan/soft/elastic-stack/logstash/logstash/pipeline.conf/filter-mutate/mutate.csv"]        start_position => "beginning"        sincedb_path => "/Users/huan/soft/elastic-stack/logstash/logstash/pipeline.conf/filter-mutate/sincedb.db"        codec => csv {            columns => ["user_real_name","user_english_name","age","address","education","strip_blank","language","default_value","create_time"]            charset => "UTF-8"            separator => ","            skip_empty_columns => false            convert => {                "age" => "integer"            }        }    }}output {    stdout {        codec => rubydebug {                    }    }}

• csv codec 插件解释

  • columns ：定义一组解析后的csv的列名，也是前期的字段名
  • charset：字符编码
  • separator：定义读取到的一行数据，以什么作为分隔，csv文件个别是以,或tab等进行分隔，默认是逗号

  • skip_empty_columns：如果值为空，是否跳过空列。

    • true：跳过
    • false: 不跳过
  • convert：数据类型转换，默认读取到的值的类型都是string，此处将 age字段的值的数据类型转换成了integer

4、mutate 插件的应用

前置条件：

1、如无非凡阐明，测试数据的数据为 实现步骤>筹备须要读取的文件数据 中的数据

注意事项：

1、update和replace都是更新字段的值，然而如果update更新的字段的值不存在，那么没有成果，然而replace会新减少这个字段。

2、copy字段的目标值，如果存在则笼罩值，否则新减少一个字段。

1、coerce 给字段设置默认值

如果某个字段曾经存在，并且它的值是null,那么咱们能够应用coerce来为它设置默认值

1、配置文件的写法

filter {    mutate {        coerce => {            "default_value" => "该字段没有值，设置一个默认值"        }    }}

2、执行后果

2、rename 给字段重命名

1、配置文件的写法

filter {    mutate {        rename => {            "user_real_name" => "[user][real_name]"            "user_english_name" => "[user][english_name]"            "age" => "年龄"        }    }}

2、执行后果

3、update 更新字段的值

1、配置文件的写法

filter {    mutate {        # 1、更新字段的值        update => {            "user_address" => "用户的地址是: %{address}"        }    }}

2、执行后果

3、解释

update进行更新值，更新的字段必须要存在，否则没有任何成果。

4、replace 更新字段的值

1、配置文件的写法

filter {    mutate {        # 1、更新字段的值        replace => {            "user_address" => "用户的地址是: %{address}"        }     }}

2、执行后果

5、convert 数据类型转换

1、能够转换的数据类型

integer、integer_eu、float、float_eu、string、boolean

2、配置文件的写法

filter {    mutate {        # 1、数据类型转换        convert => {            "age" => "string"        }    }}

3、执行后果

6、gsub 对字段内容进行替换

1、配置文件的写法

filter {    mutate {        # 1、替换字段的内容, 第二个参数能够写正则表达式 ，替换的字段 只能是 string 类型或者 string类型的数组        gsub => [            "address", ";", "--"        ]    }}

2、执行后果

7、uppercase、capitalize、lowercase 大写、首字母大写、小写

1、配置文件的写法

filter {    mutate {        # 1.1 大写        uppercase => ["language"]        # 2.2 小写        # lowercase => ["user_english_name"]        # 3.3 首字母大写        capitalize => ["user_english_name"]    }}

须要留神优先级。

2、执行后果

8、strip 去除首尾空格

1、配置文件的写法

filter {    mutate {        # 去除首尾空格        strip => ["strip_blank"]    }}

2、执行后果

9、remove 移除字段

1、配置文件的写法

filter {    mutate {        # 移除字段 ，如果 Event 中 username 的值是 zhangsan ,那么会移除字段名是 foo_zhangsan 这个字段。        remove_field => ["user_real_name","foo_%{username}"]    }}

2、执行后果

10、split 切割字段

1、配置文件的写法

filter {    mutate {        # 1、切割字段        split => {            "address" => ";"        }    }}

2、执行后果

11、 join 连贯字段

1、配置文件的写法

filter {    mutate {        # 1、切割字段        split => {            "address" => ";"        }        # 2、连贯字段        join => {            "address" => "***"        }    }}

先应用 split 切割成数组，而后应用 join 连贯

2、执行后果

12、merge 字段合并

1、能够合并的状况

`array` + `string` will work`string` + `string` will result in an 2 entry array in `dest_field``array` and `hash` will not work

2、配置文件的写法

filter {    mutate {        # 1、字段合并        merge => {            "user_real_name" => "user_english_name"        }    }}

3、执行后果

13、copy 复制字段

1、配置文件的写法

filter {    mutate {        # 1、字段复制, 如果 user_name 这个字段曾经存在了，那么此字段的值会被笼罩，否则新增一个字段的值        copy => {            "user_real_name" => "user_name"        }    }}

2、执行后果

四、mutate的优先级

1、mutate 在配置文件中的执行程序

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

corece最先执行,copy最初执行。

2、多个mutate块的优先级

filter {    # mutate块1 会比 下方的 mutate2 先执行    mutate {}    # mutate块2    mutate {}}

留神⚠️：

假如 咱们须要字段 age先复制copy一下，而后在转换一下数据类型convert，那么能够应用上方的 多个 mutate 块来执行。

1、配置文件的写法

filter {      # 测试多 mutate 块的优先级    mutate {        copy => {            "age" => "new_age"        }    }    mutate {        convert => {            "age" => "string"        }    }}

将 convert 和 copy 搁置在一个 mutate 块中会发现后果不一样。

2、执行后果

五、参考文档

1、https://www.elastic.co/guide/en/logstash/7.12/working-with-plugins.html

2、https://www.elastic.co/guide/en/logstash/current/plugins-codecs-csv.html

3、https://www.elastic.co/guide/en/logstash/current/plugins-filters-mutate.html