Kafkaスループットを極める：プロデューサー調整の必須テクニック

Kafkaプロデューサーのスループットは、通常、バッチ処理、圧縮、確認応答、パーティショニングで決まります。ブローカー側も重要ですが、小さな非圧縮リクエストを一度に1つずつ送信するプロデューサーは、強力なクラスターを無駄にする可能性があります。

実用的な目標はシンプルです。レイテンシと耐久性の要件を損なうことなく、より少なく、より完全なリクエストを送信することです。つまり、別のワークロードから単一の「高速」設定をコピーするのではなく、測定に基づいて調整する必要があります。

---

Kafkaプロデューサースループットの基礎を理解する

Kafkaプロデューサーのスループットは、クライアントがレコードを収集し、リクエストにパッケージ化し、適切なブローカーパーティションに送信する効率によって決まります。バッチ処理はメッセージごとのオーバーヘッドを削減しますが、レイテンシの動作も変化させます。数ミリ秒待機するバッチは、分析パイプラインには最適でも、インタラクティブなリクエストパスには受け入れられない場合があります。

スループット分析のための主要メトリクス

調整する際は、以下の領域に焦点を当てます。

1. バッチサイズ: 送信前に蓄積されるデータ量（バイト単位）。
2. リンガー時間: 不完全なバッチを送信する前に、プロデューサーがさらにメッセージを待機する時間。
3. 圧縮: 送信前にデータを圧縮する際のオーバーヘッド。

---

コア調整パラメータ1：バッチサイズ（batch.size）

batch.size設定パラメータは、リンガー時間に関係なく、プロデューサーがブローカーに送信する前に蓄積するバッチの最大サイズ（バイト単位）を指定します。

batch.sizeがスループットに与える影響

• 大きいbatch.size: ネットワーク使用率が最大化され、メッセージごとのオーバーヘッドが削減されるため、一般的に高いスループットにつながります。より多くのレコードをより少ないネットワークリクエストに収めることができます。
• 小さいbatch.size: プロデューサーが多数の小さな非効率的なリクエストを送信し、ネットワークオーバーヘッドが増加し、レイテンシが高くなる可能性があるため、低いスループットにつながる可能性があります。

実用的なヒント: 64KBや128KBなどの中程度の増加から始め、バッチサイズとリクエストレートのメトリクスを監視します。非常に大きなバッチは一部のワークロードに役立ちますが、アクティブなパーティションごとにより多くのメモリを消費し、最悪の場合のレイテンシを増加させる可能性があります。

設定例（プロデューサープロパティ）

# バッチサイズを64キロバイトに設定
batch.size=65536

サイズ超過に関する警告: batch.sizeは、処理中のレコードがあるパーティションごとに割り当てられます。多くのパーティションに書き込むプロデューサーは、これを積極的に上げると、予想よりもはるかに多くのメモリを使用する可能性があります。

---

コア調整パラメータ2：リンガー時間（linger.ms）

linger.msパラメータは、現在のバッチを強制的に送信する前に、プロデューサーが追加のレコードが到着してバッチを満たすのを待機する時間を制御します。これは、レイテンシとスループットのバランスを管理するための主要な制御です。

linger.msがスループットに与える影響

• 高いlinger.ms: プロデューサーがバッチを満たすための時間が長くなるため、スループットが向上することがよくあります。
• 低いlinger.ms: プロデューサー側の待機時間が短くなることが多いですが、リクエストが小さくなる可能性があります。

スループット指向のサービスの場合は、5や10などの小さな値から試し、レイテンシ予算が許せば増やします。リクエスト/レスポンスパスの場合は、値を低く保ち、テールレイテンシへの影響を確認してから増やします。

設定例（プロデューサープロパティ）

# バッチを満たすために最大50ミリ秒待機
linger.ms=50

---

コア調整パラメータ3：メッセージ圧縮

