Kafka はどのようにして 1 秒間に 1,500 万件のメッセージを処理するのでしょうか?

[[245749]]

[51CTO.com からのオリジナル記事] Apache Kafka は、New Relic (データ インテリジェンス プラットフォーム)、Uber、Square (モバイル決済会社) などの大企業で、スケーラブルで高スループット、信頼性の高いリアルタイム データ ストリーム システムを構築するために広く使用されている、人気の高い分散データ ストリーム プラットフォームです。

たとえば、New Relic の運用環境では、Kafka クラスターは 1 秒あたり 1,500 万件を超えるメッセージを処理でき、合計データ レートは 1Tbps に近づいています。

Kafka はデータ ストリームの処理を大幅に簡素化するため、多くのアプリケーション開発者やデータ管理の専門家からも支持を得ています。

ただし、大規模なシステムでの Kafka の適用はより複雑になります。消費者がデータフローに追いつけない場合、メッセージは表示される前に消えてしまうことがよくあります。

同時に、自動データ保持、高トラフィックのパブリッシュ/サブスクライブ (pub/sub) モードなどの制限により、システムのパフォーマンスに影響が出る可能性があります。

データ ストリームを保存するシステムが必要に応じて拡張できなかったり、安定性が信頼できない場合、睡眠や食事に支障が出ることが頻繁にあると言っても過言ではありません。

上記の複雑さを軽減するために、高スループットに対処するための Kafka クラスターに関する New Relic の 20 のベスト プラクティスを共有したいと思います。

次の 4 つの側面について詳しく説明します。

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

Kafkaの概念とアーキテクチャを素早く理解する

Kafka は効率的な分散メッセージング システムです。パフォーマンスの面では、データの冗長性と復元力が組み込まれており、高いスループットとスケーラビリティも備えています。

機能的には、自動化されたデータ ストレージ制限をサポートし、アプリケーションのデータ変換を「ストリーム」方式で提供し、「キーと値」のモデリング関係に従ってデータ ストリームを「圧縮」できます。

さまざまな *** プラクティスを理解するには、まず次のキーワードを理解しておく必要があります。

メッセージ

Kafka 内のレコードまたはデータの単位。各メッセージにはキーと対応する値があり、オプションのメッセージ ヘッダーが含まれる場合もあります。

プロデューサー

プロデューサーは Kafka トピックにメッセージを公開します。プロデューサーは、ランダムなポーリング方法やメッセージ キーに基づくパーティション分割アルゴリズムなど、トピック パーティションに公開する方法を決定します。

ブローカ

Kafka は分散システムまたはクラスターとして実行されます。クラスター内の各ノードはブローカーと呼ばれます。

トピック

トピックは、公開されるデータ レコードまたはメッセージのカテゴリです。コンシューマーはトピックをサブスクライブすることで、自分に書き込まれたデータを読み取ります。

トピックパーティション

異なるトピックは異なるパーティションに分割され、各メッセージにはオフセットが割り当てられます。通常、各パーティションは少なくとも 1 回または 2 回複製されます。

各パーティションにはリーダーがあり、各フォロワーには 1 つ以上のレプリカ (つまり、データのコピー) が保存されます。この方法により、ブローカーの障害を防ぐことができます。

クラスター内のすべてのブローカーはリーダーおよびフォロワーとして機能できますが、ブローカーはトピック パーティションのコピーを最大 1 つしか持つことができません。リーダーは、すべての読み取りおよび書き込み操作を実行するために使用できます。

オフセット

単一のパーティション内の各メッセージにはオフセットが割り当てられます。オフセットは、パーティション内のメッセージの一意の識別子として使用できる単調に増加する整数です。

消費者

コンシューマーは、トピック パーティションをサブスクライブすることで、Kafka からさまざまなトピック メッセージを読み取ります。次に、消費アプリケーション プロセスはメッセージを受信して​​、指定された作業を完了します。

消費者団体

消費者は消費者グループに従って論理的に分類できます。トピック パーティションは、グループ内のすべてのコンシューマーに均等に分散されます。

したがって、同じコンシューマー グループ内では、すべてのコンシューマーが負荷分散された方法で動作します。

つまり、同じグループ内のすべてのコンシューマーがすべてのメッセージを見ることができます。コンシューマーが「オフライン」状態の場合、パーティションは同じグループ内の別のコンシューマーに割り当てられます。これを「リバランス」と呼びます。

