CentOS Filebeat如何输出到Kafka

在 CentOS 上把 Filebeat 的日志输出到 Kafka，核心就是修改 filebeat.yml，把 output.elasticsearch 换成 output.kafka。下面是一个完整、可直接用的配置示例和步骤说明。

一、环境准备

1. 已安装 Filebeat

filebeat version

如果未安装：

sudo yum install filebeat -y

2. Kafka 已运行

确保 Kafka 可访问，例如：

kafka1:9092
kafka2:9092
kafka3:9092

二、Filebeat 输出到 Kafka 配置

1. 编辑配置文件

sudo vi /etc/filebeat/filebeat.yml

2. 核心配置示例（✅ 推荐）

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/*.log
      - /opt/app/logs/*.log

# ===================== Kafka Output =====================
output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]
  topic: "filebeat-logs"
  partition.round_robin:
    reachable_only: true
  required_acks: 1
  compression: gzip
  max_message_bytes: 1048576

# ===================== 可选 =====================
logging.level: info
logging.to_files: true

三、关键参数说明

四、动态 Topic（按字段路由）

示例：按 fields.log_topic 发送到不同 topic

filebeat.inputs:
  - type: log
    paths: ["/var/log/app.log"]
    fields:
      log_topic: app-log
    fields_under_root: false

output.kafka:
  hosts: ["kafka1:9092"]
  topic: '%{[fields.log_topic]}'

五、启动并验证

1. 测试配置

filebeat test config
filebeat test output

2. 启动 Filebeat

sudo systemctl restart filebeat
sudo systemctl enable filebeat

六、Kafka 验证是否收到数据

kafka-console-consumer.sh \
  --bootstrap-server kafka1:9092 \
  --topic filebeat-logs \
  --from-beginning

如果看到 JSON 日志，说明成功 ✅

七、常见问题

1. Kafka 连不上

• 检查防火墙
• 检查 listeners / advertised.listeners
• 确认端口 9092

2. Filebeat 报 JSON 序列化错误

• 不要同时配置 output.elasticsearch 和 output.kafka

3. 中文乱码

Filebeat 默认 UTF-8，Kafka 消费者也需 UTF-8

八、典型架构

Filebeat → Kafka → Logstash / Flink → ES / ClickHouse

如果你愿意，我可以帮你：

• ✅ 按 多 topic / 多业务 拆分配置
• ✅ 对接 Logstash / Flink
• ✅ 配置 Kafka SASL / SSL
• ✅ 生产环境调优参数

直接告诉我你的 Kafka 版本和日志来源即可。