Kafka输出插件

• 插件版本：v7.1.3
• 发布于：2018-08-27
• 更新日志

其他版本，请参阅版本化的插件文档。

获取帮助

有关插件的问题，请在讨论论坛中打开一个主题，对于bug或特性请求，在Github中打开一个issue，关于Elastic支持的插件列表，请考虑Elastic支持矩阵。

描述

写入事件到Kafka主题。

这个插件使用Kafka客户端1.1.0，有关broker兼容性，请参阅官方Kafka兼容性参考资料，如果链接的兼容wiki不是最新的，请联系Kafka支持/社区确认兼容性。

如果你需要这个插件中还没有提供的特性（包括客户端版本升级），请提交一个关于你需要什么细节的问题。

这个输出支持连接到Kafka：

• SSL（要求插件版本3.0.0或以上）
• Kerberos SASL（需要插件版本5.1.0或以上）

默认情况下，安全性是禁用的，但是可以根据需要打开。

唯一需要的配置是topic_id。

默认编解码器是plain，Logstash将使用消息字段以及时间戳和主机名对事件进行编码。

如果希望将事件的完整内容以json的形式发送，则应该在输出配置中设置编解码器，如下所示：

output {
  kafka {
    codec => json
    topic_id => "mytopic"
  }
}

有关更多信息，请参阅http://kafka.apache.org/documentation.html#theproducer。

Kafka生产者配置：http://kafka.apache.org/documentation.html#newproducerconfigs。

Kafka输出配置选项

这个插件支持以下配置选项以及后面描述的通用选项。

还可以查看所有输出插件支持的通用选项列表。

acks

• 值可以是：0，1，all
• 默认值为“1”
• 在考虑完成一个请求之前，生产者要求leader已经收到的确认的数量。acks=0，生产者将不会等待任何来自服务器的确认。acks=1，这意味着leader会将记录写入本地日志，但不会等待所有follower的完全确认。acks=all，这意味着leader将等待完整的同步副本来确认记录。

batch_size

• 值类型为number
• 默认值为16384
• 每当将多条记录发送到同一个分区时，生产者将尝试将记录批处理为更少的请求，这有助于客户端和服务器的性能，此配置以字节为单位控制默认批处理大小。

bootstrap_servers

• 值类型为string
• 默认值为"localhost:9092"
• 这是用于引导的并且生产者将仅用于获取元数据（主题、分区和副本），发送实际数据的socket连接将基于元数据中返回的broker信息建立，格式是host1:port1,host2:port2，并且列表可以是broker的子集，也可以是VIP，指向broker的子集。

buffer_memory

• 值类型为number
• 默认值为33554432
• 生产者可以使用的总内存字节来缓冲等待发送到服务器的记录。

client_id

• 值类型为string
• 此设置没有默认值
• 请求时要传递给服务器的id字符串，这样做的目的是通过允许在请求中包含逻辑应用程序名称来跟踪请求源，而不仅仅是ip/端口。

compression_type

• 值可以为：none、gzip、snappy、lz4
• 默认值为"none"
• 生产者生成的所有数据的压缩类型，默认值是none（即没有压缩），有效值是none、gzip或snappy

jaas_path

• 值类型为path
• 此设置没有默认值

• Java身份验证和授权服务（JAAS）API为Kafka提供用户身份验证和授权服务，这个设置提供了JAAS文件的路径，Kafka客服端的样例JAAS文件：

  KafkaClient {
    com.sun.security.auth.module.Krb5LoginModule required
    useTicketCache=true
    renewTicket=true
    serviceName="kafka";
    };

  请注意，在配置文件中指定jaas_path和kerberos_config将会添加到全局JVM系统属性中，这意味着如果你有多个Kafka输入，它们都共享相同的jaas_path和kerberos_config。如果不希望这样做，则必须在不同的JVM实例上运行Logstash的独立实例。

kerberos_config

• 值类型为path
• 此设置没有默认值
• kerberos配置文件路径的选项，这是krb5.conf样式，详见https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html。

key_serializer

• 值类型为string
• 默认值为"org.apache.kafka.common.serialization.StringSerializer"
• 用于序列化记录key的类。

linger_ms

• 值类型为number
• 默认值为0
• 生产者将在请求传输之间到达的任何记录组合成单个批处理请求，通常情况下，只有当记录到达的速度快于发送的速度时，才会出现这种情况。然而，在某些情况下，客户端可能希望减少请求的数量，即使在中等负载下也是如此，此设置通过添加少量人工延迟来完成此任务 - 也就是说，生产者不会立即发送一条记录，而是等待到给定的延迟，以便允许发送其他记录，以便将发送的记录打包在一起。

