Scala を使用した Apache Kafka 開発に役立つベスト 20

この記事の著者は、20 人の開発者とデータ サイエンティストが Apache Kafka をどのように使用しているかを徹底的に調査したソフトウェア エンジニアです。彼は最終的に、生産現場で注意を払う必要がある問題を、この記事に記載されている 20 の推奨事項にまとめました。

Apache Kafka は、New Relic、Uber、Square などの何千もの企業がスケーラブルで高スループット、信頼性の高いリアルタイム ストリーミング システムを構築するために使用している、人気の分散ストリーミング プラットフォームです。たとえば、New Relic の Kafka クラスターは 1 秒あたり 1,500 万件以上のメッセージを処理し、合計データ レートは 1 Tbps に近づきます。

Kafka は、データ ストリームの処理プロセスを大幅に簡素化するため、アプリケーション開発者やデータ サイエンティストの間で非常に人気があります。ただし、Kafka を Scala に実装するのは複雑です。自動データ保持制限を備えた高スループットのパブリッシュ/サブスクライブ パターンは、消費者がデータ フローに追いつけず、メッセージが消費者に表示される前に消えてしまう場合にはあまり役に立ちません。同様に、データ ストリームをホストするシステムが需要に合わせて拡張できなかったり、信頼性が低い場合は、ほとんど役に立ちません。

この複雑さを軽減するために、著者はユーザーの理解を容易にするために、考えられる質問を合計 20 項目の 4 つのカテゴリに分割しています。

• パーティション
• 消費者
• プロデューサー
• ブローカー

Kafka は、高いスループットとスケーラビリティを維持しながら、組み込みのデータ冗長性と回復力を提供する効率的な分散メッセージング システムです。自動データ保持制限が含まれているため、データをストリームとして表示するアプリケーションに適しており、キーと値のペアのマッピングをモデル化する「圧縮」ストリームもサポートされています。

*** の実践を理解する前に、いくつかの重要な用語を理解しておく必要があります。

• メッセージ: Kafka 内のレコードまたはデータの単位。各メッセージにはキーと値、およびオプションのヘッダーがあります。
• プロデューサー:プロデューサーは Kafka トピックにメッセージを公開します。プロデューサーは、ランダム (ラウンドロビン) に、またはメッセージ キーに基づくパーティショニング アルゴリズムを使用して、どのトピック パーティションに公開するかを決定します。
• ブローカー: Kafka は分散システムまたはクラスターで実行され、クラスター内の各ノードはブローカーと呼ばれます。
• トピック:トピックは、データ レコードまたはメッセージを公開するためのカテゴリです。コンシューマーはトピックをサブスクライブして、トピックに書き込まれたデータを読み取ります。
• トピック パーティション:トピックは複数のパーティションに分割され、各メッセージにはオフセットがあります。通常、各パーティションは少なくとも 1 回または 2 回複製されます。各パーティションにはリーダーがあり、ブローカーの障害から保護するためにフォロワー上に少なくとも 1 つのレプリカ (データ コピー) が存在します。クラスター内のすべてのブローカーはリーダーとフォロワーですが、ブローカーにはトピック パーティションのコピーが最大 1 つあり、すべての読み取りおよび書き込み操作にはリーダーが使用されます。
• オフセット:パーティション内の各メッセージにはオフセットが割り当てられます。オフセットは単調に増加する整数で、パーティション内のメッセージの一意の識別子として機能します。
• コンシューマー:コンシューマーは、トピック パーティションをサブスクライブし、アプリケーションを消費し、メッセージを処理して必要な作業を完了することにより、Kafka トピックからメッセージを読み取ります。
• コンシューマー グループ:コンシューマーをコンシューマー グループに編成し、トピック パーティションを割り当てて、グループ内のすべてのコンシューマーのバランスをとることができます。コンシューマー グループでは、すべてのコンシューマーが負荷分散モードで動作します。つまり、グループ内のすべての消費者がすべてのメッセージを見ることになります。コンシューマーが離脱すると、パーティションはグループ内の他のコンシューマーに割り当てられます。このプロセスはリバランスと呼ばれます。グループ内のコンシューマーがパーティションよりも多い場合、一部のコンシューマーはアイドル状態になります。グループ内のコンシューマーがパーティションの数より少ない場合、一部のコンシューマーは複数のパーティションからのメッセージを消費します。
• ラグ:コンシューマーがパーティションからメッセージを読み取ることができない場合、コンシューマーは、パーティションの先頭からのオフセット数として表されるラグを経験します。 Lag 状態から回復するのにかかる時間は、コンシューマーが 1 秒あたりにメッセージを消費する速度によって異なります。

時間= メッセージ / ( 1 秒あたりの消費率 - 1 秒あたりの生成率)

パート 1: パーティションの使用に関するベスト プラクティス!

パーティション分割の部分では、適切な予約領域があることを確認するために、パーティションのデータ レートを理解する必要があります。パーティションのデータ レートは、データが生成される速度です。つまり、平均メッセージ サイズに 1 秒あたりのメッセージ数を掛けた値になります。データ レートによって、特定の時点で必要な予約スペースの量 (バイト単位) が決まります。データ レートがわからないと、基本的な保持目標を満たすために必要なスペースの量を正確に計算することは不可能です。データ レートは、単一の消費者が遅延なくサポートする必要がある最大パフォーマンスを指定します。

