› › ›

Kafka 输出

Kafka 输出编辑

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

兼容性： 此输出可以连接到 Kafka 0.8.2.0 及更高版本。旧版本也可能工作，但不支持。

此示例在 Elastic Agent elastic-agent.yml 文件中配置了一个名为 kafka-output 的 Kafka 输出，其设置将在

outputs:
  kafka-output:
    type: kafka
    hosts:
      - 'kafka1:9092'
      - 'kafka2:9092'
      - 'kafka3:9092'
    client_id: Elastic
    version: 1.0.0
    compression: gzip
    compression_level: 4
    username: <my-kafka-username>
    password: <my-kakfa-password>
    sasl:
      mechanism: SCRAM-SHA-256
    partition:
      round_robin:
        group_events: 1
    topic: 'elastic-agent'
    headers: []
    timeout: 30
    broker_timeout: 30
    required_acks: 1
    ssl:
      verification_mode: full

Kafka 输出配置设置编辑

中进一步描述。 kafka 输出支持以下设置，按类别分组。其中许多设置都有合理的默认值，允许您以最少的配置运行 Elastic Agent。

设置描述

设置	描述
`enabled`	（布尔值）启用或禁用输出。如果设置为 `false`，则禁用输出。
`hosts`	您的 Elastic Agent 将用于连接到一个或多个 Kafka 代理的地址。以下是定义了三个主机的示例 `hosts` 设置 hosts: - 'localhost:9092' - 'mykafkahost01:9092' - 'mykafkahost02:9092'
`version`	Elastic Agent 连接时将请求的 Kafka 协议版本。默认为 1.0.0。协议版本控制 Elastic Agent 可用的 Kafka 客户端功能；它不会阻止 Elastic Agent 连接到比协议版本更新的 Kafka 版本。

enabled

（布尔值）启用或禁用输出。如果设置为 false，则禁用输出。

hosts

您的 Elastic Agent 将用于连接到一个或多个 Kafka 代理的地址。

以下是定义了三个主机的示例 hosts 设置

    hosts:
      - 'localhost:9092'
      - 'mykafkahost01:9092'
      - 'mykafkahost02:9092'

version

Elastic Agent 连接时将请求的 Kafka 协议版本。默认为 1.0.0。

协议版本控制 Elastic Agent 可用的 Kafka 客户端功能；它不会阻止 Elastic Agent 连接到比协议版本更新的 Kafka 版本。

身份验证设置编辑

设置	描述
`username`	用于连接到 Kafka 的用户名。如果配置了用户名，则还必须配置密码。
`password`	用于连接到 Kafka 的密码。
`sasl.mechanism`	连接到 Kafka 时使用的 SASL 机制。它可以是以下之一 `PLAIN` 用于 SASL/PLAIN。 `SCRAM-SHA-256` 用于 SCRAM-SHA-256。 `SCRAM-SHA-512` 用于 SCRAM-SHA-512。如果未设置 `sasl.mechanism`，则在提供 `username` 和 `password` 时使用 `PLAIN`。否则，将禁用 SASL 身份验证。
`ssl`	通过 `kafka` 输出将数据发送到安全集群时，Elastic Agent 可以使用 SSL/TLS。有关可用设置的列表，请参阅 SSL/TLS，特别是表 4，“常见配置选项” 和表 5，“客户端配置选项” 下的设置。

内存队列设置编辑

内存队列将所有事件保存在内存中。

内存队列等待输出确认或丢弃事件。如果队列已满，则无法将新事件插入内存队列。只有在收到来自输出的信号后，队列才会释放空间以接受更多事件。

内存队列由参数 flush.min_events 和 flush.timeout 控制。 flush.min_events 对单个批次中可以包含的事件数量进行了限制，而 flush.timeout 指定队列应等待多长时间才能完全填充事件请求。如果输出支持 bulk_max_size 参数，则最大批次大小将是 bulk_max_size 和 flush.min_events 中的较小者。

flush.min_events 是一个遗留参数，新配置应优先使用 bulk_max_size 控制批次大小。从 8.13 版本开始，使用 flush.min_events 而不是 bulk_max_size 限制批次大小永远不会带来性能优势。

在同步模式下，只要有可用事件，就会立即填充事件请求，即使没有足够的事件来填充请求的批次也是如此。这在必须最小化延迟时非常有用。要使用同步模式，请将 flush.timeout 设置为 0。

为了向后兼容，还可以通过将 flush.min_events 设置为 0 或 1 来激活同步模式。在这种情况下，批次大小将限制为队列容量的 1/2。

在异步模式下，事件请求将等待指定的超时时间，以尝试完全填充请求的批次。如果超时到期，队列将返回包含所有可用事件的部分批次。要使用异步模式，请将 flush.timeout 设置为正持续时间，例如 5 秒。

此示例配置在有足够事件填充输出的请求（通常由 bulk_max_size 控制，并由 flush.min_events 限制为最多 512 个事件）时，或事件已等待

  queue.mem.events: 4096
  queue.mem.flush.min_events: 512
  queue.mem.flush.timeout: 5s

设置描述

设置	描述
`queue.mem.events`	队列可以存储的事件数。此值应可以被 `queue.mem.flush.min_events` 或 `bulk_max_size` 中的较小者整除，以避免向输出发送部分批次。默认值： `3200 个事件`
`queue.mem.flush.min_events`	`flush.min_events` 是一个遗留参数，新配置应优先使用 `bulk_max_size` 控制批次大小。从 8.13 版本开始，使用 `flush.min_events` 而不是 `bulk_max_size` 限制批次大小永远不会带来性能优势。默认值： `1600 个事件`
`queue.mem.flush.timeout`	（整数）满足 `queue.mem.flush.min_events` 的最大等待时间。如果设置为 0 秒，则事件立即可供输出使用。默认值： `10 秒`