max_request_size

• 值类型为number
• 默认值为1048576
• 请求的最大大小。

message_key

• 值类型为string
• 此设置没有默认值
• 消息的key。

metadata_fetch_timeout_ms

• 值类型为number
• 默认值为60000
• 获取主题元数据的初始元数据请求的超时设置。

metadata_max_age_ms

• 值类型为number
• 默认值为300000
• 元数据刷新之前的最大时间（毫秒）。

receive_buffer_bytes

• 值类型为number
• 默认值为32768
• 读取数据时使用的TCP接收缓冲区的大小。

reconnect_backoff_ms

• 值类型为number
• 默认值为10
• 当连接失败时，在尝试重新连接到给定主机之前等待的时间量。

request_timeout_ms

• 值类型为string
• 此设置没有默认值
• 配置控制客户端等待请求响应的最长时间，如果超时之前没有收到响应，如有必要客户端将重新发送请求，或者在重试耗尽时失败请求。

retries

• 值类型为number
• 此设置没有默认值
• 默认的重试行为是重试直到成功，为了防止数据丢失，不鼓励使用此设置。如果你选择设置retries，大于零的值将导致客户端只重试固定次数，如果传输错误存在的时间超过了重试计数（网络中断、Kafka宕机等），则会导致数据丢失。小于零的值是配置错误。

retry_backoff_ms

• 值类型为number
• 默认值为100
• 在尝试重试对给定主题分区的生成请求失败之前等待的时间。

sasl_kerberos_service_name

• 值类型为string
• 此设置没有默认值
• Kafka broker运行的Kerberos主体名称，这可以在Kafka的JAAS配置或Kafka的配置中定义。

sasl_mechanism

• 值类型为string
• 默认值为"GSSAPI"
• 用于客户端连接的SASL机制，这可能是安全提供者可用的任何机制，GSSAPI是默认机制。

security_protocol

• 值可以是：PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL
• 默认值为"PLAINTEXT"
• 要使用的安全协议，可以是PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL。

send_buffer_bytes

• 值类型为number
• 默认值为131072
• 发送数据时使用的TCP发送缓冲区的大小。

ssl_key_password

• 值类型为password
• 此设置没有默认值
• 密钥存储文件中私有密钥的密码。

ssl_keystore_location

• 值类型为path
• 此设置没有默认值
• 如果需要客户端身份验证，则此设置存储密钥存储路径。

ssl_keystore_password

• 值类型为password
• 此设置没有默认值
• 如果需要客户端身份验证，则此设置存储密钥库密码

ssl_keystore_type

• 值类型为string
• 此设置没有默认值
• 密钥存储库类型。

ssl_truststore_location

• 值类型为path
• 此设置没有默认值
• JKS信任存储库路径用于验证Kafka broker的证书。

ssl_truststore_password

• 值类型为password
• 此设置没有默认值
• 信任存储库的密码。

ssl_truststore_type

• 值类型为string
• 此设置没有默认值
• 信任存储库类型。

topic_id

• 这是必需的设置
• 值类型为string
• 此设置没有默认值
• 生成消息的主题。

value_serializer

• 值类型为string
• 默认值为"org.apache.kafka.common.serialization.StringSerializer"
• 用于序列化记录值的类。

通用选项

所有输出插件都支持以下配置选项：

codec

• 值类型为codec
• 默认值为"plain"
• 用于输出数据的编解码器，输出编解码器是一种方便的方法，可以在数据离开输出之前对其进行编码，而不需要在你的Logstash管道中使用单独的过滤器。

enable_metric

• 值类型为boolean
• 默认值为true
• 为这个特定的插件实例禁用或启用指标日志记录，默认情况下，我们记录了所有我们可以记录的指标，但是你可以禁用特定插件的指标集合。

id

• 值类型为string
• 此设置没有默认值

• 在插件配置中添加唯一的id，如果没有指定ID，则Logstash将生成一个，强烈建议在配置中设置此ID，当你有两个或多个相同类型的插件时，这一点特别有用。例如，如果你有两个电子邮件输出，在本例中添加一个命名ID将有助于在使用监控API时监控Logstash。

  output {
    kafka {
      id => "my_plugin_id"
    }
  }