logstash读取kafka的topics,根据内容提取指定字段然后自动创建es索引。

input {
kafka{
bootstrap_servers => "192.168.1.15:9092"
auto_offset_reset => "latest"
topics_pattern => "svc.*" #topics_pattern支持正则匹配，topics不支持
consumer_threads => 5
codec => "json"
}

}
filter {
mutate {

gsub => [

"fieldname", "#", "-"

] 用于替换指定字符

     split => ["message","#"]          #分割字符串获取服务名
     add_field => {  "service" => "%{[message][3]}" }
 }        
#下面移除不必要的字段
 mutate {
 remove_field => ["@version"]
 remove_field => ["@timestamp"]
 remove_field => ["tags"]
 remove_field => ["_id"]
 remove_field => ["_type"]
 remove_field => ["_index"]
 remove_field => ["_score"]

}

}

output {
elasticsearch{
hosts => "192.168.1.15:9200"
index => "log-%{service}"
}
stdout {
codec => rubydebug
}
}