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

1、rename

对于已经存在的字段，重命名其字段名称。

filter {mutate {rename => ["syslog_host", "host"]}
}

2、update

更新字段内容，如果字段不存在，不会新建

filter {mutate {update => { "sample" => "My new message" }}
}

3、replace

与 update 功能相同，区别在于如果字段不存在则会新建字段

filter {mutate {replace => { "message" => "%{source_host}: My new message" }}
}

4、convert

数据类型转换。

filter {mutate {convert => ["request_time", "float"]}
}

5、gsub

gsub 提供了通过正则表达式实现文本替换的功能。

filter {mutate {gsub => [# replace all forward slashes with underscore"fieldname", "/", "_",# replace backslashes, question marks, hashes, and minuses# with a dot ".""fieldname2", "[\\?#-]", "."]}
}

6、uppercase/lowercase

大小写转换

filter {mutate {uppercase => [ "fieldname" ]}
}

7、split

将提取到的某个字段按照某个字符分割

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

针对字符串 "123|321|adfd|dfjld*=123"，可以看到输出结果：

{"message" => [[0] "123",[1] "321",[2] "adfd",[3] "dfjld*=123"],"@version" => "1","@timestamp" => "2014-08-20T15:58:23.120Z","host" => "raochenlindeMacBook-Air.local"
}

8、strip

类似 trim，只去除首尾的空白字符

filter {mutate {strip => ["field1", "field2"]}
}

9、remove_field

删除字段：

filter {mutate {remove_field => [ "foo_%{somefield}" ]}
}

10、join

将类型为 array 的字段中的 array 元素使用指定字符为分隔符聚合成一个字符串。
如我们可以将 split 分割的结果再重新聚合起来：

filter {mutate {split => ["message", "|"]}mutate {join => ["message", ","]}
}

输出结果：

{"message" => "123,321,adfd,dfjld*=123","@version" => "1","@timestamp" => "2014-08-20T16:01:33.972Z","host" => "raochenlindeMacBook-Air.local"
}

11、merge

对于几个类型为 array 或 hash 或 string 的字段，我们可以使用 merge 合并

filter {mutate {merge => [ "dest_field", "added_field" ]}
}

需要注意的是，array 和 hash 两个字段是不能 merge 的

注意：建议正则放在单引号内，例如'^\[?[0-9][0-9]:?[0-9][0-9]|^[[:graph:]]+'。

架构一：
filebeat -> logstash1 -> redis -> logstash2 -> elasticsearch(集群) -> kibana
这里就不写安装程序的步骤了相信大家都没有难度：
（软件安装可自行设计）
230，安装filebeat, logstash1 ,elasticsearch
232，安装logstash2, redis, elasticsearch  ,kibana注意：filebeat文件很注重文件格式
1，配置filebeat文件：
[root@localhost filebeat]# cat /etc/filebeat/filebeat.yml
filebeat:prospectors:# - #每个日志文件的开始#   paths: #定义路径#     - /var/www/logs/access.log #绝对路径#   input_type: log #日志类型为log#   document_type: api4-nginx-accesslog # 此名称要与logstash定义的名称相对应，logstash要使用此名称做type判断使用-paths:- /opt/apps/huhu/logs/ase.loginput_type: logdocument_type: "ase-ase-log"encoding: utf-8tail_files: true  #每次最后一行multiline.pattern: '^\[' #分割符multiline.negate: truemultiline.match: after    #最后合并#tags: ["ase-ase"]-paths:   #收集json格式日志- /var/log/nginx/access.loginput_type: logdocument_type: "nginx-access-log"tail_files: truejson.keys_under_root: true      json.overwrite_keys: true  registry_file: /var/lib/filebeat/registry
output:      #输出到230logstash:hosts: ["192.168.0.230:5044"]shipper:logging:to_files: truefiles:path: /tmp/mybeat2.配置230:logstash-->input-redis
[root@web1 conf.d]# pwd
/etc/logstash/conf.d
[root@web1 conf.d]# cat nginx-ase-input.conf 
input {beats {port => 5044codec => "json"}}output {                         if [type] == "nginx-access-log" {redis {                            #nginx日志写到redis信息data_type => "list"key => "nginx-accesslog"host => "192.168.0.232"port => "6379"db => "4"password => "123456"}}if [type] == "ase-ase-log" {redis {                            #写到ase日志写到redis信息data_type => "list"key => "ase-log"host => "192.168.0.232"port => "6379"db => "4"password => "123456"}}      }3.redis写到elstach里，232服务器配置:logstash-->output-->resid->elasticsearch
[root@localhost conf.d]# pwd
/etc/logstash/conf.d
[root@localhost conf.d]# cat nginx-ase-output.conf 
input {redis {type => "nginx-access-log"data_type => "list"key => "nginx-accesslog"host => "192.168.0.232"port => "6379"db => "4"password => "123456"codec  => "json"}redis {type => "ase-ase-log"data_type => "list"key => "ase-log"host => "192.168.0.232"port => "6379"db => "4"password => "123456"}
}output {if [type] == "nginx-access-log" { elasticsearch {  hosts => ["192.168.0.232:9200"] index => "nginx-accesslog-%{+YYYY.MM.dd}" }}if [type] == "ase-ase-log" {elasticsearch {hosts => ["192.168.0.232:9200"]index => "ase-log-%{+YYYY.MM.dd}"}}
}4，在232上配置elsaticsearch--->kibana
在kibana上找到ELS的索引即可。架构二：
filebeat -> redis -> logstash --> elsasctic --> kibana  #缺点filebeat写进redis有限制，占时还没找到多个写入。1.feilebeat配置：
[root@localhost yes_yml]# cat filebeat.yml 
filebeat:prospectors:# - #每个日志文件的开始#   paths: #定义路径#     - /var/www/logs/access.log #绝对路径#   input_type: log #日志类型为log#   document_type: api4-nginx-accesslog # 此名称要与logstash定义的名称相对应，logstash要使用此名称做type判断使用-paths:- /opt/apps/qpq/logs/qpq.loginput_type: logdocument_type: "qpq-qpq-log"encoding: utf-8tail_files: truemultiline.pattern: '^\['multiline.negate: truemultiline.match: after#tags: ["qpq-qpq-log"]registry_file: /var/lib/filebeat/registryoutput:redis:host: "192.168.0.232"port: 6379db: 3password: "123456"timeout: 5reconnect_interval: 1index: "pqp-pqp-log"shipper:logging:to_files: truefiles:path: /tmp/mybeat2.由232redis-->els--kibana
[root@localhost yes_yml]# cat systemlog.conf 
input {redis {type => "qpq-qpq-log"data_type => "list"key => "qpq-pqp-log"host => "192.168.0.232"port => "6379"db => "3" password => "123456"}}
output {if [type] == "qpq-qpq-log"{elasticsearch {  hosts => ["192.168.0.232:9200"] index => "qpq-qpq-log-%{+YYYY.MM.dd}" }}
}3.在232上配置elsaticsearch--->kibana
在kibana上找到ELS的索引即可