摘要：
输入｛kafka｛bootstrap_servers=˃“192.168.32.36:9092192.168.32.37:9092192.168.38.32.38:9092”topics=˃“msa log prod”编解码器=˃“json”group_id=˃“msa-log prod elsearch”consumer_threads=˃4decorate_events=˃true｝kafka｛boottrap_servers˃˃“192.168.32.36：9092192.162.32.37:909219 2.168.32.38:902”topics=˃“msa日志测试”编解码器-˃“json“group_id=˃”msa log test elsearch“consumer_tthreads=˃2decorate_eevents=˃true}｝｝输出｛if[inv]==“prod”｛elasticsearch｛hosts=˃[“192.168.32.36:9200”，“192.168.32.37:9200”、“192.168.32.38:9200”]index=˃“msa log prod-%｛+YYYY.MM.dd｝”｝｝if[ienv]==”test“｛elaticsearch｛hosts=˃“192.168.32.36：9200”，“92.16832.37:92000”，“”192.168.32.38：9200“]index=˃”msa log test-%｛+YYYY.MM.mdd｝“｝描述：kafka消息格式应为json并包含env要区分的字段如果只有一个输入和输出，则可以删除If判断。
input
{
kafka
{
bootstrap_servers => "192.168.32.36:9092,192.168.32.37:9092,192.168.32.38:9092"
topics => "msa-log-prod"
codec => "json"
group_id => "msa-log-prod-elsearch"
consumer_threads => 4
decorate_events => true
}
kafka
{
bootstrap_servers => "192.168.32.36:9092,192.168.32.37:9092,192.168.32.38:9092"
topics => "msa-log-test"
codec => "json"
group_id => "msa-log-test-elsearch"
consumer_threads => 2
decorate_events => true
}
}
output
{
if [env] == "prod" {
elasticsearch
{
hosts => ["192.168.32.36:9200","192.168.32.37:9200","192.168.32.38:9200"]
index => "msa-log-prod-%{+YYYY.MM.dd}"
}
}
if [env] == "test" {
elasticsearch
{
hosts => ["192.168.32.36:9200","192.168.32.37:9200","192.168.32.38:9200"]
index => "msa-log-test-%{+YYYY.MM.dd}"
}
}
}说明：需要kafka消息格式是json并且包含一个env字段用于区分环境，如果仅一个input和output可以去掉if判断。