ELK+filebeat+kafka+zooKeeper搭建（单机版）

日薪月亿

发布于 2019-05-14 11:40:35

1.2K0

发布于 2019-05-14 11:40:35

本教程中使用的ELK和filebeat版本是 6.2.4

1. elk安装

关于elk的配置参考我之前的一篇文章，不在累述： elk安装地址： https://jjlu521016.github.io/2018/05/01/springboot-logback-log4j-elk.html#2-elk%E9%85%8D%E7%BD%AE

2. kafka及zookeeper安装

参考我之前的文章、把对应的配置改成单机即可：

2.1. 安装kafka：

https://jjlu521016.github.io/2018/01/12/%E8%BF%90%E7%BB%B4/Kafka%E9%9B%86%E7%BE%A4%E6%90%AD%E5%BB%BA%E6%AD%A5%E9%AA%A4.html#more

2.2. 安装zookeeper：

https://jjlu521016.github.io/2018/01/01/%E8%BF%90%E7%BB%B4/zookeeper%E9%9B%86%E7%BE%A4%E6%90%AD%E5%BB%BA%E6%AD%A5%E9%AA%A4.html#more

2.3 启动zookeeper

我本地zookeeper配置了环境变量

zkServer.sh start

2.4 启动kafka

启动kafka

# bin/kafka-server-start.sh config/server.properties

创建topic

# bin/kafka-topics.sh --create --zookeeper 192.168.188.110:2181 --replication-factor 1 --partitions 1 --topic test

创建生产者

# bin/kafka-console-producer.sh --broker-list 192.168.188.110:9092 --topic test

创建消费者

bin/kafka-console-consumer.sh --zookeeper 192.168.188.110:2181 --topic test --from-beginning

此时生产者输入的内容可以在消费者输出，证明kafka调用通了。

3. 集成logstash与kafka

参考： https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html vi logstash_agent.conf

input {
  kafka {
    bootstrap_servers => "192.168.188.110:9092"
    #topic id
    topics => "test"
    codec => "json"
  }
}
filter {
    json {
      source => "message"
      remove_field => "message"
    }
}
output {
   elasticsearch{
      hosts => ["192.168.188.110:9200"]
	    index => "kafka-%{+YYYY.MM.dd}"
    }
   stdout { codec => rubydebug }
}

启动elk之后，在kafka生产者输入一条信息之后logstash打印出对应的内容，说明集成成功了。在kibana设置对应的日志。

4. 安装filebeat

解压

tar -zxvf filebeat-6.2.3-linux-x86_64.tar.gz

配置filebeat.yml

filebeat.prospectors:
# 这里面配置实际的日志路径
- type: log
  enabled: true
  paths:
    - /var/log/test1.log
  fields:
    log_topics: test1
- type: log
  enabled: true
  paths:
    - /var/log/test2.log
  fields:
    log_topics: test2
output.kafka:
  enabled: true
  hosts: ["10.112.101.90:9092"]
  topic: '%{[fields][log_topics]}'

运行

./filebeat -e -c filebeat.yml

把logstash停止，修改配置文件，添加filebeat配置的topic。重启logstash。打开kibana刷新页面，看到filebeat收集到的日志。

注意：在logstash中，当input里面有多个kafka输入源时，client_id => “xxx”必须添加且需要不同，否则会报错javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=logstash-0。

input {
  kafka {
    
     group_id => "es1"
   
  }
 
  kafka {
     client_id => "es2"
  }
}

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2018-05-09，如有侵权请联系 cloudcommunity@tencent.com 删除

kafka

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度