ELK+kafka集成

1、因为本项目采用的log4j2，所以在log4j2中直接配置 

<Kafka name="Kafka" topic="XX_log">     <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss}||%p||%c{1}||XX_web||%m%n"/>       <Property name="bootstrap.servers">127.0.0.1:9092</Property>       <Property name="timeout.ms">500</Property>   </Kafka>

PatternLayout 中格式采用了||将内容连接起来目的为了logstash进行切分，其中增加timeout.ms属性为了保证日志系统挂掉的情况不会对业务系统产生较大影响，当然kafka可以采用集群的方式，bootstrap.servers多个地址用“,”分隔。XX_web代表当前业务平台。 

2、搭建kafka集群这里就不多介绍了官网很全, 

zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183

3、创建logstash动态模板 

{       "template": "*",       "settings": {           "index.refresh_interval": "5s",           "number_of_replicas": "0",           "number_of_shards": "3"       },       "mappings": {           "_default_": {               "_all": {                   "enabled": false               },               "dynamic_templates": [                   {                       "message_field": {                           "match": "message",                           "match_mapping_type": "string",                           "mapping": {                               "type": "string",                               "index": "analyzed"                           }                       }                   },                   {                       "string_fields": {                           "match": "*",                           "match_mapping_type": "string",                           "mapping": {                               "type": "string",                               "index": "not_analyzed"                           }                       }                   }               ],               "properties": {                   "dateTime": {                       "type": "date",                       "format": "yyy-MM-dd HH:mm:ss"                   },                   "@version": {                       "type": "integer",                       "index": "not_analyzed"                   },                   "context": {                       "type": "string",                       "index": "analyzed"                   },                   "level": {                       "type": "string",                       "index": "not_analyzed"                   },                   "class": {                       "type": "string",                       "index": "not_analyzed"                   },                   "server": {                       "type": "string",                       "index": "not_analyzed"                   }               }           }       }   }

4、配置logstash 

input{          kafka {                   zk_connect =>"127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183"                   group_id =>"logstash"                   topic_id =>"XX_log"                   reset_beginning => false                   consumer_threads => 5                   decorate_events => true          }   }   filter {      mutate{           split=>["message","||"]           add_field => {                "dateTime" => "%{[message][0]}"           }           add_field => {                 "level" => "%{[message][1]}"           }           add_field => {                  "class" => "%{[message][2]}"           }           add_field => {                   "server" => "%{[message][3]}"            }           add_field => {                   "context" => "%{[message][4]}"            }            remove_field => ["message"]       }       date {           match => ["logdate", "yyyy-MM-dd HH:mm:ss"]       }   }   output{         elasticsearch {               hosts => ["127.0.0.1:9200"]               index => "XX_log-%{+YYYY-MM}"               codec => "json"               manage_template => true               template_overwrite => true               flush_size => 50000               idle_flush_time => 10               workers => 2               template => "E:\logstash\template\template_log.json"         }   }

按照年月将日志保存进ES索引中index => "XX_log-%{+YYYY-MM}"，logstash从kafka集群中读取日志信息。 

5、搭建ZK集群，这里就不多介绍了，网上资料比较多

6、搭建ES集群，ES集群比较简单，设置的参数不要太多就可以使用。

7、配置kibana 

server.port: 5601 # 服务端口   # The host to bind the server to.   server.host: "115.28.240.113"   elasticsearch.url: http://127.0.0.1:9200   ES地址-集群   kibana.index: "kibana"

8、版本 JKD 1.7  ES-2.4, logstash 2.4, kafka-2.10,kibana-4.6.4