logstash的-dissect

描述

Dissect过滤器是一种拆分操作。与常规拆分操作(其中一个分隔符应用于整个字符串)不同，此操作将一组分隔符应用于字符串值。Dissect不使用正则表达式，速度非常快。

filter {
    dissect {
        mapping => {
            "message" => "%{ts} %{+ts} %{+ts} %{src} %{} %{prog}[%{pid}]: %{msg}"
        }
    }
}

语法解释

我们看到上面使用了和Grok很类似的%{}语法来表示字段，这显然是基于习惯延续的考虑。不过示例中%{+ts}的加号就不一般了。dissect 除了字段外面的字符串定位功能以外，还通过几个特殊符号来处理字段提取的规则：

%{+key}这个+表示，前面已经捕获到一个key字段了，而这次捕获的内容，自动添补到之前 key 字段内容的后面。
%{+key/2}这个/2表示，在有多次捕获内容都填到 key字段里的时候，拼接字符串的顺序谁前谁后。/2表示排第2位。
%{}是一个空的跳过字段。
%{?string}这个?表示，这块只是一个占位，并不会实际生成捕获字段存到事件里面。
%{?string} %{&string}当同样捕获名称都是string，但是一个?一个&的时候，表示这是一个键值对。

填充符

字段的->后缀例如%{function->}，表示忽略它右边的填充，否则右边的多余填充将拆分到下一个字段中。

例如有填充->：

%{id} %{function->} %{server}

输出：

{
        "server" => "machine-123",
    "@timestamp" => 2019-01-17T15:04:01.694Z,
      "function" => "Calc",
      "@version" => "1",
          "host" => "VM_0_10_centos",
            "id" => "f3000a3b",
       "message" => "f3000a3b Calc            machine-123"
}

例如无填充->：

%{id} %{function} %{server}

输入：

f3000a3b Calc            machine-123

输出：

{
        "server" => "           machine-123",
    "@timestamp" => 2019-01-17T15:04:53.972Z,
      "function" => "Calc",
      "@version" => "1",
          "host" => "VM_0_10_centos",
            "id" => "f3000a3b",
       "message" => "f3000a3b Calc            machine-123"
}

参数

convert_datatype

使用此设置int和float数据类型进行转换。

filter {
    dissect {
        convert_datatype => {
            "cpu" => "float"
            "code" => "int"
        }
    }
}

mapping

解析事件。如果需要匹配换行符，使用单引号定义值，并在单引号内使用实际的换行符来匹配换行，如下：

filter {
    dissect {
        mapping => {
            # 单引号中使用实际换行匹配换行
            "message" => '"%{field1}" "%{field2}"
            "%{description}"'
            "description" => "%{field3} %{field4} %{field5}"
        }
    }
}

使用 dissect 导入 CSV 格式文档

test.csv

"device1","London","Engineering","Computer"
"device2","Toronto","Consulting","Mouse"
"device3","Winnipeg","Sales","Computer"
"device4","Barcelona","Engineering","Phone"
"device5","Toronto","Consulting","Computer"
"device6","London","Consulting","Computer"

配置logstash

input {
    stdin{}
}
 
filter {
  mutate {
    gsub => [ 
      "message", """, ""
    ]
  }
 
  dissect {
    mapping => {
      "message" => "%{Device_ID},%{Device_Location},%{Device_Owner},%{Device_Type}"
    }
  }
 
  mutate {
    remove_field => ["message"]
  }
}
 
output {
  stdout { 
    codec => "rubydebug"
  }
  
  elasticsearch {
    index => "devices"
  }
}

就像上面展示的那样，它接受一个从 stdin 输入的数据，并使用 filters：

mutate - gsub：把输入的 message 中的引号去掉
dissect: 提取相应的字段，所有的字段是以逗号分开的
mutate - remove_field：去掉 message 字段

我们可以使用如下的方式来运行 Logstash:

cat test.csv | sudo ./bin/logstash -f ./logstash_dissect_csv.conf

我们可以在 Logstash 的 console 中看到如下的输出：