Kafka はなぜこんなに速いのでしょうか?

ソフトウェア アーキテクチャの分野はここ数年で劇的に変化しました。すべてのシステムが単一のデータベースを共有する必要があるとは考えられなくなりました。

[[320430]]

画像はPexelsより

マイクロサービス、イベント駆動型アーキテクチャ、CQRS (コマンド クエリ責任分離) は、現代のビジネス アプリケーションを構築するための主なツールです。

さらに、モノのインターネット、モバイル デバイス、ウェアラブル デバイスの普及により、システムのほぼリアルタイムの機能に対する課題がさらに増大しています。

まず、多面的で複雑、そして非常に曖昧な「速い」という言葉について合意に達しましょう。 1 つの説明としては、「レイテンシ、スループット、ジッター」を「高速」の尺度として使用するというものがあります。

また、産業用アプリケーションなどの分野では、業界自体が「高速」に対する基準と期待を設定します。したがって、「速い」というのは、あなたの基準に大きく依存します。

Apache Kafka は、レイテンシとジッターを犠牲にしてスループットを最適化しますが、耐久性、厳密なレコード順序、少なくとも 1 回の配信セマンティクスなどの特性は犠牲にしません。

誰かが「Kafka は高速だ」と言った場合、少なくともある程度の能力がある人であれば、短時間で大量のレコードを配信できる Kafka の能力を指していると推測できます。

Kafka は、LinkedIn が 1 時間あたり数 TB のデータに相当する大量の情報を効率的に送信する必要があったときに LinkedIn で誕生しました。

当時は、ニュースの伝播が遅れることは許容範囲だと考えられていた。結局のところ、LinkedIn は高頻度取引を行う金融機関ではなく、定められた時間制限内で稼働する産業用制御システムでもありません。 Kafka はほぼリアルタイムのシステムに使用できます。

注: 「リアルタイム」は「高速」を意味するのではなく、「予測可能」を意味します。具体的には、リアルタイムとは、アクションを完了するための時間制限、つまり期限があることを意味します。

システムがこの要件を満たしていない場合、「リアルタイム システム」として分類することはできません。一定範囲の遅延を許容できるシステムは、「ほぼリアルタイム」システムと呼ばれます。スループットの観点から見ると、リアルタイム システムは、通常、準リアルタイム システムまたは非リアルタイム システムよりも遅くなります。

Kafka の速度については、別途議論する価値のある重要な側面が 2 つあります。

• これは、クライアントとサーバー間の非効率的な実装に関連しています。
• ストリーム処理の並列処理から派生したものです。

サーバー側の最適化

ログ保存

Kafka は、セグメント化されたログへの追加方法を使用して、読み取りと書き込みをシーケンシャル I/O に大幅に制限します。これは、ほとんどのストレージ メディアで高速です。ハードドライブは遅いというのはよくある誤解です。

ただし、ストレージ メディアのパフォーマンスは、データへのアクセス パターンに大きく依存します。一般的な 7200 RPM SATA ハード ドライブでは、ランダム I/O のパフォーマンスはシーケンシャル I/O よりも 3 ～ 4 桁遅くなります。

さらに、最新のオペレーティング システムでは、大量のデータをブロック単位で事前に読み取り、小さな論理書き込み操作を大きな物理書き込み操作にマージできる先読みおよび書き込み遅延テクノロジが提供されています。

したがって、フラッシュやその他のソリッド ステート不揮発性メディアでは、シーケンシャル I/O とランダム I/O のパフォーマンスの違いは依然として大きいですが、SSD などの回転ストレージではそれほど顕著ではありません。

レコードのバッチ処理

シーケンシャル I/O はほとんどのストレージ メディアで非常に高速であり、ネットワーク I/O の最高のパフォーマンスに匹敵します。実際には、これは、ネットワークの読み取りおよび書き込み速度に対応できる、適切に設計されたログ永続化レイヤーを意味します。実際、Kafka のパフォーマンスのボトルネックとなるのは、通常、ハードディスクではなくネットワークです。

したがって、オペレーティング システムによって提供されるバッチ処理に加えて、Kafka クライアントとサーバーは、読み取りレコードと書き込みレコードを含む複数のレコードをバッチで蓄積してから、それらをネットワーク経由で送信します。

レコードをバッチ処理すると、ネットワークのラウンドトリップのオーバーヘッドが軽減され、より大きなパケットが使用され、帯域幅の効率が向上します。

