filebeat 7.10.0

现在filebeat.yml文件中只会有一个outputs生效，因此不能配置多个outputs发送到不同的地方，但是可以通过’%{}'引用参数的方法动态的设置发送目标。

通过fields参数设置kafka的topic，如下配置文件所示：

# ============================== Filebeat inputs ===============================
filebeat.inputs:
# Each - is an input. Most options can be set at the input level, so
# you can use different inputs for various configurations.
# Below are the input specific configurations.
# ======== 发送到first topic========
- type: log
  enabled: true
  paths:
    - e:\Lab\Logs\new.log
  fields:
    kafka_topic: first
# ======== 发送到second topic========
- type: log
  enabled: true
  paths:
    - e:\Lab\Logs\new1.log
  fields:
    kafka_topic: second
# ---------------------------- kafka Output----------------------------
output.kafka:
  # initial brokers for reading cluster metadata hosts是string数组类型，支持多个kafka实例地址(此处不是配置zookeeper地址)
  hosts: ["139.XXXXX:9092"]
  # 在这里通过引用fields来指定目标topic
  topic: '%{[fields.kafka_topic]}'
  version: 0.11.0.0
  compression: gzip
  max_message_bytes: 1000000
  codec.format:
    string: '%{[message]}'
                    版本filebeat 7.10.0方法通过fields参数设置kafka的topic，如下配置文件所示：# ============================== Filebeat inputs ===============================filebeat.inputs:# Each - is an input. Most options can be set at the input level, so# you can use different inputs f
				
logkafka-收集日志并将行发送到Apache Kafka 0.8+
logkafka逐行将日志文件内容发送到kafka 0.8。 它将一行文件视为一条kafka消息。
 如果您想将其部署在生产环境中，请参阅。
Zookeeper进行日志收集配置管理
[带有时间格式的日志路径（按时间顺序收集文件）]（docs / Features.md＃日志路径模式）
 日志文件旋转
批处理消息
压缩（无，gzip，snappy）
 [邮件正则表达式过滤器]（docs / Features.md＃Regex过滤器）
 [用户定义的行定界符]（docs / Features.md＃Line定界符）
与其他日志聚合和监视工具的差异
与水槽，流利的，原木的主要区别是
日志收集配置和状态的管理：
 flume，fluentd，logstash将日志文件配置和状态保留在本地：启动本地服务器以管理日志
				
系统记录器
Syslogger是用于将Rsyslog消息转发到。
 是一个“高性能，分布式消息传递系统”，非常适合于业务和系统事件数据的整理。 有关更多信息，请参阅Jay Kreps精彩的 。
 Syslogger将帮助您将syslog消息转发到Kafka。 消息从转发通过TCP连接到syslogger。 Rsyslog已经有很多东西可以使转发消息尽可能可靠，处理背压以及将排队的消息写入磁盘等。有关更多信息，请参见 。
Syslogger试图通过减轻Rsyslog的故障处理责任来成为Rsyslog的良好公民。
 通过向Kafka同步发送消息来实现可靠性（在仅使用TCP的情况下尽可能多）：如果在将消息转发到Kafka时出现问题或延迟，我们将对Rsyslog施加尽可能多的背压。
 默认情况下，Syslogger在端口1514上启动一个TCP侦听器。它还尝试连接到ZooKeeper，以检索
				
目录1.环境准备：2.ElasticSearch3.Logstash5.FilebeatFilebeat其他配置filebeat完整配置：多行合并日志字段添加压缩filebeat运行日志限制内存发向多个logstashfilebeat->>logstash通过ssl加密Ssl测试断网测试注册服务filebeat拆分日志踩坑：提醒：
1.环境准备：
1.1测试机PC1
IP:192.168.1.99
服务器环境：Linux Centos 6.10
Java环境：java 1.7
安装：filebe
				
平时在物理机上使用 Filebeat 收集日志并输出到 Kafka 中时，会编写多个 filebeat 配置文件然后启动多个 filebeat 进程来收集不同路径下的日志并推送到不同的 Topic。那么如果将所有的日志路径都写到一个 filebeat 配置文件中，那么就需要根据不同的日志来设置 Topic 了。
其实 logstash 也可以实现这个功能。但是此处只演示在 Filebeat 上实现。步骤和讲解如下：
例如现在有如下三个日志文件，需要输出到不同的 Topic：
access.log  --
				
主题的管理包括创建主题、查看主题信息、修改主题和删除主题等操作。可以通过Kafka提供的kafka-topics.sh 脚本来执行这些操作，这个脚本位于$KAFKA_HOME/bin/ 目录下，其核心代码仅有一行，具体如下：
exec $(dirname $0)/kafka-run - class.sh kafka.admin.Top ic Command ＂$＠＂
Topic相关
auto .create.topics .enable 设置为true会自动创建主题，和...
				
关注 iteblog_hadoop 公众号并在评论区留言（认真写评论，增加上榜的机会）。留言点赞数排名前5名的粉丝，各免费赠送一本《大数据之路：阿里巴巴大数据实践》，明天18点就结束了，赶紧去参加吧。
如下面图片不清晰，请访问https://www.iteblog.com/archives/2219.html，或点击下面阅读原文进行阅读。
熟悉 Kafka的同学肯定知道，每个主题有多个
				
结合官方文档,进行配置说明注释,不记得可以来此看看,因为最近在别的领域突破就很少来写博文了....emmm那就直接上干货
FileBeat主要用于监控文件目录,进行各项操作输出结合ELK使用的组件,例如,文件型日志收集,格式化等..
基本使用的就是这些配置:
# Ps : 使用时 去掉 "#"
# 输入源
filebeat.inputs:
- type: log
  enabled: true
  # 日志路径下面可以多个 注意 "-"符号
  paths:
    - /xxxxxxxx/logs
				
根据引用\[1\]和引用\[3\]的内容，要将Filebeat的输出配置为输出到Kafka，需要进行以下步骤：
1. 在Filebeat的配置文件中，将output.elasticsearch和output.logstash的配置都注释掉，以禁用这两个输出配置。
2. 增加Kafka集群的输出配置，配置如下图所示。可以指定Kafka集群的地址和端口。
3. 启动Filebeat，它将开始将读取到的日志输出到Kafka集群中。
这样，Filebeat就会将收集到的日志发送到Kafka消息队列中。其他应用程序可以通过消费Kafka中的消息来获取Filebeat发送的日志信息。可以使用命令`./kafka-console-consumer.sh --bootstrap-server 192.168.192.132:9092 --topic test --from-beginning`来启动一个消费者，查看Filebeat发送过来的消息。如果能够看到消息，说明Filebeat的output.kafka配置成功。
#### 引用[.reference_title]
-  *1* *2* *3* [Filebeat部署+Kafka接收消息](https://blog.csdn.net/qq_43499416/article/details/126402809)[target="_blank" data-report-click={"spm":"1018.2226.3001.9630","extra":{"utm_source":"vip_chatgpt_common_search_pc_result","utm_medium":"distribute.pc_search_result.none-task-cask-2~all~insert_cask~default-1-null.142^v91^control,239^v3^insert_chatgpt"}} ] [.reference_item]
[ .reference_list ]