バッチのサイズが完璧でも、ネットワーク経由でデータを転送する時間は全体的なスループットに影響します。メッセージ圧縮により、ブローカーに送信されるデータの物理サイズが削減され、ネットワーク転送時間が短縮され、同じ時間枠内でより多くのメッセージを処理できるようになることがよくあります。

圧縮タイプと選択

compression.type設定で使用するアルゴリズムを決定します。一般的なオプションは次のとおりです。

アルゴリズム	特性
none	圧縮なし。圧縮のCPUコストを回避しますが、ネットワーク経由でより多くのバイトを送信します。
gzip	非常に優れた圧縮率。中程度のCPUオーバーヘッド。
snappy	非常に高速な圧縮/解凍。低CPUオーバーヘッド、中程度の圧縮率。多くの場合、最適なバランス。
lz4	高速な圧縮/解凍で、多くのワークロードに実用的なバランスを提供します。
zstd	多くの最新システムで強力な圧縮率と優れた速度を提供しますが、CPUコストをテストしてください。

圧縮は、ネットワーク帯域幅やブローカーのI/Oが制約となっている場合に、実効スループットを向上させることがよくあります。プロデューサーがすでにCPUバウンドの場合は、逆効果になる可能性があります。プロデューサーのCPU、ブローカーのネットワークバイト、リクエストレイテンシ、コンシューマーの解凍コストを測定します。

設定例（プロデューサープロパティ）

# 最適なバランスのためにsnappy圧縮を使用
compression.type=snappy

# GZIPを使用する場合、レベルをさらに調整できます（1が最速/最低圧縮）
# gzip.compression.level=6 

---

最大スループットのための高度なテクニック

基本的なバッチ処理パラメータが設定されたら、他のいくつかの設定でスループットの限界を押し上げることができます。

1. プロデューサースレッド数の増加（該当する場合）

アプリケーションロジックが許せば、並列性（同時にデータを送信するスレッド数）を増やすことで、スループットを直接スケーリングできます。各スレッドは独自のプロデューサーインスタンスとバッファを管理し、異なるパーティションやトピックへのデータ送信を同時に行えます。

2. Acks設定

acks設定は、耐久性の保証を制御します。プロデューサーが送信成功とみなす前に、いくつのブローカーが受信を確認する必要があるかを指定します。

• acks=0: ファイアアンドフォーゲット。高いスループットの可能性がありますが、プロデューサーはブローカーの確認を待ちません。
• acks=1: リーダーレプリカが確認します。良いバランス。
• acks=all（または-1）: すべての同期レプリカが確認します。最高の耐久性、最低のスループット。

重要なビジネスイベントの場合、べき等性を有効にしたacks=allは、スループットのコストに見合う価値があることがよくあります。使い捨てのテレメトリの場合、acks=1で許容できる場合があります。acks=0は、デフォルトの調整トリックではなく、意識的なデータ損失のトレードオフであるべきです。

3. バッファメモリ（buffer.memory）

この設定は、プロデューサー内のバッファリングに割り当てられる合計メモリを定義します。このバッファがいっぱいになると、プロデューサーは（送信成功またはタイムアウト/レコードドロップのいずれかによって）空き容量ができるまでブロックします。

ピーク時のデータ入力レートが持続的な送信レートを超える場合は、buffer.memoryを増やして、プロデューサーがすぐにブロックせずにバーストを吸収できるようにします。

# 内部バッファに64MBを割り当て
buffer.memory=67108864 

結果を変えるその他の設定

max.in.flight.requests.per.connectionは、プロデューサーが1つの接続で持つことができる未確認のリクエストの数を制御します。値を大きくするとスループットが向上する可能性がありますが、順序と再試行の動作が重要です。最新のKafkaクライアントでべき等性が有効になっている場合、クライアントは安全性を維持するために関連設定を制約します。

retriesとdelivery.timeout.msは、送信が失敗するまでプロデューサーが再試行を続ける時間を決定します。エラーを無視するスループットテストは誤解を招きます。プレッシャー下でレコードをドロップするために高速に見える設定は、ほとんどのシステムにとってスループットの向上ではありません。

