Logstash|Logstash 参考指南（Kafka输出插件） Logstash参考指南（Kafka输出插件

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`	string，[`"0"`, `"1"`, `"all"`]中之一	No
`batch_size`	number	No
`bootstrap_servers`	string	No
`buffer_memory`	number	No
`client_id`	string	No
`compression_type`	string，[`"none"`, `"gzip"`, `"snappy"`, `"lz4"`]中之一	No
`jaas_path`	有效的文件系统路径	No
`kerberos_config`	有效的文件系统路径	No
`key_serializer`	string	No
`linger_ms`	number	No
`max_request_size`	number	No
`message_key`	string	No
`metadata_fetch_timeout_ms`	number	No
`metadata_max_age_ms`	number	No
`receive_buffer_bytes`	number	No
`reconnect_backoff_ms`	number	No
`request_timeout_ms`	string	No
`retries`	number	No
`retry_backoff_ms`	number	No
`sasl_kerberos_service_name`	string	No
`sasl_mechanism`	string	No
`security_protocol`	string，[`"PLAINTEXT"`,`"SSL"`,`"SASL_PLAINTEXT"`, `"SASL_SSL"`]中之一	No
`send_buffer_bytes`	number	No
`ssl_key_password`	password	No
`ssl_keystore_location`	有效的文件系统路径	No
`ssl_keystore_password`	password	No
`ssl_keystore_type`	string	No
`ssl_truststore_location`	有效的文件系统路径	No
`ssl_truststore_password`	password	No
`ssl_truststore_type`	string	No
`topic_id`	string	Yes
`value_serializer`	string	No

还可以查看所有输出插件支持的通用选项列表。
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
此设置没有默认值
【Logstash|Logstash 参考指南（Kafka输出插件）】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	No
`enable_metric`	boolean	No
`id`	string	No

codec

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

enable_metric

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

id

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

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