アーキテクチャで別途要求されない限り、トピックに書き込むときにランダム パーティショニングを使用します。大規模に運用する場合、パーティション間のデータ レートの不均一性を管理するのは困難です。注目すべき点が 3 つあります。

1. まず、「ホット」(スループットが高い) パーティション内のコンシューマーは、コンシューマー グループ内の他のコンシューマーよりも多くのメッセージを処理する必要があり、処理とネットワークのボトルネックが発生する可能性があります。

2. 次に、データ レートが最も高いパーティションに合わせてトピックの予約領域のサイズを調整する必要があり、これによりトピック内の他のパーティションのディスク使用量が増加する可能性があります。

3. 最後に、パーティション リーダーシップに関して絶対的なバランスを達成することは、すべてのブローカーに単純に拡張するよりも複雑です。 「ホット」パーティションは、同じトピック内の別のパーティションの 10 倍の大きさになる場合があります。

パート 2: 消費者プライバシー慣行の使用!

コンシューマーが 0.10 未満のバージョンの Kafka を実行している場合は、アップグレードしてください。 0.8.x リリースでは、コンシューマーはコンシューマー グループの調整に Apache ZooKeeper を使用していましたが、長時間にわたるバランス調整や、再バランス調整アルゴリズムの失敗 (「再バランス調整ストーム」と呼んでいます) を引き起こす可能性のある既知のバグがいくつかありました。再バランス調整中に、コンシューマー グループ内の各コンシューマーに 1 つ以上のパーティションが割り当てられます。再バランス調整中は、パーティションの所有権が消費者間で変更され続けるため、どの消費者も消費を実際に進めることができなくなります。

4. 高速取得のためにコンシューマー ソケット バッファーを調整します。 Kafka 0.10.x では、パラメータはreceive.buffer.bytes で、デフォルトは 64kB です。 Kafka 0.8.x では、パラメータは socket.receive.buffer.bytes で、デフォルトは 100kB です。両方のデフォルト値は、特にブローカーとコンシューマー間のネットワーク帯域幅の遅延がローカル エリア ネットワーク (LAN) よりも大きい場合、高スループット環境には小さすぎる可能性があります。レイテンシが 1 ミリ秒以上の高帯域幅ネットワーク (10 Gbps 以上) の場合は、ソケット バッファーを 8 MB または 16 MB に設定することを検討してください。メモリが不足している場合は、1 MB を検討するか、または -1 の値を使用して、基盤となるオペレーティング システムがネットワークの状態に基づいてバッファー サイズを調整できるようにします。ただし、「ホットスポット」消費者をアクティブ化する必要があるシステムの場合、自動調整は遅くなる可能性があります。

5. プロセスが停止してコンシューマー グループから抜けてしまうほど消費するのではなく、効率的に処理できる分だけ消費し、保証されている場所でバックプレッシャーを実装するように高スループットのコンシューマーを設計します。コンシューマーは固定サイズのバッファを使用する必要があります (Disruptor パターンを参照)。Java 仮想マシン (JVM) で実行する場合は、オフヒープが望ましいです。固定サイズのバッファにより、コンシューマが大量のデータをヒープ上にプルすることを防ぎ、JVM がメッセージの処理という本来の目的を果たす代わりに、ガベージ コレクションにすべての時間を費やすことを防ぎます。

6. JVM 上でコンシューマーを実行する場合は、ガベージ コレクションがコンシューマーに与える影響に注意してください。たとえば、ガベージ コレクションの一時停止が長くなると、ZooKeeper セッションまたはコンシューマー グループのバランスが崩れる可能性があります。ブローカーについても同様であり、ガベージ コレクションの一時停止が長すぎる場合はブローカーがクラスターから排除される可能性があります。

パート 3: Producer*** を実際に使用してみましょう!

7. 確認を待つようにプロデューサーを構成します。これにより、プロデューサーはメッセージが実際にブローカーのパーティションに送信されたことを認識します。 Kafka 0.10.x では、これは acks に設定されています。 0.8.x では、request.required.acks です。 Kafka はレプリケーションを通じてフォールト トレランスを提供するため、単一ノードの障害やパーティション リーダーの変更が可用性に影響を与えることはありません。プロデューサーが ACK を持たないように構成されている場合 (「ファイア アンド フォーゲット」とも呼ばれます)、メッセージは暗黙的に失われる可能性があります。

8. プロデューサーの再試行回数を設定します。デフォルト値は 3 ですが、通常は低すぎます。正しい値は要件によって異なります。データ損失を許容できないアプリケーションの場合は、Integer.MAX_VALUE (実質的に無限大) を検討してください。これにより、リーダー パーティションのブローカーがプロダクション要求に即座に応答できなくなることが防止されます。