もちろん、グループ内のコンシューマーがパーティションの数よりも多い場合、一部のコンシューマーはアイドル状態になります。

逆に、グループ内のコンシューマーの数がパーティションの数より少ない場合、一部のコンシューマーは複数のパーティションからメッセージを受け取ります。

遅れ

コンシューマーの速度がメッセージが生成される速度に追いつけない場合、コンシューマーはパーティションからメッセージを読み取ることができないため遅延が発生します。

遅延は、パーティション ヘッダーの後ろのオフセットの数として表されます。遅延状態から回復する（「追いつく」）ために必要な時間は、コンシューマーが 1 秒あたりに処理できるメッセージの速度によって異なります。

計算式は次のとおりです: 時間 = メッセージ / (1 秒あたりの消費率 - 1 秒あたりの生成率)

パーティションのベストプラクティス

① パーティションのデータレートを理解し、適切なデータ保存スペースが確保されていることを確認する

ここでの「パーティション データ レート」とは、データが生成されるレートを指します。つまり、「平均メッセージ サイズ」と「1 秒あたりのメッセージ数」を掛けて得られるデータ レートによって、特定の時間内に保証できるデータ ストレージ領域のサイズ (バイト単位) が決まります。

データ レートがわからないと、特定の期間にデータを保存するために必要なスペースの量を正確に計算することはできません。

同時に、データ レートは、遅延を発生させずに単一のコンシューマーがサポートする必要がある最大パフォーマンス値も識別できます。

②他のアーキテクチャ上の要件がない限り、トピックを記述する際にはランダムパーティションを使用してください。

大規模な操作を行う場合、パーティション間でデータ レートが異なると管理が難しくなる可能性があります。

その理由は次の3つの側面から来ています。

• まず、「ホット」（スループットが高い）パーティションのコンシューマーは、必然的に同じグループ内の他のコンシューマーよりも多くのメッセージを処理するため、処理とネットワークのボトルネックが発生する可能性があります。
• 次に、データ レートが最も高いパーティションに設定された最大予約領域により、トピック内の他のパーティションのディスク使用量がそれに応じて増加します。
• 3 番目に、パーティション化されたリーダー関係に基づいて実装された負荷分散ソリューションは、リーダー関係をすべてのブローカーに単純に分散するよりも複雑です。同じトピックでは、「ホット」パーティションは他のパーティションの 10 倍の重量を「運ぶ」ことになります。

Topic Partition の使用については、「Kafka Topic Partition のさまざまな効果的な戦略」https://blog.newrelic.com/engineering/effective-strategies-kafka-topic-partitioning/ を参照してください。

消費者のためのベストプラクティス

③消費者がKafka 0.10より古いバージョンを実行している場合は、すぐにアップグレードしてください。

バージョン 0.8.x では、Consumer は Apache ZooKeeper を使用して Consumer グループを調整しますが、多くの既知のバグにより、再バランス状態が長時間続いたり、再バランス アルゴリズムの失敗 (「再バランス ストーム」と呼びます) に直接つながったりする可能性があります。

したがって、再バランス調整中は、同じグループ内の各コンシューマーに 1 つ以上のパーティションが割り当てられます。

再バランス ストームの間、パーティションの所有権はコンシューマー間で引き続き渡されるため、どのコンシューマーもパーティションの所有権を実際に取得できなくなります。

④ 高速なデータ流入に対応するためにコンシューマのソケットバッファを調整する

Kafka バージョン 0.10.x では、パラメータ receive.buffer.bytes のデフォルト値は 64 KB です。 Kafka バージョン 0.8.x では、パラメータ socket.receive.buffer.bytes のデフォルト値は 100 KB です。

両方のデフォルト値は、特にブローカーとコンシューマー間のネットワークの帯域幅遅延積がローカル エリア ネットワーク (LAN) より大きい場合、高スループット環境には小さすぎます。

レイテンシが 1 ミリ秒以上の高帯域幅ネットワーク (10 Gbps 以上など) の場合は、ソケット バッファーを 8 MB または 16 MB に設定することを検討してください。

メモリが不足している場合は、少なくとも 1 MB に設定することを検討してください。もちろん、これを -1 に設定することもできます。これにより、基盤となるオペレーティング システムが実際のネットワーク状況に基づいてバッファー サイズを調整できるようになります。

