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 }
}