バッチ圧縮

圧縮が有効になっている場合、データのサイズが大きくなるにつれて圧縮の効果が増すため、バッチ処理への影響は特に顕著になります。

特に JSON などのテキストベースの形式を使用する場合、圧縮の効果は非常に大きくなり、圧縮率は通常 5 倍から 7 倍の範囲になります。

また、レコードのバッチ処理は主にクライアント操作として実行され、処理中に負荷が転送されるため、ネットワーク帯域幅だけでなく、サーバー側のディスク I/O 使用率にもプラスの影響が及びます。

安っぽい消費者

メッセージが消費されるとそれを削除する従来のメッセージ キュー モデル (ランダム I/O が発生) とは異なり、Kafka はメッセージが消費された後に削除せず、代わりに各コンシューマー グループのオフセットを個別に追跡します。

詳細については、Kafka の内部トピック __consumer_offsets を参照してください。繰り返しになりますが、これは単なる追加操作なので、非常に高速です。メッセージのサイズは、特定のコンシューマー グループの最後の既知のオフセットのみを保持するために、バックグラウンドでさらに縮小されます (Kafka の圧縮機能を使用)。

このモデルを、通常複数の異なるメッセージ配信トポロジを提供する従来のメッセージング モデルと比較してください。

1 つはメッセージ キューです。これは、ポイントツーマルチポイント機能のないポイントツーポイント メッセージングに使用される永続的な送信です。

もう 1 つは、パブリッシュ/サブスクライブ トピックではポイントツーマルチポイントのメッセージ通信が可能になりますが、永続性が犠牲になるという点です。従来のメッセージ キュー モデルで永続的なポイントツーマルチポイント メッセージ通信モデルを実装するには、ステートフル コンシューマーごとに専用のメッセージ キューを維持する必要があります。

これにより、読み取りと書き込みのコストが増加します。メッセージ プロデューサーは、複数のメッセージ キューにメッセージを書き込む必要があります。もう 1 つのオプションは、1 つのキューからレコードを消費し、それを複数の他のキューに書き込むファンアウト リレーを使用することですが、これによりレイテンシが増大するだけです。

また、一部のコンシューマーは、サーバー上で、順次とランダムの両方の読み取りと書き込みの I/O が混在する負荷を生成しています。

Kafka のコンシューマーは、ログ ファイルを変更しない限り「安価」です (変更できるのはプロデューサーまたは Kafka の内部プロセスのみです)。

つまり、クラスターに負担をかけることなく、多数のコンシューマーが同じトピックから同時にデータを読み取ることができます。コンシューマーを追加すると、依然としていくらかのコストがかかりますが、ほとんどはシーケンシャル読み取りと少数のシーケンシャル書き込みが混在します。

したがって、多様な消費者のシステムでは、トピックが共有されるのがごく普通です。

フラッシュされていないバッファ書き込み

Kafka のパフォーマンスのもう 1 つの根本的な理由 (さらに調査する価値がある理由) は、Kafka が書き込み操作を確認する前に fsync を呼び出さないことです。 ACK の唯一の要件は、レコードが I/O バッファに書き込まれていることです。

これはあまり知られていない事実ですが、非常に重要な事実です。実際、Kafka はメモリ内キューであるかのように動作します。Kafka は実際にはディスクでバックアップされたメモリ内キューです (バッファ/ページ キャッシュのサイズによって制限されます)。

ただし、この形式の書き込みは安全ではありません。レコードが ACK されたように見えても、レプリカ障害によりデータが失われる可能性があるためです。

つまり、リレーショナル データベースとは異なり、バッファーに書き込むだけでは永続性が確保されるわけではありません。 Kafka の耐久性は、複数の同期されたレプリカを実行することによって保証されます。

いずれか 1 つに障害が発生しても、他のコピー (複数あると想定) は引き続き実行されます (障害の原因が他のコピーにも障害を引き起こさないと想定)。

したがって、fsync を使用しない非ブロッキング I/O 方式と冗長同期レプリカの組み合わせにより、Kafka は高いスループット、耐久性、可用性を実現します。

クライアントの最適化

ほとんどのデータベース、キュー、およびその他の形式の永続化ミドルウェアは、本格的なサーバー (またはサーバーのクラスター) とシン クライアントの概念に基づいて設計されています。

一般的に、クライアント側はサーバー側よりも実装がはるかに簡単であると考えられています。サーバーが負荷の大部分を処理し、クライアントはサーバーのファサードとして機能するだけです。

