logstashkafka到hive(hdfs)的配置文件

input {
kafka {
auto_offset_reset => "earliest"
codec => "plain"
group_id => "es1"
topics => ["mytopic"]
bootstrap_servers => "10.10.4.128:9092,10.10.4.190:9092,10.10.4.197:9092"
}
}
output {
webhdfs {
codec => plain {
format => "%{message}"
}
host => "10.10.4.128"
port => 50070
user => "hdfs"
path => "/data/jsonTestDate/%{+YYYY}-%{+MM}-%{+dd}/mytopic-%{+HH}.log"
flush_size => 500
idle_flush_time => 10
retry_interval => 0.5
}
#stdout { codec => rubydebug }
}