1filebeat.inputs:
2
3- input_type: log 
4
5# 此处为监听文件路径
6  paths:
7    - /var/log/*.log
8#------------------------------- Kafka output ----------------------------------
9output.kafka:
10  # 是否启用
11  enabled: true
12 
13  # The list of Kafka broker addresses from which to fetch the cluster metadata.
14  # The cluster metadata contain the actual Kafka brokers events are published
15  # to
16  # 获取集群元数据的 Kafka 代理地址列表（接入点）
17  hosts: ["x.x.x.x:9095","x.x.x.x:9095","x.x.x.x:9095"]
18 
19  # 用于生成事件的 Kafka 主题, 可以使用任何事件 field 的格式字符串, 若要从文档类型设置主题，请使用 `%{[type]}`.
20  topic: test
21  
22  # 设置sasl协议字段
23  sasl.mechanism: "SCRAM-SHA-512" 
24 
25 # 设置 SSL协议字段
26 #SSL/SASL_SSL方式请下载证书文件： kafka-key.zip
27 #ssl.certificate_authorities:[/***/client_ssl.properties]
28  partition.random:
29    # If enabled, events will only be published to partitions with reachable leaders. Default is false.
30    # reachable_only 设置为true，则事件将仅发布到可用的分区
31    # 必须是 random, round_robin, hash 三种的一种
32    # 默认为 false
33    reachable_only: false
34 
35  # Authentication details. Password is required if username is set.
36  # 身份验证的细节。如果设置了用户名，则需要密码。
37  # 没有设置可保持空值
38  username: ''
39  password: ''
40 
41  # Sets the output compression codec. Must be one of none, snappy and gzip. The default is gzip.
42  # 设置输出压缩编解码器
43  # 必须是 none, snappy 和 gzip 中的一个
44  # 默认是 gzip
45  compression: none
46 
47  # Set the compression level. Currently only gzip provides a compression level between 0 and 9. The default value is chosen by the compression algorithm.
48  # 设置压缩级别
49  # 目前只有 gzip 提供 0 到 9 之间的压缩级别
50  # 默认值由压缩算法选择
51  compression_level: 4
52 
53  # The maximum permitted size of JSON-encoded messages. Bigger messages will be dropped. The default value is 1000000 (bytes). This value should be equal to r less than the broker's message.max.bytes.
54  # json编码消息的最大允许大小, 更大的消息将被删除。默认值是 1000000 ( 字节 ) 。这个值应该等于 r，小于 broker 的 message.max.bytes
55  max_message_bytes: 1000000
56  # The ACK reliability level required from broker. 0=no response, 1=wait for local commit, -1=wait for all replicas to commit. The default is 1.  Note: If set to 0, no ACKs are returned by Kafka. Messages might be lost silently on error.
57  # 代理要求的ACK可靠性级别
58  # 0=无响应，1=等待本地提交，-1=等待所有副本提交
59  # 默认值是1
60  # 注意:如果设置为0,Kafka不会返回任何ack。出错时，消息可能会悄无声息地丢失。
61  required_acks: 1
62  # The configurable ClientID used for logging, debugging, and auditing purposes.  The default is "beats".
63  # 可配置的ClientID，用于日志记录、调试和审计。默认是“beats”。
64  client_id: beats

1./filebeat -e -c filebeat.yml 

1filebeat.inputs:
2- type: kafka
3  hosts:
4    - ip:端口
5    - ip:端口
6    - ip:端口
7    
8  # 身份验证的细节。如果设置了用户名，则需要密码。
9  # 没有设置可保持空值
10  username: ""
11  password: ""
12  #Topic的名称。
13  topics: ["filebeat_test"]
14  #Group的名称。
15  group_id: "filebeat_group"
16  #证书所在位置
17 #SSL/SASL_SSL方式请下载证书文件： kafka-key.zip
18 #ssl.certificate_authorities:[/***/client_ssl.properties]
19
20  ssl.verification_mode: none
21
22output.console:
23  pretty: true

1./filebeat -c ./input.yml