消息队列Kafka版可以作为Input接入Filebeat。本文说明如何在VPC环境下通过Filebeat从消息队列Kafka版消费消息。

背景信息

在开始本教程前,请确保您已完成以下操作:

步骤一:获取接入点

Filebeat通过消息队列Kafka版的接入点与消息队列Kafka版建立连接。

  1. 登录消息队列Kafka版控制台
  2. 在左侧导航栏,单击实例详情
  3. 实例详情页面,选择要作为Input接入Filebeat的实例。
  4. 基本信息区域,获取实例的接入点。
    endpoint
    说明 不同接入点的差异,请参见接入点对比

步骤二:创建Topic

创建用于存储消息的Topic。

  1. 消息队列Kafka版控制台的左侧导航栏,单击Topic管理
  2. Topic管理页面,单击创建Topic
  3. 创建Topic页面,输入Topic信息,然后单击创建
    create_topic

步骤三:发送消息

向创建的Topic发送消息。

  1. 消息队列Kafka版控制台的Topic管理页面,找到创建的Topic,在其右侧操作列,单击发送消息
  2. 发送消息对话框,输入消息信息,然后单击发送
    send_msg

步骤四:创建Consumer Group

创建Filebeat所属的Consumer Group。

  1. 消息队列Kafka版控制台的左侧导航栏,单击Consumer Group管理
  2. Consumer Group管理页面,单击创建Consumer Group
  3. 创建Consumer Group页面,输入Consumer Group信息,然后单击创建
    create_cg

步骤五:Filebeat消费消息

在安装了Filebeat的机器上启动Filebeat,从创建的Topic中消费消息。

  1. 执行cd命令切换到Filebeat的安装目录。
  2. 创建input.yml配置文件。
    1. 执行命令vim input.yml创建空的配置文件。
    2. i键进入插入模式。
    3. 输入以下内容。
      filebeat.inputs:
      - type: kafka
        hosts:
          - 192.168.XX.XX:9092
          - 192.168.XX.XX:9092
          - 192.168.XX.XX:9092
        topics: ["filebeat_test"]
        group_id: "filebeat_group"
      
      output.console:
        pretty: true
      参数 描述 示例值
      type Filebeat的Input类型。 kafka
      hosts 消息队列Kafka版提供以下VPC接入点:
      • 默认接入点
      • SASL接入点
      - 192.168.XX.XX:9092
      - 192.168.XX.XX:9092
      - 192.168.XX.XX:9092
      topics Topic的名称。 filebeat_test
      group_id Consumer Group的名称。 filebeat_group

      更多参数说明,请参见Kafka input plugin

    4. Esc键回到命令行模式。
    5. 键进入底行模式,输入wq,然后按回车键保存文件并退出。
  3. 执行以下命令消费消息。
    ./filebeat -c ./input.yml
    result