AI 助理
备案控制台
文档
输入文档关键字查找
首页 云消息队列 Kafka 版 云消息队列 Kafka 版 实践教程 生态对接 开源生态 Filebeat 公网 作为Output接入

作为Output接入

更新时间：
产品详情
相关技术圈
我的收藏

云消息队列 Kafka 版可以作为Output接入Filebeat。本文说明如何在公网环境下通过Filebeat向云消息队列 Kafka 版发送消息。

前提条件

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

步骤一：获取接入点与用户名密码

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

  1. 登录云消息队列 Kafka 版控制台。

  2. 在概览页面的资源分布区域，选择地域。

  3. 在实例列表页面，单击作为Output接入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名称。

    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。

步骤三：Filebeat发送消息

在安装了Filebeat的机器上启动Filebeat，向创建的Topic发送消息。

  1. 执行cd命令切换到Filebeat的安装目录。

  2. 执行以下命令下载CA证书文件。

    wget https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20220826/ytsw/only-4096-ca-cert

  3. 创建output.conf配置文件。

    1. 执行命令vim output.conf创建空的配置文件。

    2. 按i键进入插入模式。

    3. 输入以下内容。

      filebeat.inputs:
- type: stdin

output.kafka:
  hosts: ["alikafka-pre-cn-zv**********-1.alikafka.aliyuncs.com:9093", "alikafka-pre-cn-zv**********-2.alikafka.aliyuncs.com:9093", "alikafka-pre-cn-zv**********-3.alikafka.aliyuncs.com:9093"]
  username: "alikafka_pre-cn-v641e1d***"
  password: "aeN3WLRoMPRXmAP2jvJuGk84Kuuo***"

  topic: 'filebeat_test'
  partition.round_robin:
    reachable_only: false
  ssl.certificate_authorities: ["/home/admin/filebeat/filebeat-7.7.0-linux-x86_64/tasks/vpc_ssl/ca-cert"]
  ssl.verification_mode: none

  required_acks: 1
  compression: none
  max_message_bytes: 1000000

      参数

      描述

      示例值

      hosts

      云消息队列 Kafka 版提供的公网接入点为SSL接入点。

      alikafka-pre-cn-zv**********-1.alikafka.aliyuncs.com:9093，alikafka-pre-cn-zv**********-2.alikafka.aliyuncs.com:9093，alikafka-pre-cn-zv**********-3.alikafka.aliyuncs.com:9093

      username

      公网/VPC实例的用户名。

      alikafka_pre-cn-v641e1d***

      password

      公网/VPC实例的密码。

      aeN3WLRoMPRXmAP2jvJuGk84Kuuo***

      topic

      Topic的名称。

      filebeat_test

      reachable_only

      消息是否只发送到可用的分区。取值：

      • true：如果主分区不可用，输出可能阻塞。

      • false：即使主分区不可用，输出不被阻塞。

      false

      ssl.certificate_authorities

      CA证书所在位置。

      /home/admin/filebeat/filebeat-7.7.0-linux-x86_64/ca-cert

      ssl.verification_mode

      认证模式。

      none

      required_acks

      ACK可靠性。取值：

      • 0：无响应

      • 1：等待本地提交

      • -1：等待所有副本提交

      默认值为1。

      1

      compression

      数据压缩编码器。默认值为gzip。取值：

      • none：无

      • snappy：用来压缩和解压缩的C++开发包

      • lz4：着重于压缩和解压缩速度的无损数据压缩算法

      • gzip：GNU自由软件的文件压缩程序

      none

      max_message_bytes

      最大消息大小。单位为字节。默认值为1000000。该值应小于您配置的云消息队列 Kafka 版最大消息大小。

      1000000

      更多参数说明，请参见Kafka output plugin。

    4. 按Esc键回到命令行模式。

    5. 按：键进入底行模式，输入wq，然后按回车键保存文件并退出。

  4. 向创建的Topic发送消息。

    1. 执行./filebeat -c ./output.yml。

    2. 输入test，然后按回车键。

步骤四：查看Topic分区

查看消息发送到Topic的情况。

  1. 登录云消息队列 Kafka 版控制台。

  2. 在概览页面的资源分布区域，选择地域。

  3. 在实例列表页面，单击目标实例名称。

  4. 在左侧导航栏，单击Topic 管理。

  5. 在Topic 管理页面，找到目标Topic，在其操作列中，选择更多 > 分区状态。

    表 1. 分区状态信息

    参数

    说明

    分区ID

    该Topic分区的ID号。

    最小位点

    该Topic在当前分区下的最小消费位点。

    最大位点

    该Topic在当前分区下的最大消费位点。

    分区消息量

    该Topic在当前分区下的消息总量。

    最近更新时间

    本分区中最近一条消息的存储时间。

    分区状态信息

步骤五：按位点查询消息

您可以根据发送的消息的分区ID和位点信息查询该消息。

  1. 登录云消息队列 Kafka 版控制台。

  2. 在概览页面的资源分布区域，选择地域。

  3. 在实例列表页面，单击目标实例名称。

  4. 在左侧导航栏，单击消息查询。

  5. 在消息查询页面的查询方式列表中，选择按位点查询。

  6. 在Topic列表中，选择消息所属Topic名称；在分区列表中，选择消息所属的分区；在起始位点文本框，输入消息所在分区的位点，然后单击查询。

    展示该查询位点及以后连续的消息。例如，指定的分区和位点都为“5”，那么返回的结果从位点“5”开始。

    表 2. 查询结果参数解释

    参数

    描述

    分区

    消息的Topic分区。

    位点

    消息的所在的位点。

    Key

    消息的键（已强制转化为String类型）。

    Value

    消息的值（已强制转化为String类型），即消息的具体内容。

    消息创建时间

    发送消息时，客户端自带的或是您指定的ProducerRecord中的消息创建时间。

    说明

    • 如果配置了该字段，则按配置值显示。

    • 如果未配置该字段，则默认取消息发送时的系统时间。

    • 如果显示值为1970/x/x x:x:x，则说明发送时间配置为0或其他有误的值。

    • 0.9及以前版本的云消息队列 Kafka 版客户端不支持配置该时间。

    操作

    • 单击下载 Key：下载消息的键值。

    • 单击下载 Value：下载消息的具体内容。

    重要

    • 查询到的每条消息在控制台上最多显示1 KB的内容，超过1 KB的部分将自动截断。如需查看完整的消息内容，请下载相应的消息。

    • 下载的消息最大为10 MB。如果消息超过10 MB，则只下载10 MB的内容。

上一篇：作为Input接入下一篇：Filebeat发送失败问题
文档推荐