はじめに

fluentdからKafkaにデータ送信する際に、気にしておくべきfluentd側のチューニングポイントのまとめです。
fluent-plugin-kafkaのBufferedOutput前提です。
最初にデータが送信される際の流れを簡単に説明した後、チューニングポイントについて記載します。

使用ソフトウェアとバージョン

kafka 0.10
fluentd v0.12
fluent-kafka-plugin(BufferdOutput) v0.57

Kafkaへ送信される流れ

BufferedOutputプラグインではレコードはchunkと呼ばれる単位に区切られてバッファリングされます。
以下の記事にBufferedOutputの処理の流れが纏められており、参考になります。

本記事ではchunkがキューに入った後、Kafkaに送信されるにあたりどのような処理が行われるかについて追加で説明します。

chunk内の各レコードは1レコードづつ、Kafka送信用のキューにキューイングされます。
キューイングにはkafka_agg_max_bytesというリミットが設けられています。送信キューがkafka_agg_max_bytes未満の間はレコードは順次キューイングされていきます。
キューイング後の合計サイズがkafka_agg_max_bytesを超える場合、キューイングの前にキューの中身をまとめてKafkaにバッチ転送します。
Kafkaへバッチ転送をした後の挙動は、required_ackというパラメータやKafkaのレプリケーション有無によって動作が異なってきます。ここではKafkaがレプリケーションされてる前提で説明しています。

required_ack = -1：Kafkaは受け取ったデータに対して、全レプリケーションフォロワまでの書き込みが完了したタイミングでackを返します。fluentdはackを待って、次の処理に移ります。
required_ack = 1：Kafkaは受け取ったデータに対して、レプリケーションリーダ（自分自身）への書き込みが完了したタイミングでackを返します。fluentdはackを待って、次の処理に移ります。※並行してレプリケーションの処理は行われます。
required_ack = 0：Kafkaはackを返さず、fluentdもackを待たずに次の処理に移ります。

上記の1~4までの流れをチャンク内のレコードが無くなるまで、処理するというのがfluent-plugin-kafkaのざっくばらんな処理の流れになります。

チューニングポイント

上記を踏まえてfluentdのチューニングポイントです。

`num_threads`を上げる

デフォルトのnum_threadsは1です。
上述した参考リンクに説明がありますが、num_threadsを上げれば、chunkを複数スレッドで送信してくれます。
chunkの送信処理が詰まっていたなど、複数chunkがエンキューされていた状況であればスループットの上昇が期待できます。
ただし、複数チャンクが並列にKafkaに送信されるとレコードの順序性は保てなくなるので、その点で問題無いか注意が必要です。

`kafka_agg_max_bytes`を上げる

これを上げるのがスループットに最も効果があります。
fluentd - > KafkaでのスループットのキモはKafkaへバッチ転送した後のack待ちをいかに減らすかだと思っています。
そのため、キューイングの上限サイズを上げてack待ちの頻度を減らす事が最も有効です。
デフォルトのkafka_agg_max_bytesは4096(byte)です。
これは結構少ない値だな、と個人的には思っており、レコードのサイズや件数が多い環境では1mあたりまで上げておくとスループットの上昇が期待できます。

`required_ack`を変更する

デフォルトのrequire_ackは-1です。
ackを返すまでに必要な処理が多いため、ackが返ってくるまでの待ち時間は長めになり、待ち時間に引きづられる形で送信スループットも低下します。
このパラメータを1または0にすると、kafkaのack待ちの時間を減らす、あるいは0にする事ができるためスループットを上げる事ができます。
しかし、障害時などにログが欠損する可能性は上がるため、信頼度とスループットの兼ね合いで決めるのが良いでしょう。

おわりに

fluent-plugin-kafkaのBufferdOutputのパフォーマンスチューニングについて色々と書きました。
が、結論とりあえずkafka_agg_max_bytesを上げとけば、スループット上限をデフォルトからかなり伸ばす事ができます。
という訳で、ログの流量に対してスループットが十分に出なくて困った場合、まずはkafka_agg_max_bytesを上げるのを検証してみるのが良いのではないかなと思います。