Logstash与Kafka集成

2018-05-14 15:59:19 浏览数 (1)

在ELKK的架构中，各个框架的角色分工如下： ElasticSearch1.7.2：数据存储 全文检索 聚合计算 服务端 Logstasch2.2.2：日志收集与分发推送 Kafka0.9.0.0：分布式高可靠消息队列 数据中转存储（失效期默认7天，可配置时间或大小来控制删除策略） Kibana4.1.2：全文检索 查询 图形化页面展示 客户端 拓扑架构如下：

本篇主要讲logstash与kafka的集成： （1）logstash作为kafka的生产者，就是logstash收集的日志发送到kafka中 （2）logstash作为kafka的消费者，消费kafka里面的数据打印到终端 （一）安装kafka集群，请参考散仙上篇文章： http://qindongliang.iteye.com/blog/2278194 （二）安装logstash 这个非常简单，直接下载最新版的logstash，经测试logstash1.5.4有问题，不能正常安装插件 wget https://download.elastic.co/logstash/logstash/logstash-2.2.2.tar.gz 为了能够快速下载logstash的相关插件，然后修改logstash的代理 （方案一） 安装ruby的gem yum -y install ruby rubygems 安装国内淘宝的代理源： gem sources --remove http://rubygems.org/ gem sources -a https://ruby.taobao.org/ gem sources -l

Java代码

  1. *** CURRENT SOURCES ***
  2. https://ruby.taobao.org/

（方案二） 修改logstash目录下的Gemfile里面的source的url为 https://ruby.taobao.org/ 然后就不用用方案一的方法了 最新版的logstash2.2支持修改Gemfile里面的地址为淘宝的镜像地址 使用的是最新版本2.2.2的logstash

Java代码

  1. //安装logstash输出到kafka的插件：
  2. bin/plugin install logstash-output-kafka
  3. //安装logstash从kafka读取的插件：
  4. bin/plugin install logstash-input-kafka

logstash-consume-kafka.conf消费者配置

Java代码

  1. input{
  2. kafka{
  3. //zk的链接地址
  4. zk_connect=>"h1:2181,h2:2181,h3:2181/kafka"
  5. //topic_id，必须提前在kafka中建好
  6. topic_id=>'logstash'
  7. //解码方式json，
  8. codec => json
  9. //消费者id，多个消费者消费同一个topic时，做身份标识
  10. consumer_id => "187"
  11. //消费者组
  12. group_id=> "logstash"
  13. //重新负载时间
  14. rebalance_backoff_ms=>5000
  15. //最大重试次数
  16. rebalance_max_retries=>10
  17. }
  18. }
  20. output{
  21. stdout{
  22. codec=>line
  23. }
  24. }

procuder_kafka_es.conf生产者配置：

Java代码

  1. input{
  2. //监听log文件
  3. file{
  4. path=> ["/ROOT/server/logstash-2.2.2/t.log"]
  5. }
  6. }
  8. output{
  9. //输出端1=>Kafka
  10. kafka{
  11. bootstrap_servers=> 'h1:9092,h2:9092,h3:9092'
  12. topic_id=> 'logstash'
  13. }
  15. //输出端2=>ElasticSearch
  16. elasticsearch{
  17. hosts=> ["192.168.1.187:9200","192.168.1.184:9200","192.168.1.186:9200"]
  18. }
  20. }

如果想用Logstash读取kafka某个topic的所有数据，需要加上下面2个配置：

Java代码

  1. auto_offset_reset => 'smallest'
  2. reset_beginning => true

但需要注意的是，如果是读取所有的数据，那么此时，对于kafka的消费者同时只能有一个，如果有多个 那么会报错，因为读取所有的数据，保证顺序还不能重复读取消息，只能使用一个消费者，如果不是 读取所有，仅仅读取最新传过来的消息，那么可以启动多个消费者，但建议消费者的数目，与该topic的 partition的个数一致，这样效果最佳且能保证partition内的数据顺序一致，如果不需要保证partition分区内数据 有序，可以接受乱序，那就无所谓了 参考资料 http://bigbo.github.io/pages/2015/08/07/logstash_kafka_new/ http://soft.dog/2016/01/08/logstash-plugins/ http://www.rittmanmead.com/2015/10/forays-into-kafka-01-logstash-transport-centralisation/

es2 存储

0 人点赞