logstash從kafka取數據輸出到es

input {
  kafka {
    bootstrap_servers => ["192.168.13.139:9092"]
    client_id => "test"
    group_id => "test"
    auto_offset_reset => "latest"
    consumer_threads => 5
    decorate_events => true
    topics => ["hbtest"]
    type => "bhy"
    codec => "json"   
    }
}
filter {
  grok {
    match => {"message"=>"%{COMBINEDAPACHELOG}"}
  }
}
output {
  elasticsearch {
    hosts => ["192.168.13.139:9200"]
    index => "%{app_name}-%{+YYYY.MM.dd}"
    }
}