一、背景

当咱们应用 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