Kafka の高パフォーマンススループットを公開

最も人気のあるオープンソースメッセージングシステムである Kafka は、データバッファリング、非同期通信、ログ収集、システム分離などで広く使用されています。RocketMQ などの他の一般的なメッセージングシステムと比較して、Kafka はほとんどの機能特性を保証するだけでなく、最高レベルの読み取りおよび書き込みパフォーマンスも提供します。

[[312096]]

この記事では、Kafka のパフォーマンスを簡単に分析します。まず、Kafka のアーキテクチャと関連する用語を簡単に紹介しましょう。

トピック: メッセージを分割するために使用される論理概念。トピックは複数のブローカーに分散できます。
パーティション: Kafka における水平拡張とすべての並列化の基礎となります。各トピックは少なくとも 1 つのパーティションに分割されます。
オフセット: パーティション内のメッセージ番号。番号付けのシーケンスはパーティションを越えません。
コンシューマー: ブローカーからメッセージを取得/消費するために使用されます。
プロデューサー: ブローカーにメッセージを送信/生成するために使用されます。
レプリケーション: Kafka は、パーティション単位でのメッセージの冗長バックアップをサポートします。各パーティションは、少なくとも 1 つのレプリケーションで構成できます (レプリケーションが 1 つしかない場合は、パーティション自体のみです)。
リーダー: レプリケーションセット内の各パーティションは一意のリーダーを選択し、すべての読み取りおよび書き込み要求はリーダーによって処理されます。他のレプリカは、リーダーからのデータ更新をローカルマシンに同期します。このプロセスは、MySQL でよく知られている Binlog 同期に似ています。
ブローカー: ブローカーは、プロデューサーとコンシューマーからのリクエストを受け入れ、メッセージをローカルディスクに保持するために Kafka で使用されます。各クラスターでは、コントローラーとして機能するブローカーが選出され、パーティションのリーダー選出の処理とパーティションの移行の調整を担当します。
ISR (In-Sync Replica): レプリカのサブセットであり、現在稼働しており、リーダーに「追いつく」ことができるレプリカのセットを示します。読み取りと書き込みは両方とも最初にリーダーで実行されるため、一般的に、同期メカニズムを通じてリーダーからデータを取得するレプリカノードには、リーダーとの遅延 (遅延時間と遅延回数を含む) が発生します。いずれかがしきい値を超えると、レプリカノードは ISR から追い出されます。各パーティションには独自の独立した ISR があります。

上記は、Kafka を使用するときに遭遇する可能性のある名詞のほぼすべてです。同時に、それらはすべて中核となる概念またはコンポーネントです。設計自体から見ると、Kafka はまだ十分にシンプルであると感じます。この記事では、Kafka の優れたスループット性能に焦点を当て、その設計と実装で使用されているさまざまな「ブラックテクノロジー」を 1 つずつ紹介します。

ブローカ

Redis や MemcacheQ などのメモリメッセージキューとは異なり、Kafka は、より強力なストレージ機能と引き換えに、すべてのメッセージを低速で大容量のハードディスクに書き込むように設計されています。実際、Kafka のハードディスクの使用はパフォーマンスの低下をそれほどもたらさず、「適切な」方法で「ショートカット」をとります。

まず、「適切に動作する」と言う理由は、Kafka がディスク上でシーケンス I/O のみを実行するためです。メッセージシステムでの読み取りと書き込みの特殊性により、これは問題を引き起こしません。ディスク I/O のパフォーマンスに関しては、Kafka (Raid-5、7200rpm) によって提供されたテストデータセットを以下に示します。

シーケンスI/O: 600MB/秒

ランダムI/O: 100KB/秒

したがって、シーケンス I/O の使用のみを制限することで、ディスクアクセス速度の低下によるパフォーマンスへの影響を回避できます。

次に、Kafka がどのように「近道」をするかについて説明します。

まず、Kafka は基盤となるオペレーティングシステムによって提供される PageCache 機能に大きく依存しています。上位層で書き込み操作が行われると、オペレーティングシステムはデータを PageCache に書き込み、Page 属性を Dirty としてマークします。読み取り操作が発生すると、最初に PageCache が検索されます。ページが欠落している場合は、ディスクスケジューリングが実行され、最終的に必要なデータが返されます。実際、PageCache はディスクキャッシュとして可能な限り多くの空きメモリを使用します。同時に、他のプロセスがメモリを要求した場合、PageCache を再利用するためのコストは非常に小さいため、最近の OS はすべて PageCache をサポートしています。