Kafka はクライアント設計に対して異なるアプローチを採用しています。レコードがサーバーに到達する前に、クライアント上で多くの作業が行われます。

これには、アキュムレータ内のレコードのセグメント化、正しいパーティション インデックスを取得するためのレコード キーのハッシュ、レコードの検証、レコード バッチの圧縮が含まれます。

クライアントはクラスターのメタデータを認識し、サーバーのトポロジの変更に対応するためにメタデータを定期的に更新します。これにより、クライアントはより正確な転送決定を行うことができます。

レコードを盲目的にクラスターに送信し、クラスターがレコードを適切なノードに転送するのを待つのではなく、プロデューサー クライアントは書き込み要求をパーティション マスターに直接転送できます。

同様に、コンシューマ クライアントは、読み取りクエリを発行するときに地理的にコンシューマ クライアントに近いレプリカを使用するなど、レコードを取得する際によりスマートな決定を下すことができます。 (この機能は、Kafka バージョン 2.4.0 以降で利用できます。)

ゼロコピー

典型的な非効率的な方法は、バッファ間でバイトをコピーすることです。 Kafka は、プロデューサー、コンシューマー、サーバーによって共有されるバイナリ メッセージ形式を使用するため、データ ブロックが圧縮されている場合でも、データを変更せずに配信できます。

通信する当事者間のデータ構造の違いをなくすことは重要なステップですが、それだけではデータの重複を回避できるわけではありません。

Kafka は、Java の NIO フレームワーク、具体的には java.nio.channels.FileChannel の transferTo() メソッドを使用して、Linux および UNIX システム上でこの問題を解決します。

このメソッドを使用すると、アプリケーションが転送の仲介役として機能する必要なく、ソース チャネルからシンク チャネルにバイトを転送できます。

NIO がどのように異なるかを理解するには、従来のアプローチで、ソース チャネルをバイト バッファーに読み取り、次にシンク チャネルに書き込むという 2 つの別々の操作を実行することを考えてみましょう。

 File.read (fileDesc, buf, len) ;
 Socket.send(ソケット、バッファ、len);

これは次の図で表すことができます。

この図は単純に見えますが、内部的には、コピー操作にはユーザー モードとカーネル モード間のコンテキスト スイッチが 4 回必要であり、操作が完了するまでにデータを 4 回コピーする必要があります。

次の図は、各ステップでのコンテキスト切り替えの概要を示しています。

詳細な説明:

• 最初の read() メソッドにより、コンテキストがユーザー モードからカーネル モードに切り替わります。ファイルが読み取られ、その内容が DMA (ダイレクト メモリ アクセス) エンジンによってカーネル アドレス空間内のバッファーにコピーされます。これは、コード スニペットで使用されるバッファーとは異なります。
• read() メソッドが返される前に、データはカーネル バッファーからユーザー空間バッファーにコピーされます。この時点で、アプリケーションはファイルの内容を読み取ることができます。
• 後続の send() メソッドはカーネル モードに戻り、ユーザー空間バッファーからカーネル アドレス空間にデータをコピーします。今回は、宛先ソケットに関連付けられた別のバッファーにデータをコピーします。バックグラウンドでは、DMA エンジンが処理を引き継ぎ、カーネル バッファーからプロトコル スタックにデータを非同期的にコピーします。 send() メソッドは、この操作が完了するまで待機せずに戻ります。
• send() メソッド呼び出しが戻り、ユーザー モードに戻ります。

ユーザー モードとカーネル モード間のコンテキスト切り替えは非効率的であり、追加のコピーが必要になりますが、多くの場合、パフォーマンスを向上させることができます。

これは先読みキャッシュとして機能し、非同期で事前読み取りを行い、アプリケーションからのリクエストを事前に実行します。ただし、要求されたデータの量がカーネル バッファーのサイズよりもはるかに大きい場合、カーネル バッファーがパフォーマンスのボトルネックになります。

データを直接コピーするのではなく、すべてのデータが転送されるまで、システムがユーザー モードとカーネル モードを頻繁に切り替えるように強制します。

対照的に、ゼロコピーアプローチは単一の操作で処理されます。前の例のコードは、1 行のコードとして書き直すことができます。

 fileDesc.transferTo(オフセット、長さ、ソケット);

ゼロコピーの詳しい説明は以下のとおりです。

