› › › › ›

配置 Kafka 输出

Kafka 输出将事件发送到 Apache Kafka。

要使用此输出，请编辑 APM Server 配置文件，通过注释掉 Elasticsearch 输出禁用它，并通过取消注释 Kafka 部分启用 Kafka 输出。

配置示例

output.kafka:
  # initial brokers for reading cluster metadata
  hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]

  # message topic selection + partitioning
  topic: '%{[fields.log_topic]}'
  partition.round_robin:
    reachable_only: false

  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000

大于 max_message_bytes 的事件将被丢弃。为避免此问题，请确保 APM Server 生成的事件不大于 max_message_bytes。

兼容性编辑

此输出适用于 0.11 到 2.2.2 之间的所有 Kafka 版本。较旧的版本可能也可以工作，但不受支持。

配置选项编辑

您可以在 apm-server.yml 配置文件的 kafka 部分中指定以下选项

enabled编辑

enabled 配置是一个布尔值设置，用于启用或禁用输出。如果设置为 false，则禁用输出。

默认值为 false。

hosts编辑

从中获取集群元数据的 Kafka 代理地址列表。集群元数据包含实际发布事件的 Kafka 代理。

version编辑

apm-server 假定运行的 Kafka 版本。默认为 1.0.0。

如果启用 0.10.0.0+ 版本，将添加事件时间戳。

有效值是 0.8.2.0 和 2.0.0 之间的所有 Kafka 版本。

有关支持的版本的信息，请参阅兼容性。

username编辑

连接到 Kafka 的用户名。如果配置了用户名，则还必须配置密码。

password编辑

连接到 Kafka 的密码。

sasl.mechanism编辑

此功能处于 beta 阶段，可能会发生变化。设计和代码不如官方 GA 功能成熟，并且按原样提供，不提供任何保证。Beta 功能不受官方 GA 功能的支持 SLA 的约束。

连接到 Kafka 时使用的 SASL 机制。它可以是以下之一

PLAIN 用于 SASL/PLAIN。
SCRAM-SHA-256 用于 SCRAM-SHA-256。
SCRAM-SHA-512 用于 SCRAM-SHA-512。

如果未设置 sasl.mechanism，如果提供了 username 和 password，则使用 PLAIN。否则，将禁用 SASL 身份验证。

topic编辑

用于生成事件的 Kafka 主题。

您可以使用格式字符串访问任何事件字段来动态设置主题。例如，此配置使用自定义字段 fields.log_topic 为每个事件设置主题

topic: '%{[fields.log_topic]}'

有关动态设置主题的其他方法，请参阅topics设置。

topics编辑

主题选择器规则的数组。每个规则都指定用于匹配规则的事件的 topic。在发布期间，APM Server 根据数组中第一个匹配的规则为每个事件设置 topic。规则可以包含条件、基于格式字符串的字段和名称映射。如果缺少 topics 设置或没有规则匹配，则使用topic字段。

规则设置

topic: 要使用的主题格式字符串。如果此字符串包含字段引用，例如 %{[fields.name]}，则字段必须存在，否则规则将失败。
mappings: 一个字典，它接受 topic 返回的值并将其映射到新名称。
default: 如果 mappings 找不到匹配项，则使用的默认字符串值。
when: 为了执行当前规则而必须成功的条件。

以下示例根据消息字段是否包含指定的字符串来设置主题

output.kafka:
  hosts: ["localhost:9092"]
  topic: "logs-%{[agent.version]}"
  topics:
    - topic: "critical-%{[agent.version]}"
      when.contains:
        message: "CRITICAL"
    - topic: "error-%{[agent.version]}"
      when.contains:
        message: "ERR"

此配置生成名为 critical-8.17.0、error-8.17.0 和 logs-8.17.0 的主题。

key编辑

可选格式化字符串，指定 Kafka 事件键。如果配置了，可以使用格式字符串从事件中提取事件键。

有关特定键选择的含义，请参阅 Kafka 文档；默认情况下，该键由 Kafka 集群选择。

partition编辑

Kafka 输出代理事件分区策略。必须是 random、round_robin 或 hash 之一。默认情况下，使用 hash 分区器。

random.group_events：设置发布到同一分区的事件数，之后分区器会随机选择新分区。默认值为 1，表示每个事件后都会随机选择一个新分区。

round_robin.group_events：设置发布到同一分区的事件数，之后分区器会选择下一个分区。默认值为 1，表示每个事件后都会选择下一个分区。

