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

背景信息

在开始本教程前,请确保您已完成以下操作:
  • 购买并部署云消息队列 Kafka 版实例。具体操作,请参见VPC接入
  • 下载并安装Filebeat。具体操作,请参见Download Filebeat
  • 下载并安装JDK 8。具体操作,请参见Download JDK 8

步骤一:获取接入点

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

  1. 登录云消息队列 Kafka 版控制台
  2. 概览页面的资源分布区域,选择地域。
  3. 实例列表页面,单击要作为Input接入Filebeat的实例名称。
  4. 实例详情页面的接入点信息区域,获取实例的接入点。在配置信息区域,获取用户名密码
    endpoint
    说明 不同接入点的差异,请参见接入点对比

步骤二:创建Topic

创建用于存储消息的Topic。

  1. 登录云消息队列 Kafka 版控制台
  2. 概览页面的资源分布区域,选择地域。
    重要 Topic需要在应用程序所在的地域(即所部署的ECS的所在地域)进行创建。Topic不能跨地域使用。例如Topic创建在华北2(北京)这个地域,那么消息生产端和消费端也必须运行在华北2(北京)的ECS。
  3. 实例列表页面,单击目标实例名称。
  4. 在左侧导航栏,单击Topic 管理
  5. Topic 管理页面,单击创建 Topic
  6. 创建 Topic面板,设置Topic属性,然后单击确定
    创建Topic
    参数说明示例
    名称Topic名称。demo
    描述Topic的简单描述。demo test
    分区数Topic的分区数量。12
    存储引擎
    说明 当前仅专业版实例支持选择存储引擎类型,标准版暂不支持,默认选择为云存储类型。
    Topic消息的存储引擎。

    云消息队列 Kafka 版支持以下两种存储引擎。

    • 云存储:底层接入阿里云云盘,具有低时延、高性能、持久性、高可靠等特点,采用分布式3副本机制。实例的规格类型标准版(高写版)时,存储引擎只能为云存储
    • Local 存储:使用原生Kafka的ISR复制算法,采用分布式3副本机制。
    云存储
    消息类型Topic消息的类型。
    • 普通消息:默认情况下,保证相同Key的消息分布在同一个分区中,且分区内消息按照发送顺序存储。集群中出现机器宕机时,可能会造成消息乱序。当存储引擎选择云存储时,默认选择普通消息
    • 分区顺序消息:默认情况下,保证相同Key的消息分布在同一个分区中,且分区内消息按照发送顺序存储。集群中出现机器宕机时,仍然保证分区内按照发送顺序存储。但是会出现部分分区发送消息失败,等到分区恢复后即可恢复正常。当存储引擎选择Local 存储时,默认选择分区顺序消息
    普通消息
    日志清理策略Topic日志的清理策略。

    存储引擎选择Local 存储(当前仅专业版实例支持选择存储引擎类型为Local存储,标准版暂不支持)时,需要配置日志清理策略

    云消息队列 Kafka 版支持以下两种日志清理策略。

    • Delete:默认的消息清理策略。在磁盘容量充足的情况下,保留在最长保留时间范围内的消息;在磁盘容量不足时(一般磁盘使用率超过85%视为不足),将提前删除旧消息,以保证服务可用性。
    • Compact:使用Kafka Log Compaction日志清理策略。Log Compaction清理策略保证相同Key的消息,最新的value值一定会被保留。主要适用于系统宕机后恢复状态,系统重启后重新加载缓存等场景。例如,在使用Kafka Connect或Confluent Schema Registry时,需要使用Kafka Compact Topic存储系统状态信息或配置信息。
      重要 Compact Topic一般只用在某些生态组件中,例如Kafka Connect或Confluent Schema Registry,其他情况的消息收发请勿为Topic设置该属性。具体信息,请参见云消息队列 Kafka 版Demo库
    Compact
    标签Topic的标签。demo
    创建完成后,在Topic 管理页面的列表中显示已创建的Topic。

步骤三:发送消息

向创建的Topic发送消息。

  1. 登录云消息队列 Kafka 版控制台
  2. 概览页面的资源分布区域,选择地域。
  3. 实例列表页面,单击目标实例名称。
  4. 在左侧导航栏,单击Topic 管理
  5. Topic 管理页面,找到目标Topic,在其操作列中,选择更多 > 体验发送消息
  6. 快速体验消息收发面板,发送测试消息。
    • 发送方式选择控制台
      1. 消息 Key文本框中输入消息的Key值,例如demo。
      2. 消息内容文本框输入测试的消息内容,例如 {"key": "test"}。
      3. 设置发送到指定分区,选择是否指定分区。
        • 单击,在分区 ID文本框中输入分区的ID,例如0。如果您需查询分区的ID,请参见查看分区状态
        • 单击,不指定分区。
      4. 根据界面提示信息,通过SDK订阅消息,或者执行Docker命令订阅消息。
    • 发送方式选择Docker,运行Docker容器。
      1. 执行运行 Docker 容器生产示例消息区域的Docker命令,发送消息。
      2. 执行发送后如何消费消息?区域的Docker命令,订阅消息。
    • 发送方式选择SDK,根据您的业务需求,选择需要的语言或者框架的SDK以及接入方式,通过SDK体验消息收发。

步骤四:创建Group

创建Filebeat所属的Group

  1. 登录云消息队列 Kafka 版控制台
  2. 概览页面的资源分布区域,选择地域。
  3. 实例列表页面,单击目标实例名称。
  4. 在左侧导航栏,单击Group 管理
  5. Group 管理页面,单击创建 Group
  6. 创建 Group面板的Group ID文本框输入Group的名称,在描述文本框简要描述Group,并给Group添加标签,单击确定
    创建完成后,在Group 管理页面的列表中显示已创建的Group。

步骤五:Filebeat消费消息

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

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

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

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