このモデルでは、コンテキストスイッチの回数が 1 回に削減されます。具体的には、transferTo() メソッドは、DMA エンジンを介してブロック デバイスにデータを読み取りバッファに読み込むように指示します。

次に、データは読み取りバッファからソケットバッファにコピーされます。最後に、データは DMA 経由でソケット バッファーから NIC バッファーにコピーされます。

したがって、レプリケーションの数は 4 から 3 に削減され、レプリケーション操作のうち 1 つだけが CPU を必要とします。また、コンテキストスイッチの数も 4 から 2 に削減されました。

これは大きな改善ですが、まだゼロコピークエリではありません。 Linux カーネル 2.4 以上を実行し、収集操作をサポートするネットワーク カード上では、さらなる最適化が可能です。

次の図に示すように:

前の例に従って、transferTo() メソッドを呼び出すと、デバイスは DMA エンジンを介してカーネル バッファーにデータを読み取ります。

ただし、収集操作の場合、読み取りバッファとソケット バッファ間のコピーは行われません。代わりに、NIC には、オフセットと長さとともに、読み取りバッファへのポインタが与えられます。いかなる場合でも、CPU はバッファのコピーに関与しません。

ファイルサイズは数 MB から 1 GB の範囲で、従来のコピーとゼロ コピーを比較すると、ゼロ コピーのパフォーマンスが 2 ～ 3 倍向上するという結果が出ています。

しかし、さらに印象的なのは、Kafka がネイティブ ライブラリや JNI コードを使用せずに、純粋な JVM を使用してこれを実現していることです。

ガベージコレクションの回避

チャネル、バッファ、ページ キャッシュを広範に使用すると、ガベージ コレクターの作業負荷が軽減されるという追加の利点もあります。

たとえば、32 GB の RAM を搭載したマシンで Kafka を実行すると、ページ キャッシュの空き領域が 28～30 GB になり、ガベージ コレクターの手に負えない状態になります。

スループットの差は非常に小さい (数パーセント程度) ですが、適切に調整されたガベージ コレクターのスループットは、特に短命のオブジェクトを処理する場合に非常に高くなる可能性があります。本当のメリットはジッターの減少にあります。

ガベージ コレクションを回避することで、サーバーでガベージ コレクションによるプログラム一時停止が発生する可能性が低くなり、クライアントに影響を与えたり、記録される通信遅延が増加したりする可能性が低くなります。

Kafka の初期の頃と比較すると、ガベージ コレクションを回避することは今ではそれほど問題ではありません。 Shenandoah や ZGC などの最新のガベージ コレクターは、巨大な数テラバイトのヒープに拡張でき、最悪の場合でもガベージ コレクションの一時停止時間を数ミリ秒まで自動的に調整できます。

最近では、オフヒープ キャッシュの代わりにヒープ キャッシュを使用する Java 仮想マシン ベースのアプリケーションが多数見られます。

ストリーム処理における並列処理

ログの I/O 効率はパフォーマンスの重要な側面であり、パフォーマンスへの主な影響は書き込みにあります。 Kafka のトピック構造とコンシューマー エコシステムにおける並列処理の処理は、読み取りパフォーマンスの基礎となります。

この組み合わせにより、全体的に非常に高いエンドツーエンドのメッセージ スループットが実現します。パーティション スキームとコンシューマー グループの操作に同時実行性を組み込むことは、実際には Kafka の負荷分散メカニズムであり、コンシューマー間でパーティションを均等に分散します。

これを従来のメッセージ キューと比較すると、RabbitMQ セットアップでは、複数の同時コンシューマーがラウンドロビン方式でキューからデータを読み取ることができますが、その際にメッセージの順序は失われます。

パーティショニング メカニズムにより、Kafka サーバーの水平拡張が容易になります。各パーティションには専用のリーダーが存在します。したがって、重要なマルチパーティション トピックでは、書き込み操作にサーバー クラスター全体を利用できます。

これは、Kafka と従来のメッセージ キューのもう 1 つの違いです。後者はクラスタリングを活用して可用性を向上させますが、Kafka は負荷分散を通じて可用性、耐久性、スループットを向上させます。

複数のパーティションを持つトピックに公開する場合、プロデューサーはレコードを公開するときにパーティションを指定します。 (おそらく、単一のパーティションのトピックであれば、問題にはならないでしょう)

