CentOS下Filebeat如何输出到Kafka

在 CentOS 下，将 Filebeat 采集的日志输出到 Kafka 是非常常见的架构（Filebeat → Kafka → Logstash/消费者）。下面给你一个完整、可落地的配置示例和步骤说明。

一、环境准备

1. 系统

• CentOS 7 / 8
• Filebeat 版本建议：7.x / 8.x（与 Kafka 兼容）

2. 已具备

• Kafka 集群已部署（假设 broker 地址：192.168.1.10:9092）
• 目标 topic 已创建（如：filebeat-logs）

二、安装 Filebeat（CentOS）

1. 使用 RPM 安装（推荐）

# 导入 Elastic GPG Key
rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

# 创建 yum 源
cat > /etc/yum.repos.d/filebeat.repo <<EOF
[elastic-8.x]
name=Elastic repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
EOF

# 安装
yum install filebeat -y

三、Filebeat 输出到 Kafka 配置

1. 配置文件路径

/etc/filebeat/filebeat.yml

2. 示例配置（核心）

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

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

# ===================== 可选 =====================
processors:
  - add_host_metadata:
      when.not.contains:
        tags: "no_host_metadata"

四、关键参数说明

五、动态 Topic（按服务/环境区分）

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

filebeat.inputs:
  - type: log
    paths:
      - /var/log/app.log
    fields:
      log_topic: app-logs
    fields_under_root: false

六、启动并验证

1. 测试配置

filebeat test config
filebeat test output

2. 启动 Filebeat

systemctl enable filebeat
systemctl start filebeat

3. 查看状态

systemctl status filebeat
journalctl -u filebeat -f

七、Kafka 侧验证

# 消费 topic 数据
kafka-console-consumer.sh \
  --bootstrap-server 192.168.1.10:9092 \
  --topic filebeat-logs \
  --from-beginning

你应该能看到 JSON 格式的日志数据。

八、常见问题

1. Kafka 连接失败

• 防火墙：9092 端口
• Kafka advertised.listeners 配置是否正确

2. Filebeat 启动报错

Exiting: error loading config

→ YAML 缩进错误（不能用 Tab）

3. 消息过大

max_message_bytes: 2097152

九、典型架构示意

Filebeat → Kafka → Logstash → Elasticsearch
                ↓
              Flink / 自研消费

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

• ✅ 针对 Kafka 3.x / 2.x 给出最佳参数
• ✅ 配置 多 topic / 多 input
• ✅ 配合 ELK / 容器 / Kubernetes

直接告诉我你的 Kafka 版本和使用场景即可。