ELK整合kafka

编写logstash到kafka的接口，如下图所示。
[root@rac01 ~]# cat kafka.conf 
input {
    stdin {
        add_field => {"key" => "value"}
        codec => "plain"
        tags => ["add"]
        type => "std"
    }
}
output {
        stdout { codec=> rubydebug }
kafka {
        broker_list => "192.168.56.104:9092"
        topic_id => "test"
        compression_codec => "snappy"
}
}

启动logstash，输入
[root@rac01 ~]# /usr/local/logstash-1.5.0/bin/logstash agent -f kafka.conf 
Logstash startup completed
this is kafka test
{
       "message" => "this is kafka test",
      "@version" => "1",
    "@timestamp" => "2015-11-03T02:35:18.006Z",
          "type" => "std",
          "tags" => [
        [0] "add"
    ],
           "key" => "value",
          "host" => "rac01"
}

查看kafka接受端：
[root@rac01 bin]# ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
{"message":"this is kafka test","@version":"1","@timestamp":"2015-11-03T02:35:18.006Z","type":"std","tags":["add"],"key":"value","host":"rac01"}

当然logstash也可以从文件中读取数据，其conf文件为
[root@rac01 ~]# cat logstash-kafka.conf 
input {
file {
type =>"syslog"
      path => ["/home/a.log"]
      start_position => "beginning"
      }
}
output {
        stdout { codec=> rubydebug }
kafka {
        broker_list => "192.168.56.104:9092"
topic_id => "test"
        compression_codec => "snappy"
}
}