filebeat+kafka简单使用

filebeat6.3.1+kafka1.1.0简单使用

前提

kafka1.1.0版本集群搭建和常用命令

下载

下载页面：https://www.elastic.co/cn/downloads/past-releases/filebeat-6-3-1
wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.3.1-linux-x86_64.tar.gz

使用

解压

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

查看当前配置egrep -v "#|^$" filebeat.yml

name: "192.168.97.147"
processors: #这个地方需要注意，此配置是将日志输出格式过滤掉，一般情况下，一些无用的日志字段我们可以删除，只看关键性的信息
- drop_fields:
fields: ["beat", "input_type", "source", "offset"]
filebeat.inputs:
- type: log
enabled: true
paths:
- /home/www/log/song/*.log
fields:
log_topic: songTopic #定义Kafka消息队列的topic主题名称，如果kafka消息队列中没有该名称，会自动生成
json.keys_under_root: true # 默认情况下，解析的JSON位于输出文档中的“json”键下。如果启用此设置，则会在输出文档中将键复制到顶层。
json.overwrite_keys: true # 如果启用了keys_under_root和此设置，则解码的JSON对象中的值将覆盖Filebeat通常添加的字段（类型，源，偏移等）以防发生冲突。
filebeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: false
setup.template.settings:
index.number_of_shards: 3
setup.kibana:
output.kafka: #多种输出类型，可支持想kafka，logstash，elasticsearch输出日志信，在这里是将日志信息输出到Kafka中，
enabled: true
hosts: ["192.168.97.211:9192","192.168.97.212:9192","192.168.97.213:9192"]
topic: '%{[fields][log_topic]}' #指定要发送数据到kafka集群的哪个topic，与上述的"fields: log_topic："相对应，这是6.x的配置
partition.round_robin: #开启kafka的partition分区
reachable_only: true
worker: 2
required_acks: 1
compression: gzip
#压缩格式
max_message_bytes: 10000000
#压缩格式字节大小

启动
[root@peer3 filebeat-6.3.1-linux-x86_64]# nohup ./filebeat -e -c filebeat.yml &

参考：

1. Filebeat6.31整合Kafka集群消息队列（三）

最后

以上就是大意高跟鞋为你收集整理的filebeat+kafka简单使用的全部内容，希望文章能够帮你解决filebeat+kafka简单使用所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。