配置 Kafka 输出
编辑配置 Kafka 输出
编辑Kafka 输出将事件发送到 Apache Kafka。
要使用此输出,请编辑 Filebeat 配置文件,通过注释掉来禁用 Elasticsearch 输出,并通过取消注释 Kafka 部分来启用 Kafka 输出。
对于 Kafka 0.10.0.0+ 版本,消息创建时间戳由 beats 设置,并等于事件的初始时间戳。这会影响 Kafka 中的保留策略:例如,如果一个 beat 事件在 2 周前创建,保留策略设置为 7 天,并且来自 beats 的消息今天到达 Kafka,那么由于时间戳值早于过去的 7 天,它将立即被丢弃。可以通过在消息到达时设置时间戳来更改此行为,这样消息就不会被丢弃,而是保留 7 天。为此,请在 Kafka 配置中将 log.message.timestamp.type
设置为 LogAppendTime
(默认 CreateTime
)。
配置示例
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
的事件将被丢弃。为了避免这个问题,请确保 Filebeat 不会生成大于 max_message_bytes
的事件。
兼容性
编辑此输出可以连接到 Kafka 0.8.2.0 及更高版本。较旧的版本可能也可以工作,但不受支持。
配置选项
编辑您可以在 filebeat.yml
配置文件的 kafka
部分中指定以下选项
hosts
编辑从中获取集群元数据的 Kafka 代理地址列表。集群元数据包含事件实际发布到的 Kafka 代理。
version
编辑Filebeat 连接时将请求的 Kafka 协议版本。默认为 1.0.0。
有效值是 0.8.2.0
和 2.6.0
之间的所有 Kafka 版本。
协议版本控制 Filebeat 可用的 Kafka 客户端功能;它不会阻止 Filebeat 连接到比协议版本更新的 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 身份验证。
要使用 GSSAPI
机制通过 Kerberos 进行身份验证,您必须将此字段留空,并使用 kerberos
选项。
topic
编辑用于生成事件的 Kafka 主题。
您可以使用格式化字符串来访问任何事件字段,从而动态设置主题。例如,此配置使用自定义字段 fields.log_topic
为每个事件设置主题
topic: '%{[fields.log_topic]}'
要了解如何向事件添加自定义字段,请参阅 fields
选项。
有关动态设置主题的其他方法,请参阅 topics
设置。
topics
编辑主题选择器规则的数组。每个规则指定用于与规则匹配的事件的 topic
。在发布期间,Filebeat 根据数组中第一个匹配的规则为每个事件设置 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
:如果无法计算哈希值或密钥值,则随机分配事件。
默认情况下,所有分区器都将尝试将事件发布到所有分区。如果某个分区的领导者对于 beat 变得不可访问,则输出可能会被阻止。所有分区器都支持设置 reachable_only
来覆盖此行为。如果将 reachable_only
设置为 true
,则事件将仅发布到可用的分区。
发布到可用分区的子集可能会增加资源使用量,因为事件的分配可能不均匀。
headers
编辑标头是一个键值对,同一个 key
可以包含多个标头。仅支持字符串值。这些标头将包含在每个生成的 Kafka 消息中。
output.kafka: hosts: ["localhost:9092"] topic: "logs-%{[agent.version]}" headers: - key: "some-key" value: "some value" - key: "another-key" value: "another value"
client_id
编辑用于日志记录、调试和审计目的的可配置 ClientID。默认为 "beats"。
metadata
编辑Kafka 元数据更新设置。元数据包含有关代理、主题、分区和用于发布的活动领导者的信息。
-
refresh_frequency
- 元数据刷新间隔。默认为 10 分钟。
-
full
- 获取元数据时要使用的策略,如果此选项为
true
,则客户端将维护所有可用主题的完整元数据集,如果此选项设置为false
,则它将仅刷新已配置主题的元数据。默认值为 false。 -
retry.max
- 当集群处于领导者选举中间时,元数据更新重试的总次数。默认为 3。
-
retry.backoff
- 领导者选举期间重试之间的等待时间。默认为 250 毫秒。
max_retries
编辑Filebeat 忽略 max_retries
设置并无限期重试。
backoff.init
编辑在发生网络错误后尝试重新发布到 Kafka 之前要等待的秒数。等待 backoff.init
秒后,Filebeat 会尝试重新发布。如果尝试失败,则回退计时器会呈指数增长,直至 backoff.max
。成功发布后,回退计时器将重置。默认为 1 秒。
backoff.max
编辑在发生网络错误后尝试重新发布到 Kafka 之前要等待的最大秒数。默认为 60 秒。
bulk_max_size
编辑单个 Kafka 请求中要批量处理的最大事件数。默认为 2048。
bulk_flush_frequency
编辑发送批量 Kafka 请求之前要等待的时间。0 表示没有延迟。默认为 0。
timeout
编辑等待 Kafka 代理响应超时前的秒数。默认为 30(秒)。
broker_timeout
编辑代理等待所需 ACK 数的最大持续时间。默认为 10 秒。
channel_buffer_size
编辑每个 Kafka 代理在输出管道中缓冲的消息数。默认为 256。
keep_alive
编辑活动网络连接的保持活动时间。如果为 0 秒,则禁用保持活动。默认为 0 秒。
compression
编辑设置输出压缩编解码器。必须是 none
、snappy
、lz4
、gzip
和 zstd
之一。默认值为 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(字节)。此值应等于或小于 Broker 的 message.max.bytes
。
required_acks
编辑从 Broker 请求的 ACK 可靠性级别。 0=无响应,1=等待本地提交,-1=等待所有副本提交。默认值为 1。
注意:如果设置为 0,Kafka 不会返回任何 ACK。 发生错误时,消息可能会被静默丢失。
ssl
编辑用于配置 SSL 参数(例如 Kafka 连接的根 CA)的选项。Kafka 主机密钥库应使用 -keyalg RSA
参数创建,以确保它使用 Filebeat 的 Kafka 库支持的密码。有关更多信息,请参见SSL。