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
• 此设置没有默认值

• 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"
    }
  }