如何通过Docker Compose配置Filebeat实现Kafka到Elasticsearch消息传输？

你遇到的核心问题是Filebeat配置存在多处细节错误，导致其无法正确连接Kafka并消费消息，以下是具体问题点和修正方案：

一、关键错误点诊断 #

1. 配置文件未正确挂载：Docker Compose中Filebeat的volumes仅指定本地文件，未映射到容器内默认配置路径，导致Filebeat加载内置默认配置而非自定义配置。
2. Kafka连接端口错误：容器间通信需使用Kafka服务的内部端口9092，而非映射到宿主机的29092。
3. Elasticsearch地址错误：在Filebeat容器内，localhost指向容器自身，需使用Docker Compose服务名elasticsearch访问ES服务。
4. YAML格式错误：filebeat.inputs前存在多余缩进，导致配置解析失败。
5. 输入未显式启用：Kafka输入未设置enabled: true，默认处于禁用状态。

二、修正后的完整配置 #

1. docker-compose.yml 修正版 #

version: '3'
services:
  PostgreSQL:
    image: postgres:latest
    environment:
      - POSTGRES_DB=distillery
      - POSTGRES_USER=admin
      - POSTGRES_PASSWORD=secret
    ports:
      - "5434:5432"

  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "22181:2181"

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - "29092:29092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_LOG_RETENTION_MS: 10000
      KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: 5000

  elasticsearch:
    image: elasticsearch:7.9.2
    depends_on:
      - kafka
    ports:
      - '9200:9200'
    environment:
      - discovery.type=single-node
    limits:
      memlock:
        soft: -1
        hard: -1

  filebeat:
    depends_on:
      - kafka
      - elasticsearch
    image: docker.elastic.co/beats/filebeat:7.9.2
    container_name: filebeat
    user: root  # 避免权限问题导致配置读取失败
    volumes:
      - "./filebeat.yml:/usr/share/filebeat/filebeat.yml:ro"  # 正确映射配置文件到容器内路径
    command: ["filebeat", "-e", "-strict.perms=false"]  # 强制启用配置，关闭权限严格检查

2. filebeat.yml 修正版 #

filebeat.inputs:
  - type: kafka
    enabled: true  # 显式启用该输入
    hosts: ["kafka:9092"]  # 使用Kafka服务的内部端口
    topics: ["progress-raspberry"]
    client_id: "filebeat"
    group_id: "filebeat"

output.elasticsearch:
  hosts: ["elasticsearch:9200"]  # 使用Docker服务名访问ES
  username: "elastic"  # 7.x版本默认用户名（未修改则保留）
  password: "changeme"  # 7.x版本默认密码（需根据实际情况调整）

三、验证步骤 #

1. 启动所有服务：

   docker-compose up -d
   

2. 查看Filebeat日志，确认连接状态：

   docker logs -f filebeat
   

   日志中应出现Connected to Kafka broker和Connected to Elasticsearch的成功提示。
3. 检查Kafka消费者列表：
   进入Kafka容器执行命令查看消费者组：

   docker exec -it $(docker ps -qf "name=kafka") kafka-consumer-groups --bootstrap-server kafka:9092 --list
   

   应能看到filebeat消费者组。
4. 测试消息消费：
   向progress-raspberry主题发送测试消息，随后通过ES API检查数据：

   curl http://localhost:9200/_cat/indices?v
   

   应能看到Filebeat自动创建的索引，进一步查询索引内容可确认消息已同步。

内容的提问来源于stack exchange，提问作者Martin Dvoracek