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