ただし、「ホット」パーティションを起動する必要があるコンシューマーの場合、自動調整はそれほど高速ではない可能性があります。

⑤ 必要に応じてバックプレッシャーを実装するために、高スループットのコンシューマーを設計する

一般的に、システムがその能力の範囲内でのみデータを処理するようにし、プロセスが中断またはハングしたり、コンシューマー グループがオーバーフローしたりする可能性があるデータによるシステムの過負荷を回避する必要があります。

Java 仮想マシン (JVM) で実行している場合、コンシューマーは固定サイズのバッファ (できればオフヒープ) を使用する必要があります。 Disruptor パターンを参照してください: http://lmax-exchange.github.io/disruptor/files/Disruptor-1.0.pdf

固定サイズのバッファにより、コンシューマが大量のデータをスタックにプルして、JVM がメッセージ処理という重要な作業を実行する代わりにガベージ コレクションの実行にすべての時間を費やすことがなくなります。

⑥ JVM上でさまざまなコンシューマを実行する場合、ガベージコレクションがそれらに与える影響に注意してください。

たとえば、ガベージ コレクションの一時停止が長時間続くと、ZooKeeper セッションが破棄されたり、コンシューマー グループが再バランス状態になったりする可能性があります。

ブローカーについても同様です。ガベージ コレクションが長時間停止すると、クラスターが切断されるリスクがあります。

プロデューサーのためのベストプラクティス

⑦プロデューサーが各種確認を待つように設定する

これにより、プロデューサーはメッセージが実際にブローカーのパーティションに送信されたかどうかを知ることができます。 Kafka 0.10.x では、設定は Acks です。 0.8.x では、request.required.acks です。

Kafka はレプリケーションを通じてフォールト トレランスを提供するため、単一ノードの障害やパーティション リーダー関係の変更がシステムの可用性に影響を与えることはありません。

プロデューサーを Acks (または「fire and forget」) で構成しないと、メッセージが暗黙的に失われる可能性があります。

⑧プロデューサーごとに再試行を設定する

デフォルト値は 3 ですが、これはもちろん非常に低い値です。ただし、正しい設定はアプリケーションによって異なります。つまり、データ損失を一切許容しないアプリケーションの場合は、Integer.MAX_VALUE (有効かつ安全) に設定することを検討してください。

これにより、ブローカーのリーダー パーティションが Produce 要求にすぐに応答できない状況に対処できるようになります。

⑨ 高スループットプロデューサーの場合はバッファサイズを調整する

具体的には、buffer.memory と batch.size (バイト単位)。 batch.size はパーティションに応じて設定されるため、プロデューサーのパフォーマンスとメモリ使用量はトピック内のパーティションの数に関係する可能性があります。

したがって、ここでの設定値は次の要因によって異なります。

• プロデューサーデータレート（メッセージのサイズと数）
• 生成するパーティションの数
• 利用可能なメモリの量

バッファを大きくすることが必ずしも良いことではないことに注意してください。プロデューサーが何らかの理由で失敗した場合 (たとえば、リーダーの応答が確認応答よりも遅い場合)、ヒープ上にバッファリングされるデータが増えるほど、収集する必要があるガベージも増えます。

⑩ 生成されたメッセージの数、平均メッセージサイズ、消費されたメッセージの数などの指標を追跡するためにアプリケーションをインストルメント化する

ブローカーのためのベストプラクティス

⑪ 各ブローカーでは、トピックに必要なメモリとCPUリソースを圧縮してください。

