centos filebeat如何转kafka

在 CentOS 上把 Filebeat 采集的日志发送到 Kafka，一般步骤如下（以 CentOS 7/8 + Filebeat 7.x/8.x 为例）：

一、安装 Filebeat（CentOS）

1. 使用 RPM 安装（推荐）

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

# 添加 yum 源（以 8.x 为例）
sudo tee /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

sudo yum install filebeat -y

二、配置 Filebeat 输出到 Kafka

1. 编辑配置文件

vim /etc/filebeat/filebeat.yml

2. 示例配置（核心部分）

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/*.log
      # 可添加多个路径
      # - /opt/app/logs/*.log

output.kafka:
  # Kafka 集群地址
  hosts: ["192.168.1.10:9092", "192.168.1.11:9092"]
  
  # Kafka topic（可写死或用变量）
  topic: filebeat-logs

  # 可选
  partition.round_robin:
    reachable_only: true

  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000

✅ 说明

• hosts：Kafka broker 地址
• topic：Kafka 主题（需提前创建或允许自动创建）
• required_acks：0 / 1 / -1（生产建议 1 或 -1）

三、Kafka 准备（可选但推荐）

在 Kafka 服务器上创建 topic：

bin/kafka-topics.sh \
  --create \
  --topic filebeat-logs \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1

如果 Kafka 开启了 自动创建 topic，可跳过。

四、启动 Filebeat

# 测试配置
filebeat test config
filebeat test output

# 启动
systemctl enable filebeat
systemctl start filebeat

# 查看状态
systemctl status filebeat

五、验证是否成功

1. 消费 Kafka 消息

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

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

六、常见问题

1. Kafka 连不上

• 防火墙：firewall-cmd --add-port=9092/tcp --permanent
• Kafka listeners 配置是否对外暴露

2. 权限问题

• Filebeat 用户是否有权限读日志文件

ls -l /var/log/xxx.log

3. 使用多个 topic（按字段）

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

并在 input 中：

fields:
  log_topic: app-log

如果你有：

• Kafka 版本
• Filebeat 版本
• 日志路径 / 格式
• 是否用 SASL / SSL

我可以直接给你一份可用配置模板。