PageCache 関数を使用すると、JVM 内でのデータのキャッシュも回避できます。 JVM は強力な GC 機能を提供しますが、Kafka の設計には当てはまらないいくつかの問題も発生します。

キャッシュがヒープ内で管理されている場合、JVM の GC スレッドはヒープ領域を頻繁にスキャンし、不要なオーバーヘッドが発生します。ヒープが大きすぎる場合、フル GC を実行するとシステムの可用性に大きな問題が生じます。
JVM 内のすべてのオブジェクトには必然的にオブジェクトオーバーヘッド (過小評価してはならない) があり、これによりメモリの有効なスペース使用率が低下します。
すべてのインプロセスキャッシュは OS 内に同じ PageCache を持ちます。したがって、キャッシュを PageCache にのみ配置することで、使用可能なキャッシュ領域を少なくとも 2 倍にすることができます。
Kafka を再起動すると、すべてのインプロセスキャッシュは無効になりますが、OS によって管理される PageCache は引き続き使用できます。

PageCache は単なる最初のステップです。 Kafka は、パフォーマンスをさらに最適化するために Sendfile テクノロジーも使用します。 Sendfile を説明する前に、まずは従来のネットワーク I/O 操作プロセスを紹介します。これは、大きく分けて次の 4 つのステップに分かれています。

OS はハードディスクからデータをカーネル領域の PageCache に読み込みます。
ユーザープロセスはカーネル領域からユーザー領域にデータをコピーします。
次に、ユーザープロセスはデータをソケットに書き込み、そのデータはカーネル領域のソケットバッファーに流れ込みます。
次に、OS はバッファからネットワークカードのバッファにデータをコピーし、転送を完了します。

プロセス全体には、2 つのコンテキストスイッチと 4 つのシステムコールが含まれます。同じデータがカーネルバッファーとユーザーバッファー間で繰り返しコピーされるため、非効率的です。手順 2 と 3 は不要で、カーネル領域で直接データのコピーを行うことができます。これはまさに Sendfile が解決する問題です。 Sendfile の最適化後、I/O プロセス全体は次のようになります。

上記の紹介から、Kafka の設計の本来の意図は、外部でメッセージングシステム全体として使用する場合でも、内部で基盤となるオペレーティングシステムと対話する場合でも、データ交換をメモリ内で完了するようにあらゆる努力をすることであることが容易にわかります。プロデューサーとコンシューマー間の生産と消費の進行が適切に調整されていれば、I/O ゼロでデータ交換を完全に実現できます。これが、Kafka の「ハードディスク」の使用によってパフォーマンスがそれほど低下しないと言う理由です。以下は実稼働環境で収集したメトリックの一部です。

(ブローカー 20 個、ブローカーあたり 75 個のパーティション、110k メッセージ/秒)

現時点では、クラスターには書き込み操作のみがあり、読み取り操作はありません。パーティション間のレプリケーションにより、約 10M/s の送信トラフィックが生成されます。受信速度と書き込み速度の比較から、ディスクは非同期+バッチ方式を使用して書き込まれており、基盤となる OS によってディスクの書き込み順序も最適化されている可能性があることがわかります。読み取り要求が届く場合、2 つの状況が考えられます。 1つ目は、データの交換がメモリ内で完了することです。

送信トラフィックは平均 10M/s から平均 60M/s に増加しましたが、ディスク読み取りトラフィックは 50KB/s を超えませんでした。 PageCache を使用すると、ディスク I/O を大幅に削減できます。

次のステップは、しばらく受信され、メモリからスワップアウトされてディスクにフラッシュされた古いデータを読み取ることです。

他の指標は同じままですが、ディスク読み取りは 40 MB/秒以上に急上昇しました。この時点で、すべてのデータがハードディスクに転送されています (OS レイヤーは、ハードディスクの順次読み取りのために Prefill PageCache を最適化します)。まだパフォーマンスの問題はありません。

ヒント

Kafka は、データの信頼性はレプリカによって保証されるべきであり、データを強制的にディスクにフラッシュすると全体的なパフォーマンスに影響すると考えているため、ブローカー側で log.flush.interval.messages および log.flush.interval.ms を介してディスクへの書き込みを強制することを公式には推奨していません。

/proc/sys/vm/dirty_background_ratio と /proc/sys/vm/dirty_ratio を調整することでパフォーマンスを調整できます。

ダーティページ比率が最初のインジケーターを超えると、pdflush は Flush Dirty PageCache を開始します。

