Filebeat安装

安装Filebeat:

在test14服务器上安装Filebeat，如下：

[root@test14 ~]# tar -zxf filebeat-7.3.1-linux-x86_64.tar.gz -C /usr/local/
[root@test14 ~]# ln -s /usr/local/filebeat-7.3.1-linux-x86_64/ /usr/local/filebeat
[root@test14 ~]# cd /usr/local/filebeat

#----------------------添加hosts解析----------------
[root@test14 filebeat]# cat /etc/hosts
192.168.37.11 test11
192.168.37.12 test12
192.168.37.13 test13

配置前的思考？

• Filebeat 如何读取多个日志目录？
• Filebeat 如何区分不同日志来源？

配置Filebeat：

先启动zk和kafka，前置环境已配置好，下例是抓取nginx的日志文件，如下：

[root@test14 filebeat]# vim filebeat.yml 
#=========================== Filebeat prospectors =============================
filebeat.inputs:

- type: log                     #指定数据的输入类型，默认为log，另外还可以指定stdin
  paths:                        #配置多个paths,Filebeat可以实现读取多个目录
    - /app/logs/*/access.log    #指定日志位置
  exclude_files: ['/app/logs/ad-auth/*']
  fields:                       #定义域来标识不同的 log，实现不同日志来源的区分
    log_topic: nginx_access_log
    document_type: nginx
- type: log
  paths:
    - /app/logs/*/error.log
  exclude_files: ['/app/logs/ad-auth/*']
  fields:
    log_topic: nginx_error_log
    document_type: nginx
#================= Preocessors============================================
processors:                    #此配置是将日志输出格式过滤掉，一般情况下，一些无用的日志字段我们可以删除，只看关键性的信息
  - drop_fields:
      fields: ["beat","input.type"]
#----------------------------- Kafka output --------------------------------
output.kafka:
  # initial brokers for reading cluster metadata
  hosts: ["test11:9092", "test12:9092", "test13:9092"]
#注意：这里需要在/etc/hosts添加ip到主机的解析，直接写ip:端口后面filebeat一样要跳转到主机:端口的，hosts没解析的日志数据推不到kafka上去。

  # message topic selection + partitioning
  topic: '%{[fields.log_topic]}'#指定要发送数据到kafka集群的哪个topic，与上述的"fields: log_topic："相对应，也可以是'%{[filelds][log_topic]}'的写法
  partition.round_robin:        #开启kafka的partition分区，分区策略有 random/round_robin/hash
    reachable_only: false       #是否只发往可达分区
  version: 2.0.0                #kafka版本，但目前filebeat7.3版本中有效值输入0.8.2.0和2.0.0，我的kafka版本是2.12的，设置成2.0.0或这个有效值内的数值也可以的
  required_acks: 1              #要求的ACK可靠性等级。0 =无响应，1 =等待本地提交，-1 =等待所有副本提交。默认值为1
  compression: none             #设置输出压缩编解码器。必须是一个none，snappy，lz4和gzip。默认是gzip
  max_message_bytes: 1000000    #JSON编码消息的最大允许大小。更大的消息将被删除。默认值为1000000（字节）。该值应等于或小于经纪人的值message.max.bytes

连接到kafka查看有没有数据传进来了

[root@test13 ~]# cd /usr/local/kafka_2.12-2.3.0/bin
[root@test13 ~]# ./kafka-console-consumer.sh --bootstrap-server 192.168.37.12:9092 --topic nginx_access_log

注：Filebeat官网参考文档：https://www.elastic.co/guide/en/beats/filebeat/current/kafka-output.html