最近需要搭建一套日志监控平台，结合系统本身的特性总结一句话也就是：需要将Kafka中的数据导入到elasticsearch中。那么如何将Kafka中的数据导入到elasticsearch中去呢，总结起来大概有如下几种方式：

• Kafka->logstash->elasticsearch->kibana(简单，只需启动一个代理程序)
• Kafka->kafka-connect-elasticsearch->elasticsearch->kibana(与confluent绑定紧，有些复杂)
• Kafka->elasticsearch-river-kafka-1.2.1-plugin->elasticsearch->kibana(代码很久没更新，后续支持比较差)

 根据以上情况，项目决定采用方案一将Kafka中的数据存入到elasticsearch中去。

一、拓扑图

    项目拓扑图如下所示：

  此时消息的整体流向为：日志/消息整体流向Flume => kafka => logstash => elasticsearch => kibana

二、环境搭建

 关于Kafka及Flume的搭建在这里不再详细论述，如有需要请参见本文其它说明。在这里重点说明logstash的安装及配置。

  A.下载logstash的安装包； 

  B.新建kafka-logstash-es.conf置于logstash/conf目录下；

  C.配置kafka-logstash-es.conf如下：

input {    kafka {        zk_connect => "c1:2181,c2:2181,c3:2181"        group_id => "elasticconsumer"   ---随意取        topic_id => "xxxlog"  ---与flume中的Channel保持一致        reset_beginning => false         consumer_threads => 5          decorate_events => true         codec => "json"        }    }output {    elasticsearch {        hosts => ["c4:9200","c5:9200"]        index => "traceid"--对应kafka的json中真实存在的字段，注意：index必须小写        workers => 5        codec => "json"		  }     }

运行logstash命令为：nohup bin/logstash -f /XXX/logstash/conf/kafka-logstash-es.conf &

三、调测过程中遇到的一些坑

A.在集成ELK过程中总以为head插件是必须的，其实head插件为非必需品。elasticsearch仅提供了一个数据存储的煤介，head为了让大家更方便的去查看数据； 

B.采用以上方案进行布署时，当系统正常运行时，可以在elasticsearch服务器上http://IP:9200/*中搜索index是否创建成功

日志收集之--将Kafka数据导入Elastic