これは、パーティション インデックスを指定して直接実行することも、ハッシュ値を計算してパーティション インデックスを決定するレコード キーを介して間接的に実行することもできます。同じハッシュ値を持つレコードは同じパーティションを共有します。

トピックに複数のパーティションがある場合、異なるキーを持つレコードが異なるパーティションに表示される可能性があります。

ただし、ハッシュ衝突により、異なるハッシュ値を持つレコードも同じパーティションに格納される可能性があります。それがハッシュの本質です。ハッシュテーブルの仕組みを理解していれば、すべてが自然に理解できるはずです。

レコードの実際の処理は、オプションのコンシューマー グループ内のコンシューマーによって実行されます。 Kafka は、パーティションがコンシューマー グループ内の最大 1 つのコンシューマーに割り当てられることを保証します。 (なぜ「ほとんど」を使用するのでしょうか? すべての消費者がオフラインの場合、消費者は 0 人になります。)

グループ内の最初のコンシューマーがトピックをサブスクライブすると、そのトピックのすべてのパーティションを受信します。 2 番目のコンシューマーがトピックをサブスクライブすると、パーティションの約半分を受信するため、最初のコンシューマーの負荷が軽減されます。

必要に応じてコンシューマーを追加すると (理想的には自動スケーリング メカニズムを使用)、イベント ストリームをパーティション分割していると仮定して、イベント ストリームを並列処理できます。

レコードのスループットは次の 2 つの方法で制御されます。

①トピック分割方式。イベント ストリームの数を最大化するには、トピックをパーティション分割する必要があります。つまり、絶対に必要な場合にのみレコードの順序を指定します。

2 つのレコードに相関関係がない場合は、それらを同じパーティションにバインドしないでください。これは、Kafka がパーティション マッピングの基礎としてレコード キーのハッシュを使用するため、異なるキーを使用することを意味します。

②グループ内の消費者の数。受信レコードの負荷を分散するために、パーティションの数までコンシューマーの数を増やすことができます。 (コンシューマーをさらに追加することはできますが、パーティションごとにアクティブなコンシューマーは最大 1 つまでで、残りはアイドル状態になります)

スレッド プールを提供でき、コンシューマーによって実行されるワークロードに応じて、コンシューマーはプロセスまたはスレッドになる可能性があることに注意してください。

Kafka がなぜ高速なのか、どのように高速化するのか、そしてそれが自分に適しているかどうか疑問に思ったことがあるなら、これで答えが見つかったと思います。

明確に言えば、Kafka は最速のメッセージング ミドルウェアではなく、スループットも最高ではありません。より高いスループットを提供できる他のプラットフォームも存在します。ソフトウェア ベースのものもあれば、ハードウェア ベースのものもあります。

高いスループットと低いレイテンシの両方を同時に実現することは困難です。 Apache Pulsar[1]は、スケーラブルで、スループットとレイテンシのプロファイルが優れており、連続性と耐久性を提供する有望なテクノロジーです。

Kafka を採用する理由は、完全なエコシステムとして全体的に比類のないものであることです。

豊かで成熟した環境を提供しながら優れたパフォーマンスを発揮し、Kafka は今も羨ましいほどのペースで成長を続けています。

Kafka の設計者とメンテナーは、パフォーマンスを重視したソリューションの設計に素晴らしい仕事をしました。デザイン要素のほとんどは、後付け、または補完的なもののようには感じられません。

クライアントへのワークロードのオフロードから、サーバー側のログの永続化、バッチ処理、圧縮、ゼロコピー I/O、並列ストリーム処理まで、Kafka は、商用またはオープンソースの他のメッセージング ミドルウェア ベンダーに挑戦します。

最も印象的なのは、耐久性、レコードの順序、および少なくとも 1 回の配信セマンティクスを犠牲にすることなくこれを実現することです。

Kafka は最もシンプルなメッセージング ミドルウェア プラットフォームではなく、まだ改善の余地が大きく残っています。高性能なイベント駆動型システムを設計および構築する前に、全体順序と部分順序、トピック、パーティション、コンシューマー、コンシューマー グループの概念を習得する必要があります。

知識曲線は急峻ですが、時間をかけて学ぶ価値はあります。ことわざの「レッドピル」をご存知の場合は、「Kafka と Kafdrop によるイベントストリーミング入門[2]」をお読みください。

関連リンク:

• https://pulsar.apache.org/
• https://medium.com/swlh/introduction-to-event-streaming-with-kafka-and-kafdrop-22afdb4b380a