queue.mem.events

队列可以存储的事件数。此值应可以被 queue.mem.flush.min_events 或 bulk_max_size 中的较小者整除，以避免向输出发送部分批次。

默认值： 3200 个事件

queue.mem.flush.min_events

默认值： 1600 个事件

queue.mem.flush.timeout

（整数）满足 queue.mem.flush.min_events 的最大等待时间。如果设置为 0 秒，则事件立即可供输出使用。

默认值： 10 秒

主题设置编辑

使用这些选项为每个 Elastic Agent 事件设置 Kafka 主题。

设置	描述
`topic`	用于生成事件的默认 Kafka 主题。

分区设置编辑

创建的分区数量由 Kafka 代理根据主题列表自动设置。然后，记录将随机、轮询或根据计算的哈希值发布到分区。

在以下示例中，在将每个事件发布到分区后，分区器将以轮询方式选择下一个分区。

    partition:
      round_robin:
        group_events: 1

设置	描述
`random.group_events`	设置要发布到同一分区的事件数量，然后分区器随机选择一个新分区。默认值为 1，表示在每个事件之后随机选择一个新分区。
`round_robin.group_events`	设置要发布到同一分区的事件数量，然后分区器选择下一个分区。默认值为 1，表示在每个事件之后将选择下一个分区。
`hash.hash`	用于计算分区哈希值的字段列表。如果未配置字段，则将使用事件键值。
`hash.random`	如果无法计算哈希或键值，则随机分配事件。

标头设置编辑

标头是一个键值对，并且可以包含多个具有相同键的标头。仅支持字符串值。这些标头将包含在每个生成的 Kafka 消息中。

设置	描述
`key`	要在 Kafka 标头中设置的键。
`value`	要在 Kafka 标头中设置的值。
`client_id`	用于日志记录、调试和审计目的的可配置 ClientID。默认值为 `Elastic`。客户端 ID 是协议的一部分，用于标识消息的来源。

其他配置设置编辑

您可以在代理配置文件的 kafka-output 部分中指定这些其他选项。

设置	描述
`backoff.init`	（字符串）网络错误后尝试重新连接到 Kafka 之前的等待秒数。等待 `backoff.init` 秒后，Elastic Agent 尝试重新连接。如果尝试失败，则退避计时器会呈指数级增加，直至达到 `backoff.max`。成功连接后，退避计时器将重置。默认值： `1 秒`
`backoff.max`	（字符串）网络错误后尝试连接到 Kafka 之前的最大等待秒数。默认值： `60 秒`
`broker_timeout`	Kafka 代理在超时之前等待所需 ACK 数量的最长时间（请参阅后面的 `required_acks` 设置）。默认值： `30`（秒）
`bulk_flush_frequency`	（整数）发送批量 Kafka 请求之前的等待时间。 0` 表示无延迟。默认值： `0`
`bulk_max_size`	（整数）单个 Kafka 请求中批量处理的最大事件数。默认值： `2048`
`channel_buffer_size`	（整数）每个 Kafka 代理在输出管道中缓冲的消息数。默认值： `256`
`codec`	输出编解码器配置。您可以指定 `json` 或 `format` 编解码器。默认情况下，使用 `json` 编解码器。 `json.pretty`：如果将 `pretty` 设置为 true，则事件将格式化良好。默认值为 false。 `json.escape_html`：如果将 `escape_html` 设置为 true，则字符串中的 html 符号将被转义。默认值为 false。使用启用了漂亮打印的 `json` 编解码器将事件写入控制台的示例配置 output.console: codec.json: pretty: true escape_html: false `format.string`：用于创建自定义格式消息的可配置格式字符串。使用 `format` 编解码器将事件时间戳和消息字段打印到控制台的可配置示例 output.console: codec.format: string: '%{[@timestamp]} %{[message]}'
`compression`	选择要使用的压缩编解码器。支持的编解码器有 `snappy`、`lz4` 和 `gzip`。
`compression_level`	对于 `gzip` 编解码器，您可以选择压缩级别。级别必须在 `1`（最佳速度）到 `9`（最佳压缩）的范围内。提高压缩级别会减少网络使用量，但会增加 CPU 使用量。默认值： `4`。
`keep_alive`	（字符串）活动网络连接的保持活动时间段。如果为 `0s`，则禁用保持活动。默认值： `0s`
`max_message_bytes`	（整数）JSON 编码消息允许的最大大小。更大的消息将被丢弃。此值应等于或小于代理的 `message.max.bytes`。默认值： `1000000`（字节）
`metadata`	Kafka 元数据更新设置。元数据包含有关代理、主题、分区和用于发布的活动领导者的信息。 `refresh_frequency` 元数据刷新间隔。默认为 10 分钟。 `full` 获取元数据时使用的策略。当此选项为 `true` 时，客户端将维护所有可用主题的完整元数据集。设置为 `false` 时，它只会刷新已配置主题的元数据。默认值为 false。 `retry.max` 元数据更新重试总次数。默认值为 3。 `retry.backoff` 重试之间的等待时间。默认值为 250 毫秒。
`required_acks`	代理所需的 ACK 可靠性级别。0=无响应，1=等待本地提交，-1=等待所有副本提交。默认值为 1。注意：如果设置为 0，Kafka 不会返回任何 ACK。消息可能会在出错时静默丢失。默认值： `1`（等待本地提交）
`timeout`	在超时之前等待 Kafka 代理响应的秒数。默认值为 30（秒）。默认值： `1000000`（字节）

« 配置 Elasticsearch 输出 Logstash 输出 »