ダーティページレートが 2 番目のインジケーターを超えると、すべての書き込み操作がフラッシュのためにブロックされます。

さまざまなビジネスニーズに応じて、dirty_background_ratio を適切に下げたり、dirty_ratio を上げたりすることができます。

パーティション

パーティションは、Kafka が水平方向にスケーリングし、高同時処理を提供し、レプリケーションを実装するための基盤です。

スケーラビリティの観点から。まず、Kafka では、パーティションをクラスター内のブローカー間で任意に移動して、起こり得るデータスキューの問題のバランスを取ることができます。次に、Partition はカスタムパーティション分割アルゴリズムをサポートします。たとえば、同じキーを持つすべてのメッセージを同じパーティションにルーティングできます。同時に、リーダーは In-Sync レプリカに移行することもできます。パーティションのすべての読み取りおよび書き込み要求はリーダーによってのみ処理されるため、Kafka はネットワークトラフィックの過度の集中を避けるために、クラスター内の各ノードにリーダーを均等に分散しようとします。

並行性の側面。パーティションは、一度にコンシューマーグループ内の 1 つのコンシューマーのみが使用できます (逆に、1 つのコンシューマーが複数のパーティションを同時に使用することもできます)。 Kafka の非常にシンプルなオフセットメカニズムは、ブローカーとコンシューマー間のやり取りを最小限に抑え、他の同様のメッセージキューのように下流のコンシューマーの数の増加に比例して Kafka のパフォーマンスが低下するのを防ぎます。さらに、複数のコンシューマーが時系列で非常に類似したデータを消費した場合、非常に高い PageCache ヒット率を達成できます。したがって、Kafka は、実際には基本的に単一マシンのネットワークカードの上限に達する可能性のある高同時読み取り操作を非常に効率的にサポートできます。

ただし、パーティションの数が多いほど良いです。パーティションの数が多いほど、ブローカーごとに平均してパーティションの数が多くなります。ブローカーのダウンタイム (ネットワーク障害、フル GC) の場合を考慮すると、コントローラーはクラッシュしたすべてのブローカーのすべてのパーティションのリーダーを再選出する必要があります。各パーティションの選択に 10 ミリ秒かかると仮定すると、ブローカーに 500 個のパーティションがある場合、選択の 5 秒間に、上記のパーティションに対する読み取りおよび書き込み操作によって LeaderNotAvailableException がトリガーされます。

さらに、クラッシュしたブローカーがクラスター全体のコントローラーである場合、最初に行うべきことは、ブローカーをコントローラーとして再任命することです。新しく任命されたコントローラーは、Zookeeper からすべてのパーティションのメタ情報を取得する必要があります。各情報を取得するには約 3 ～ 5 ミリ秒かかります。パーティションが 10,000 個ある場合、この時間は 30 ～ 50 秒になります。また、これはコントローラーを再起動するのにかかる時間だけであることを忘れないでください。これを踏まえて、リーダーを選出するための上記の時間も追加する必要があります-_-!!!!!!

さらに、ブローカー側では、プロデューサーとコンシューマーの両方にバッファメカニズムが使用されます。バッファのサイズは均一に設定されており、その数はパーティションの数と同じです。パーティションが多すぎると、プロデューサーバッファーとコンシューマーバッファーが大量のメモリを占有することになります。

ヒント

事前にパーティションの数を事前割り当てするようにしてください。パーティション数は後から動的に増やすことができますが、メッセージキーとパーティションの対応関係が壊れるリスクがあります。

レプリカが多すぎないようにしてください。条件が許せば、レプリカセット内のパーティションを別のラックに調整してみてください。

ブローカーを停止するたびにクリーンシャットダウンを実行できるように最大限の努力をしてください。そうしないと、サービスの復旧に長い時間がかかるだけでなく、データの破損やその他の非常に奇妙な問題も発生する可能性があります。

プロデューサー

KafkaのR&Dチームによれば、バージョン0.8ではプロデューサー全体がJavaで書き直され、パフォーマンスが大幅に向上したとのこと。私自身は試していないので、ここではデータの比較は行いません。この記事の最後にある拡張資料には、比較的優れていると思われるコントロールグループについて言及されています。興味のある学生はぜひ試してみてください。

実際、ほとんどのメッセージシステムでは、プロデューサー側で比較的単純な最適化方法を採用しており、これは部分を全体に統合し、同期を非同期に変更するだけです。