ログ圧縮 (https://kafka.apache.org/documentation/#compaction を参照) では、各ブローカーのスタック (メモリ) と CPU サイクルが正常に連携する必要があり、失敗したログ圧縮データが増え続けると、ブローカーのパーティションにリスクが生じます。

ブローカーの log.cleaner.dedupe.buffer.size および log.cleaner.threads パラメータを調整できますが、両方の値が個々のブローカーのスタック使用量に影響することに注意してください。

ブローカーが OutOfMemoryError 例外をスローすると、ブローカーはシャットダウンされ、データが失われる可能性があります。

バッファのサイズとスレッド数は、クリアする必要があるトピック パーティションの数、およびそれらのパーティション内のメッセージのデータ レートとキー サイズによって異なります。

Kafka バージョン 0.10.2.1 以降では、ログ クリーナーのログ ファイルで ERROR エントリを監視することが、スレッドで発生する可能性のある問題を検出する最も信頼性の高い方法です。

⑫ネットワークスループットによるブローカーの監視

送信 (TX) および受信 (RX) トラフィック、ディスク I/O、ディスク容量、CPU 使用率を監視してください。容量計画は、クラスターの全体的なパフォーマンスを維持するための重要なステップです。

⑬ クラスタのブローカー間でパーティションのリーダー関係を分散する

リーダーには通常、大量のネットワーク I/O リソースが必要です。たとえば、レプリケーション係数を 3 に設定して実行する場合。

リーダーは、まずパーティション化されたデータを取得し、次に 2 セットのコピーを他の 2 人のフォロワーに送信し、次にデータを必要とする複数のコンシューマーに送信する必要があります。

したがって、この例では、単一のリーダーによって使用されるネットワーク I/O は、フォロワーの少なくとも 4 倍になります。さらに、リーダーはディスク上で読み取り操作を実行する必要がある一方、フォロワーは書き込み操作のみを実行する必要がある場合があります。

⑭ ブローカーの同期レプリカ（ISR）の縮小、レプリケーション不足のパーティション、および優先されないリーダーの監視を無視しないでください。

これらはすべて、クラスター内の潜在的な問題の兆候です。たとえば、単一のパーティションで ISR が頻繁に縮小される場合、パーティションのデータ レートがリーダーの容量を超え、コンシューマーや他のレプリカ スレッドにサービスを提供できなくなることを意味します。

⑮必要に応じてApache Log4jの各種プロパティを変更する

詳細については、https://github.com/apache/kafka/blob/trunk/config/log4j.properties を参照してください。

Kafka のブローカー ログは大量のディスク領域を消費しますが、完全にオフにすることはできません。

事故後、イベントのシーケンスを再構築する必要がある場合があり、その場合、ブローカー ログが最善、あるいは唯一の方法になります。

⑯ トピックの自動作成を無効にするか、未使用のトピックのクリーンアップ戦略を確立する

たとえば、設定された x 日間の期間内に新しいメッセージが表示されない場合は、トピックを無効と見なし、クラスターから削除する必要があります。これにより、クラスターで作成される追加のメタデータの管理に費やす時間が節約されます。

⑰持続的に高いスループットを持つブローカーの場合は、ディスクサブシステムからの読み取りを回避するために十分なメモリを提供します。

可能な限り、オペレーティング システムのキャッシュからパーティション データを直接取得するようにしてください。ただし、これは、コンシューマーが追いつくことができることを確認する必要があり、遅れているコンシューマーに対しては、ブローカーにディスクからの読み取りを強制することしかできないことを意味します。

⑱高スループットのサービスレベル目標（SLO）を持つ大規模クラスタの場合、ブローカーのサブセットごとに異なるトピックを分離することを検討してください。

分離する必要があるトピックをどのように決定するかは、完全に独自のビジネス ニーズによって異なります。たとえば、同じクラスターを使用する複数のオンライン トランザクション処理 (OLTP) システムがあるとします。

各システムのトピックを異なるブローカーのサブセットに分離すると、潜在的なイベントの影響範囲を制限するのに役立ちます。

⑲古いクライアントで新しいトピック メッセージ形式を使用します。クライアントの代わりに、各ブローカーに追加のフォーマット変換サービスをロードする必要があります。

もちろん、*** はこのような状況を避けるように努めるべきです。

⑳ ローカルホスト上でブローカーをテストすれば、本番環境での実際のパフォーマンスが反映されると誤解しないでください。

レプリケーション係数が 1 のループバック インターフェイスを介したパーティションのテストは、ほとんどの運用環境とは大きく異なることに注意してください。

ループバック インターフェイスのネットワーク遅延はほとんど無視できるほど小さく、レプリケーションが行われない場合、リーダーの確認応答を受信するのに必要な時間は大幅に変化する可能性があります。

要約する

上記の提案が Kafka をより効果的に使用するために役立つことを願っています。 Kafka の専門知識を高めたい場合は、クラスターの操作に関する役立つ情報が記載されている、Kafka ドキュメントの「操作」セクションを参照してください。

[51CTO オリジナル記事、パートナーサイトに転載する場合は、元の著者とソースを 51CTO.com として明記してください]