request.timeout.msは、ブローカーとネットワークの現実に適合する必要があります。低すぎると、ブローカーの短い一時停止中に再試行ストームが発生する可能性があります。高すぎると、実際の障害が表面化するまでに時間がかかりすぎる可能性があります。

パーティション数も重要です。単一のパーティションは一度に1つのリーダーブローカーによって処理されるため、1つのホットキーが、クラスターに余剰容量がある場合でも、トピックのボトルネックになる可能性があります。すべてのレコードが同じキーを使用する場合、プロデューサーの調整では書き込みをパーティション全体に分散できません。batch.sizeを非難する前に、パーティションごとのバイト数とリクエストハンドラのメトリクスを確認します。

実用的な初期設定

少量のレイテンシ追加が許容される高ボリュームのイベントパイプラインの場合、妥当な最初のパスは次のようになります。

acks=all
enable.idempotence=true
compression.type=lz4
batch.size=131072
linger.ms=10
buffer.memory=67108864
delivery.timeout.ms=120000

レイテンシがより重要なサービスの場合は、より控えめに開始します。

acks=all
enable.idempotence=true
compression.type=snappy
batch.size=32768
linger.ms=1
buffer.memory=33554432

これらは普遍的な最適設定ではありません。測定のための出発点です。レコードが小さなJSONイベントの場合、圧縮は非常に役立ちます。レコードがすでに圧縮された画像やアーカイブの場合、圧縮はCPUを浪費する可能性があります。プロデューサーが数十のパーティションに均等に書き込む場合、メモリプレッシャーが予想よりも早く発生する可能性があります。

調整中に監視するメトリクス

プロデューサーの調整をアプリケーションのスループットだけで判断しないでください。プロデューサーのメトリクスも監視します。

• record-send-rate: 1秒あたりに送信されたレコード数。
• record-error-rate: 失敗した送信。
• request-latency-avgおよびメトリクスシステムが取得する場合の高パーセンタイルレイテンシ。
• batch-size-avg: より大きなbatch.sizeが実際に使用されているかどうか。
• buffer-available-bytesまたはバッファ枯渇のシグナル。
• record-queue-time-avg: レコードが送信されるまでの待機時間。

ブローカー側では、ネットワークバイト、リクエストハンドラのアイドル時間、アンダーレプリケーションパーティション、ディスクI/O、プロデュースリクエストレイテンシを監視します。プロデューサーは、トピックリーダー、ディスク、レプリケーション、ネットワークが許す速度でしか動作できません。

3つの一般的な調整シナリオ

クリックストリームやメトリクスイベントの場合、レコードは小さく頻繁であることがよくあります。スループットは通常、圧縮を有効にし、batch.sizeを上げ、少しのリンガーを許可することで向上します。主なリスクは、データがダウンストリーム分析に到達する前に遅延が大きくなりすぎることです。そのようなワークロードでは、linger.ms=10、compression.type=lz4またはzstdから始め、コンシューマーラグを確認します。

支払い、注文、監査イベントの場合、耐久性は通常、生のスループットよりも重要です。acks=allを維持し、べき等性を有効にし、acks=0を避けます。スループットが十分でない場合は、配信保証を弱める前に、パーティショニング、プロデューサーの並行性、ブローカー容量、メッセージサイズを確認します。監査イベントを失うことは、許容されるパフォーマンス最適化ではありません。

非常に大きなレコードの場合、バッチ処理は同じように役立たない場合があります。Kafkaは通常、適度なサイズのメッセージで最も快適に動作します。プロデューサーが巨大なペイロードを送信する場合は、ペイロードをオブジェクトストレージに保存し、Kafkaを介して参照を送信することを検討します。それが不可能な場合は、max.request.size、ブローカーのmessage.max.bytes、トピックのmax.message.bytes、コンシューマーのフェッチ制限を一緒に確認します。プロデューサーの調整だけでは、パイプラインのすべての部分に特大のレコードをプッシュする設計を修正できません。