9. 高スループット プロデューサーの場合は、バッファー サイズ、特に buffer.memory と batch.size (バイト単位) を調整します。 batch.size はパーティションごとに設定されるため、プロデューサーのパフォーマンスとメモリ使用量はトピック内のパーティションの数と相関関係にある可能性があります。ここでの値は、プロデューサーのデータ レート (メッセージのサイズと数)、生成されるパーティションの数、使用可能なメモリの量など、いくつかの要因によって異なります。バッファが大きいほど良いとは限らないことに注意してください。プロデューサーが何らかの理由で停止した場合 (例: *** が ACK で応答するのが遅い)、ヒープ上にキャッシュされるデータが増えると、ガベージ コレクションが増える可能性があります。

10. 生成されたメッセージの数、生成されたメッセージの平均サイズ、消費されたメッセージの数などのアプリケーション追跡メトリックを開発します。

パート 4: ブローカー*** の練習!

11. トピックはブローカー上のメモリと CPU リソースを必要とし、ログ圧縮が正常に完了するにはブローカー上のヒープ (メモリ) と CPU サイクルが必要であり、ログ圧縮に失敗するとブローカーのパーティションが急速に拡大するリスクが生じます。ブローカー上で tunelog.cleaner.dedupe.buffer.size と log.cleaner.threads を操作できますが、これらの値はブローカー上のヒープ使用量に影響することに注意してください。ブローカーが OutOfMemoryError をスローすると、ブローカーはシャットダウンし、データが失われる可能性があります。バッファ サイズとスレッドの数は、クリーンアップするトピック パーティションの数と、それらのパーティション内のメッセージのデータ レートとキー サイズによって異なります。 Kafka バージョン 0.10.2.1 以降では、ログ クリーナー ログ ファイルで ERROR エントリを監視することが、ログ クリーナー スレッドの問題を検出する最も信頼性の高い方法になります。

12. ブローカーのネットワーク スループットを監視します。送信 (TX) と受信 (RX)、ディスク I/O、ディスク領域、CPU 使用率を使用してこれを実行してください。容量計画は、クラスターのパフォーマンスを維持する上で重要な部分です。

13. クラスター内のブローカー間でパーティション リーダーを分散します。これには大量のネットワーク I/O リソースが必要です。たとえば、レプリケーション係数 3 で実行する場合、リーダーはパーティション データを受信し、そのデータを使用するコンシューマーに送信する前に、それをすべてのレプリカに同期的に渡す必要があります。したがって、この例では、リーダーはディスクから読み取る必要があり、フォロワーは書き込みのみを行う必要があるため、リーダーはフォロワーの少なくとも 4 倍のネットワーク I/O を使用しています。

14. 同期レプリカ (ISR) の縮小、レプリケーション不足のパーティション、未入力のリーダーについてブローカーを監視することを怠らないでください。これらはクラスター内の潜在的な問題の兆候です。たとえば、単一のパーティションで ISR が頻繁に縮小される場合、そのパーティションのデータ レートが、リーダーがコンシューマーとレプリカ スレッドにサービスを提供する能力を超えていることを示している可能性があります。

15. 必要に応じて Apache Log4j のプロパティを変更します。 Kafka ブローカーのログ記録では、ディスク領域が大量に消費される可能性があります。ただし、ログ記録を完全に放棄しないでください。ブローカー ログは、イベントが発生した後にイベントのシーケンスを再構築するための最善の方法であり、場合によっては唯一の方法である可能性があります。

16. 自動トピック作成に関連する明示的なポリシーを無効にし、未使用のトピックを定期的にクリーンアップします。たとえば、x 日間メッセージが見られなかった場合、トピックは無効であると見なしてクラスターから削除します。これにより、クラスター内に管理する必要がある追加のメタデータが作成されるのを回避できます。

17. 持続的に高いスループットを実現するエージェントの場合は、ディスク システムからの読み取りを回避するために十分なメモリを提供し、可能な場合はオペレーティング システムのファイル システム キャッシュからパーティション データを直接提供します。ただし、これは消費者が対応できることを確認する必要があることを意味します。遅れている消費者はブローカーにディスクからの読み取りを強制します。

18. スループットの高いサービス レベル目標 (SLO) を持つ大規模なクラスターの場合は、トピックをブローカーのサブセットに分離することを検討してください。どのトピックを分離するかを決定する方法は、ビジネス ニーズによって異なります。たとえば、同じクラスターを使用する複数のオンライン トランザクション処理 (OLTP) システムがある場合、各システムのトピックをブローカーの異なるサブセットに分離すると、イベントの潜在的な影響範囲を制限するのに役立ちます。

19. 新しいトピック メッセージ形式を使用する古いクライアント (およびその逆) は、ブローカー クライアントが形式を変換するときにブローカー プログラムに追加の負担をかけます。このような状況をできるだけ避けてください。

20. ローカル デスクトップでブローカーをテストしても、実際の運用環境でのパフォーマンスが反映されると想定しないでください。レプリケーション係数 1 でパーティション化されたループバック インターフェイスをテストすることは、ほとんどの運用環境とはまったく異なるトポロジです。ループバックではネットワーク遅延はごくわずかであり、レプリケーションが関与しない場合は、リーダーからの確認応答を受信するのにかかる時間が大幅に異なる可能性があります。