hash.hash：用于计算分区哈希值的字段列表。如果未配置字段，则将使用事件 key 值。

hash.random：如果无法计算哈希值或键值，则随机分配事件。

默认情况下，所有分区器都会尝试将事件发布到所有分区。如果节拍的某个分区的领导者变得无法访问，则输出可能会阻止。所有分区器都支持设置 reachable_only 来覆盖此行为。如果 reachable_only 设置为 true，则事件将仅发布到可用的分区。

发布到可用分区的子集可能会增加资源使用量，因为事件可能会变得不均匀分布。

client_id编辑

用于日志记录、调试和审计的可配置客户端 ID。默认为“beats”。

worker编辑

并发负载平衡 Kafka 输出工作进程的数量。

codec编辑

输出编解码器配置。如果缺少 codec 部分，则事件将以 JSON 编码。

有关更多信息，请参阅更改输出编解码器。

metadata编辑

Kafka 元数据更新设置。元数据确实包含有关用于发布的代理、主题、分区和活动领导者的信息。

refresh_frequency: 元数据刷新间隔。默认为 10 分钟。
full: 获取元数据时使用的策略，当此选项为 true 时，客户端将维护所有可用主题的完整元数据，如果此选项设置为 false，它将仅刷新配置的主题的元数据。默认为 false。
retry.max: 当集群处于领导者选举过程中时，元数据更新重试的总次数。默认为 3。
retry.backoff: 领导者选举期间重试之间的等待时间。默认为 250ms。

max_retries编辑

发布失败后重试发布事件的次数。在指定的重试次数后，通常会丢弃事件。

将 max_retries 设置为小于 0 的值以重试，直到发布所有事件。

默认值为 3。

backoff.init编辑

在发生网络错误后尝试重新发布到 Kafka 之前等待的秒数。在等待 backoff.init 秒后，APM Server 会尝试重新发布。如果尝试失败，则退避计时器将呈指数级增加到 backoff.max。成功发布后，退避计时器将重置。默认为 1s。

backoff.max编辑

在发生网络错误后尝试重新发布到 Kafka 之前等待的最大秒数。默认为 60s。

bulk_max_size编辑

单个 Kafka 请求中要批量处理的最大事件数。默认为 2048。

bulk_flush_frequency编辑

发送批量 Kafka 请求之前等待的时间。0 表示无延迟。默认为 0。

timeout编辑

等待 Kafka 代理响应的秒数（在超时之前）。默认为 30（秒）。

broker_timeout编辑

代理将等待所需 ACK 数量的最大持续时间。默认为 10s。

channel_buffer_size编辑

在输出管道中缓冲的每个 Kafka 代理的消息数。默认为 256。

keep_alive编辑

活动网络连接的保持活动周期。如果 0s，则禁用保持活动。默认为 0s。

compression编辑

设置输出压缩编解码器。必须是 none、snappy、lz4 和 gzip 之一。默认为 gzip。

Azure Event Hub for Kafka 的已知问题

在针对 Azure Event Hub for Kafka 时，将 compression 设置为 none，因为不支持提供的编解码器。

compression_level编辑

设置 gzip 使用的压缩级别。将此值设置为 0 将禁用压缩。压缩级别必须在 1（最佳速度）到 9（最佳压缩）的范围内。

增加压缩级别将减少网络使用，但会增加 CPU 使用。

默认值为 4。

max_message_bytes编辑

JSON 编码消息允许的最大大小。更大的消息将被丢弃。默认值为 1000000（字节）。此值应等于或小于代理的 message.max.bytes。

required_acks编辑

代理所需的 ACK 可靠性级别。0=无响应，1=等待本地提交，-1=等待所有副本提交。默认为 1。

注意：如果设置为 0，则 Kafka 不会返回任何 ACK。消息可能会在错误时静默丢失。

enable_krb5_fast编辑

启用 Kerberos FAST 身份验证。这可能会与某些 Active Directory 安装发生冲突。它与标准的 Kerberos 设置是分开的，因为此标志仅适用于 Kafka 输出。默认值为 false。

ssl编辑

用于配置 SSL 参数的选项，例如 Kafka 连接的根 CA。应该使用 -keyalg RSA 参数创建 Kafka 主机密钥库，以确保它使用 Filebeat 的 Kafka 库支持的密码。有关详细信息，请参阅 SSL/TLS 输出设置。

« 配置 Logstash 输出配置 Redis 输出 »