filebeat+kafka简单使用

89 阅读 0 评论 59 点赞

我是靠谱客的博主大意高跟鞋，这篇文章主要介绍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

使用

解压

复制代码

1
2
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
#压缩格式字节大小

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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
#压缩格式字节大小