Kafka システムはデフォルトで MessageSet をサポートしており、複数のメッセージを自動的にグループにグループ化して送信するため、償却後の各通信の RTT が短縮されます。さらに、MessageSet を整理しながら、データを並べ替えて、バースト的なランダム書き込みを比較的安定した線形書き込みに最適化することもできます。

さらに、Producer がエンドツーエンドの圧縮をサポートしていることを強調することが重要です。データはローカルで圧縮され、ネットワーク経由で送信されます。通常、メッセージが消費され、クライアントで解凍されるまで、ブローカーでは解凍されません (Deep-Iteration が指定されていない場合)。

もちろん、ユーザーはアプリケーション層で圧縮と解凍の作業を自分で行うこともできます (結局のところ、Kafka は現在、GZIP と Snappy のみという限られた圧縮アルゴリズムをサポートしています)。ただし、そうすると予想外に効率が低下します。 Kafka のエンドツーエンドの圧縮は MessageSet と組み合わせると最も効果的に機能し、上記のアプローチは 2 つの間の接続を直接切断します。その理由は実はとても単純です。「データの繰り返しが多いほど圧縮率が高くなる」というのが圧縮アルゴリズムの基本原則です。メッセージ本文の内容や数に関係なく、ほとんどの場合、入力データの量が多いほど、圧縮率は高くなります。

ただし、Kafka の MessageSet の使用により、可用性がある程度低下することになります。データが送信されるたびに、Producer は send() の後にデータが送信されたと認識しますが、実際にはほとんどの場合、メッセージはまだメモリ内の MessageSet 内にあり、ネットワークにはまだ送信されていません。この時点でプロデューサーが電話を切ると、データは失われます。

この問題を解決するために、Kafka バージョン 0.8 の設計では、ネットワーク内の ack メカニズムを借用しました。高いパフォーマンス要件があり、ある程度のメッセージ損失を許容できる場合は、request.required.acks=0 を設定して ack をオフにし、全速力で送信することができます。送信されたメッセージを確認する必要がある場合は、request.required.acks を 1 または -1 に設定する必要があります。では、1 と -1 の違いは何でしょうか?ここで、先に説明したレプリカの数の問題について言及する必要があります。 1 に設定されている場合、メッセージはリーダーによって受信および確認されるだけでよく、他のレプリカは即時の確認なしに非同期的にプルできるため、効率を低下させることなく信頼性を確保できます。 -1 に設定されている場合、ack を返す前に、パーティションの ISR セット内のすべてのレプリカにメッセージがコミットされる必要があることを意味します。メッセージの送信はより安全になりますが、プロセス全体の遅延はレプリカの数に比例して増加します。ここでは、さまざまなニーズに応じて対応する最適化が必要です。

ヒント

特にミラーリングや移行で使用する場合は、プロデューサースレッドを多すぎる数に設定しないでください。そうしないと、ターゲットクラスター内のパーティションメッセージの無秩序が悪化します (アプリケーションシナリオがメッセージの順序に非常に敏感な場合)。

バージョン 0.8 の request.required.acks のデフォルト値は 0 (0.7 と同じ) です。

消費者

消費者側のデザインは、一般的に非常に従来型です。

コンシューマグループを通じて、プロデューサー/コンシューマモードとキューアクセスモードの両方をサポートできます。
コンシューマー API は高レベルと低レベルに分かれています。前者は Zookeeper に大きく依存しているため、パフォーマンスは悪く、無料ではありませんが、非常に安心です。 2 番目の方法は Zookeeper サービスに依存せず、自由度とパフォーマンスの点でより優れたパフォーマンスを発揮します。ただし、すべての例外 (リーダーの移行、オフセットの範囲外、ブローカーのダウンタイムなど) とオフセットのメンテナンスは、自分で処理する必要があります。
近日リリース予定の 0.9 リリースにご注目ください。開発者は、一連の Consumers を Java で書き直しました。 2 つの API セットが統合され、Zookeeper への依存関係が削除されます。パフォーマンスが大幅に向上したと言われています〜〜

ヒント

低レベル API を使用することを強くお勧めします。少し面倒ではありますが、これは、特にブローカー例外や不正シャットダウンによって発生した破損データを処理する場合に、エラーデータに対してカスタム処理を実行できる唯一の API です。そうでない場合は、スキップできず、「悪いニュース」がブローカーでローテーションされるのを待つことしかできません。この期間中、レプリカは利用できなくなります。

<<: 2020年の予測: クラウドコンピューティング業界で何が起こるか

>>: ハイブリッドクラウドのクイックガイド