Kafka+Logstash整合

1.启动ZK
./zkServer.sh start

2.daemon形式启动Kafka
kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

3.创建topic：zk
kafka-topics.sh --create --zookeeper hadoop:2181 --replication-factor 1 --partitions 1 --topic logstash_topic

4.发送消息：broker
kafka-console-producer.sh --broker-list hadoop:9092 --topic logstash_topic

5.消费消息：zk
kafka-console-consumer.sh --zookeeper hadoop:2181 --topic logstash_topic --from-beginning

6.消费消息：zk
kafka-console-consumer.sh --zookeeper hadoop:2181 --topic logstash_topic --from-beginning

7.vi kafka.conf编辑配置文件

input{
 file {
   path => "/home/feiyue/app/logstash-2.4.1/test.txt"
 }
}

output {
 kafka {
  topic_id => "logstash_topic"
  bootstrap_servers => "hadoop:9092"
  batch_size => 1
 }
}

8.启动logstash，bin/logstash -f kafka.conf

9.新开窗口往第7步的test.txt写入内容，echo "kkkkk" >> test.txt，在kafka-console-consumer.sh启动窗口看到json形式的内容输出

参考：https://www.elastic.co/guide/en/logstash/2.4/plugins-outputs-kafka.html