自分を欺かないテスト

適切なスループットテストでは、本番環境と同様のレコードサイズ、キー、圧縮、パーティション数、ブローカーレプリケーションを使用します。1つの固定文字列を1つのテストトピックに送信しても、実際のサービスを代表するものではありません。

テストするときは、次のようなメモを残します。

レコードサイズ: 900-1400バイト JSON
キー: customer_id、ほぼ均等な分布
トピックパーティション: 24
レプリケーションファクター: 3
プロデューサーインスタンス: 6
acks: all
圧縮: lz4
batch.size: 131072
linger.ms: 10
観測された問題: 15分後にp99送信レイテンシが上昇、プロデューサーCPUが限界近く

この種の記録は、次の調整ステップを明確にします。CPUが限界に近い場合、圧縮を変更すると役立つ場合があります。バッチがまだ小さい場合は、リンガーを増やすか、トラフィックがパーティションごとにまばらすぎるかどうかを確認します。1つのブローカーがホットな場合は、パーティションリーダーシップとキー分布を調査します。

また、定常状態を確認するのに十分な長さテストを実行します。短いテストはページキャッシュに収まり、ログセグメントのロール動作を見逃し、後で現れるコンシューマーラグを回避する可能性があります。Kafkaのパフォーマンス問題は、多くの場合、最初のバースト中ではなく、バッファがいっぱいになった後に現れます。

プロデューサーの調整が間違った修正である場合

プロデューサーは遅い送信を報告するコンポーネントであるため非難されることがありますが、根本的な原因は別の場所にある場合があります。ブローカーのディスクが飽和している場合、linger.msをどれだけ注意深く調整しても、プロデュースレイテンシは上昇します。トピックのパーティションが少なすぎる場合、プロデューサーは十分なリーダーに書き込みを分散できません。すべてのレコードが同じキーを使用する場合、1つのパーティションがホットになり、トピックの残りの部分はほとんどアイドル状態になります。

クライアント設定を変更する前に、ボトルネックがパターンに従っているかどうかを確認します。

1つのパーティションがホット: キー分布またはパーティション数の問題
1つのブローカーのすべてのパーティションが遅い: ブローカーディスク、ネットワーク、またはコントローラーの問題
プロデューサーCPUが高い: 圧縮、シリアル化、またはアプリケーションのオーバーヘッド
プロデューサーバッファが枯渇: ブローカーがデータを十分な速さで受け入れられない、またはトラフィックバーストが大きすぎる
調整後にのみコンシューマーラグが上昇: プロデューサーがダウンストリーム処理を上回っている

最後のケースは見逃しがちです。プロデューサーのスループットを向上させると、より遅いコンシューマーグループ、クリーンアップが重いコンパクトトピック、またはより高速に取り込めないダウンストリームデータベースが露呈する可能性があります。健全なKafka調整の演習では、送信クライアントだけでなく、パイプライン全体を確認します。

反復的な調整が鍵

Kafkaプロデューサーの調整は、小さな実験ループとして最も効果的に機能します。1つのことを変更し、現実的な負荷テストを実行し、スループット、レイテンシ、エラー率、リソース使用量を比較します。

ほとんどの高スループットのユースケースでは、最適な設定には以下が含まれます。

1. 中程度のlinger.ms（例：5ms～50ms）を設定する。
2. 大きなbatch.size（例：128KB）を設定する。
3. 効率的な圧縮（snappyなど）を有効にする。

1つだけ覚えておくべきことは、トレードオフです。バッチを大きくし、圧縮を使用すると、通常はオーバーヘッドが削減されますが、レイテンシとCPU使用率が増加する可能性があります。適切な設定は、耐久性の要件を満たし、エラーを隠すことなく実際のトラフィックに追いつくものです。