Kafka トランザクションフローの基礎から実践まで

イベント ソーシング、結果整合性、マイクロサービス、CQRS など、現代の開発者にとって馴染みのある概念が増えています。きめ細かいサービスアセンブリから複雑なビジネス中心のアプリケーションアーキテクチャまで、最も重要な部分はミドルウェアに基づくビジネス分離です。この記事では、ミドルウェアの基本的な構成要素であるトランザクション フローを紹介します。そのリーダーは、事実上のトランザクション ストリーミング プラットフォーム標準である Apache Kafka です。 Kafka の Web インターフェース ツールである Kafdrop も紹介されます。

概要

トランザクション ストリーミング プラットフォームは、従来のメッセージ キューやトピックと同様に、より広範なメッセージ指向ミドルウェア (MoM) のクラスに属しますが、ログ構造の不変性により、より強力な時間保証と大幅なパフォーマンスの向上を実現します。つまり、トランザクション ストリームの書き込み操作は順次追加に制限されるため、より効率的です。

従来のメッセージ キュー (MQ) 内のメッセージは、任意に順序付けられる傾向があり、通常は互いに独立していますが、ストリーム内のトランザクション (またはレコード) は、時系列または因果関係で順序付けられる傾向があります。また、トランザクション ストリームはレコードを保持しますが、MQ はメッセージを読み取ったらそれを破棄します。したがって、トランザクション ストリーミングは、イベント ソーシング、結果整合性、CQRS などを含むイベント駆動型アーキテクチャに適している傾向があります (もちろん、FIFO メッセージ キューも含まれますが、FIFO キューと成熟したトランザクション ストリーミング プラットフォームの違いは非常に大きく、順序付けに限定されません)。

トランザクション ストリーミング プラットフォームは、MoM 分野では比較的新しいパラダイムです。何百もの MQ スタイルのメッセージ ブローカーと比較すると、利用できる主流のブローカーはほんのわずかです。 AMQP、MQTT、XMPP、JMS などの確立された標準と比較すると、トランザクション ストリーミングの分野では同等の標準は存在しません。

トランザクション ストリーミング プラットフォームは、現在も研究と実験が行われている活発な分野です。ただし、トランザクション フロー プラットフォームは単なる商用製品ではなく、複雑な学術的問題でもありません。メッセージングやトランザクションのシナリオで幅広く使用でき、メッセージ キューの従来の使用シナリオを日常的に置き換えるために使用できます。

アーキテクチャの概要

次の図は、Kafka コンポーネント アーキテクチャの概要を簡単に示しています。スペースの制限により、Kafka が内部でどのように動作するかについては詳しく説明しません。

Kafka コンポーネント

Kafka は、次の主要コンポーネントで構成される分散システムです。

ブローカー ノード: バッチ I/O 操作とクラスター内の継続的な永続性を担当します。ブローカーは、クラスターによってホストされるトピック パーティションを含むログ ファイルを追加します。パーティションを複数のブローカーに複製することで、水平方向のスケーラビリティと耐久性の向上を実現できます。これらの複製されたパーティションはレプリカと呼ばれます。制御ノード (コントローラー) としてプロキシ ノードが 1 つあり、他のレプリカはそのプロキシ ノード (フォロワー) によって管理されます。プロキシ ノードはクラスタ コントローラとして選出され、パーティション状態の内部管理と、特定のパーティションのリーダーとフォロワーの役割の調停を担当します。

ZooKeeper ノード: 舞台裏では、Kafka はクラスター内の全体的なコントローラーの状態を管理する方法を必要とします。何らかの理由でコントローラーが終了した場合、残りのエージェント セットから別のコントローラーを選択するプロトコルがあります。 ZooKeeper は、コントローラーの選択、ハートビートなどの実際のメカニズムを主に実装します。また、ZooKeeper は、クラスターのメタデータ、リーダーとフォロワーのステータス、クォータ、ユーザー情報、ACL、およびその他のハウスキーピング項目を維持する、一種の構成リポジトリとしても機能します。基礎となる選挙およびコンセンサス プロトコルのため、ZooKeeper ノードの数は奇数でなければなりません。

プロデューサー: Kafka トピックにメッセージを公開する役割を担うクライアント アプリケーション。 Kafka のログ構造化の性質と、複数のコンシューマー エコシステム全体でトピックを共有できる機能により、基礎となるログ ファイル内のデータを変更できるのはプロデューサーのみです。実際の I/O は、プロデューサー クライアントに代わってプロキシ ノードによって実行されます。任意の数のプロデューサーが同じ Kafka トピックにメッセージを公開し、レコードを保存するパーティションを選択できます。

コンシューマー: トピックからメッセージを読み取るクライアント アプリケーション。任意の数のコンシューマーが同じトピックから読み取ることができます。ただし、コンシューマーの構成とグループ化に応じて、コンシューマー間でのレコードの配布を管理するルールが存在します。

パーティション、レコード、オフセット、トピック

パーティションは完全に順序付けられたレコードのシーケンスであり、各パーティションは Kafka の基礎となる追加ログに対応します。各レコードには、64 ビットの整数オフセットとミリ秒のタイムスタンプからなる ID があります。キーと値が含まれる場合があります。どちらもバイト配列であり、どちらもオプションです。 「完全に順序付けられた」という用語は、特定のプロデューサーについて、レコードがアプリケーションによって発行された順序で書き込まれることを意味します。レコード P が Q より前に公開された場合、パーティション内で P は Q より前になります。 (P と Q がパーティションを共有していると仮定します。) さらに、すべてのコンシューマーは同じ順序でそれらを読み取ります。すべての可能性のあるコンシューマーについて、P は常に Q の前に読み取られます。ほとんどのユースケースでは、この順序の保証が重要です。通常、公開された記録は実際の取引に対応しており、それらの取引のタイムラインを保存することが不可欠な場合がよくあります。

レコードのオフセットは、パーティション内のレコードの一意の識別子です。オフセットは、スパース アドレス空間内で厳密に単調に増加する整数であり、各レコード オフセットは常に前のレコード オフセットよりも大きくなり、隣接するオフセット間には可変のギャップが存在する場合があります。圧縮が有効になっている場合やトランザクションの結果として必然的にギャップが生じるため、オフセットは連続していない可能性があります。

アプリケーションはオフセットを文字通りに解釈しようとしたり、次のオフセットが何であるかを推測したりすべきではありません。ただし、レコードの相対的な順序はオフセットから推測できるため、レコードはオフセットによってソートされます。

次の図は、内部パーティションの構造を示しています。

最初のオフセット (ロー ウォーター マークとも呼ばれます) は、コンシューマーに表示される最初のメッセージです。 Kafka の保持期間により、これが必ずしも最初に公開されるメッセージとは限りません。レコードは、時間やパーティション サイズに基づいて整理できます。これが起こると、低水準点が後退したように見え、低水準点より古いレコードは切り捨てられます。

トピックはパーティションの論理グループです。トピックには 1 つ以上のパーティションを含めることができますが、パーティションには 1 つのトピックまたはトピックの一部のみを含めることができます。トピックは Kafka のバックボーンであり、並列処理と負荷分散を可能にします。先ほど、パーティションは全体の順序を示すと述べました。トピック内のパーティションは互いに独立しているため、トピックは部分順序を持つと言われます。つまり、一部のレコードは互いに相対的に並べ替え可能ですが、他の一部のレコードとは相対的に並べ替えることができません。全順序と部分順序の概念は、いくぶん学術的に聞こえるかもしれませんが、パフォーマンスの高いトランザクション ストリーミング パイプラインを構築する上で非常に重要です。これにより、必要に応じて順序を維持しながら、可能な場合はレコードを並行して処理できるようになります。後ほど、レコード順序、コンシューマーの並列処理、トピック サイズの概念について説明します。

例: メッセージの公開

実践こそが真実をテストする唯一の基準です。私は理論を実践し、例を通して概念を説明します。 Kafka 用と Kafdrop 用の 2 つの Docker コンテナを起動します。コンテナを有効にするには、Docker Compose を使用します。

選択したディレクトリに次の内容の docker-compose.yaml ファイルを作成します。

便宜上、Kafka と ZooKeeper を 1 つのイメージにきちんとパッケージ化した obsidiandynamics/kafka イメージを使用します。次に、docker-compose up 経由でコンテナを起動します。起動が成功したら、ブラウザで localhost:9000 にアクセスして、Kafdrop ログイン インターフェイスを確認できます。

この例は単一ブローカー クラスターであり、まだトピックはありません。 Kafka のコマンドライン ツールを使用してトピックを作成し、メッセージを公開することができます。 docker exec ツールを使用して kafka コンテナを操作し、組み込みの CLI ツールを簡単に呼び出すことができます。

docker exec -it kafka-kafdrop_kafka_1 bash

上記のコマンドを実行すると、コンテナのシェル コマンド ライン インターフェイスにアクセスできます。ツールは /opt/kafka/bin ディレクトリにあります。ディレクトリにcdします:

3 つのパーティションを持つ streams-intro というトピックを作成します。

Kafdrop インターフェースに戻ると、新しいマスターによって作成されたトピックがリストに表示されます。

次に、kafka-console-producer ツールを使用してメッセージを公開します。

注: kafka-topics は --bootstrap-server パラメータを使用して Kafka ブローカー リストを構成しますが、kafka-console-producer は --broker-list を使用します。

レコードは改行文字で区切られます。キーと値の部分は、key.separator プロパティで示されるように、コロンによって区切られます。この例では、次のように入力できます。

完了したら、CTRL+D を押してメッセージの投稿を完了します。次に、Kafdrop に戻り、streams-intro トピックをクリックします。トピックの概要と基礎となるパーティションの詳細な内訳が表示されます。

3 つのパーティションを持つトピックを作成しました。次に、2 つの一意のキー foo と bar を使用して 5 つのレコードを公開しました。 Kafka はキーを使用してレコードをパーティションにマッピングし、同じキーを持つすべてのレコードが常に同じパーティションに表示されるようにします。これは、発行者がレコードの正確な順序を指定できるため、便利かつ重要です。キーハッシュとパーティション割り当てについては、後ほど詳しく説明します。

パーティション テーブルを見ると、パーティション #0 の最初のオフセットと最後のオフセットはそれぞれ 0 と 2 です。パーティション #2 には 0 と 3 の値が設定されますが、パーティション #1 は空白です。 Kafdrop Web UI で #0 をクリックすると、テーマ ビューアーが表示されます。

bar キーの下に公開された 2 つのレコードが表示されます。これらは foo レコードとはまったく関係がないことに注意してください。

消費者と消費者団体

上記の例では、プロデューサーがメッセージを公開し、レコードをストリームに送信することがわかりました。レコードは整理されたパーティションにまとめられます。 Kafka のパブリッシュ/サブスクライブ トポロジは柔軟な多対多モデルに従うため、任意の数のプロデューサーとコンシューマーが同時にストリームと対話できます。実際のソリューションに応じて、フロー トポロジは 1 対多または多対 1 になることもあります。次に、これらのレコードをどのように使用するかについて説明します。

コンシューマーは、クライアント ライブラリを介して Kafka クラスターに接続するプロセスまたはスレッドです。消費者は通常（必ずしもそうとは限りませんが）、全体的な消費者グループのメンバーです。グループは、group.id 属性によって指定されます。コンシューマ グループは、実際には Kafka の負荷分散メカニズムであり、グループ内のコンシューマ インスタンス間でパーティションをほぼ均等に分散する役割を担っています。グループ内の最初のコンシューマーがトピックをサブスクライブすると、トピック内のすべてのパーティションを受信します。その後、2 番目のコンシューマーが参加すると、パーティションの約半分が取得され、最初のコンシューマーの負担が軽減されます。コンシューマーが（切断またはタイムアウトにより）離れると、プロセスが逆転し、残りのコンシューマーがより多くのパーティションを使用できるようになります。

したがって、コンシューマーはトピックからレコードを消費し、Kafka およびそれが属する他のコンシューマーによって割り当てられたパーティションからシェアを取得します。負荷分散に関しては、これはかなり簡単なはずです。しかし、ここで重要な点は、レコードを使用する行為によってレコードが削除されるわけではないということです。これは、特に消費行為が消費に結びついている場合、一見矛盾しているように見えるかもしれません。 (どちらかといえば、消費者は「読者」と呼ばれるべきです。) 単純な事実は、消費者がトピックとそのパーティションにまったく影響を与えないということです。トピックは追加専用であり、レコードはプロデューサーまたは Kafka 自体 (圧縮またはパージの一部として) によってのみ追加できます。コンシューマーの読み取り専用操作は「安価」であるため、多くの人がクラスターの負荷を増やすことなくログを追跡できます。これは、トランザクション ストリーミングと従来のメッセージ キューのもう 1 つの重要な違いです。

コンシューマーは、パーティション内の次のレコードを指すオフセットを内部的に維持し、連続した読み取りごとにオフセットを増やします。コンシューマーが最初にトピックをサブスクライブするときに、トピックの先頭から開始するか、末尾から開始するかを選択できます。この動作は、auto.offset.reset プロパティを latest、earliest、または none に設定することで制御できます。後者の場合、コンシューマー グループに以前のオフセットが存在しない場合は、例外がトリガーされます。

コンシューマーはオフセット状態ベクトルをローカルに保持します。異なる消費者グループの消費者が互いに干渉しないため、多くの人が同時に同じトピックを読むことが可能です。コンシューマーは独自のオフセットに従ってメッセージを読み取ります。処理が遅い、またはバックログが発生している消費者は、グループ内の他の消費者に影響を与えません。

この概念を説明するために、2 つのパーティションを含むトピックのシナリオを考えてみましょう。 2 つのコンシューマー グループ (A と B) がトピックをサブスクライブしました。各グループには 3 つのインスタンスがあり、ユーザーの名前は A1、A2、A3、B1、B2、B3 です。次の図は、2 つのグループがトピックを共有する方法と、コンシューマーが互いに独立してレコードを参照する方法を示しています。

上の写真をよく見ると、何かが欠けていることに気がつくでしょう。消費者 A3 と B1 は上記の図には存在しません。これは、Kafka では、パーティションをそのコンシューマー グループ内の 1 つのコンシューマーにのみ割り当てることができることが保証されているためです。各グループには 3 つのコンシューマーがありますが、パーティションは 2 つしかないため、1 つのコンシューマーはアイドル状態のままになり、グループ内の別のコンシューマーが退出するのを待ちます。このように、コンシューマ グループは、負荷分散メカニズムであるだけでなく、特にレコードを 1 つのスレッドまたは特定の時間にのみ処理する必要がある状況で、安全性を犠牲にすることなく高性能なパイプラインを構築するためのフェンスのような排他性制御でもあります。

可用性を確保するためにコンシューマー グループも使用されます。トピックからレコードを定期的に取得することで、コンシューマーはクラスターが「正常」であるというフィードバックをクラスターに暗黙的に提供し、パーティション割り当てのリースを延長することができます。ただし、コンシューマーが許容期間内に再度読み取りに失敗した場合、そのコンシューマーは不良とみなされ、そのパーティションはグループ内の残りの「正常な」コンシューマーの 1 つに再割り当てされます。この期限は、コンシューマー クライアント側の max.poll.interval.ms プロパティによって制御され、デフォルトでは 5 分に設定されています。

交通システムに例えると、主題は高速道路のようなもので、ゾーンは車線に相当します。レコードは車に相当し、その乗客はレコードの値に対応します。車線を維持している限り、複数の車両が同じ高速道路を安全に走行できます。同じ車線を共有する車は順番に走行し、隊列を形成します。ここで、各車線が、交通をある場所に迂回させるランプにつながっていると仮定します。 1 つのランプがバックアップされても、他のランプはスムーズに流れる可能性があります。

Kafka はこのメカニズムを使用してエンドツーエンドのスループットを確保し、1 秒あたり数百万レコードの QPS を簡単に達成します。トピックを作成するときに、パーティション数、チャネル数を選択できます。パーティションはコンシューマー グループ内のコンシューマー間でほぼ均等に分割され、同時に 2 つ (またはそれ以上) のコンシューマーに割り当てられるパーティションがないことが保証されます。

注: トピックは作成後、パーティションの数を増やすことでサイズを変更できます。ただし、トピックを再作成せずにパーティションの数を減らす方法はありません。

レコードは、イベント、メッセージ、コマンド、またはストリーミング可能なその他のものに対応します。レコードを分割する正確な方法はプロデューサーによって決定されます。プロデューサーはレコードを公開するときにパーティション インデックスを明示的に割り当てることができますが、この方法はほとんど使用されません。より一般的なアプローチは、前の例のように、レコードにキーを割り当てることです。キーは Kafka にとって完全に不透明です。つまり、Kafka はキーの内容を解釈せず、バイト配列として扱います。これらのバイトは、一貫したハッシュ手法を使用してハッシュされ、パーティション インデックスが導出されます。

同じハッシュを共有するレコードは、同じパーティションを占有することが保証されます。トピックに複数のパーティションがある場合、異なるキーを持つレコードは異なるパーティションに配置される場合があります。ただし、ハッシュ キーの衝突により、異なるハッシュ値を持つレコードが同じパーティションに配置されてしまう可能性もあります。

関連するレコードが同じパーティションに配置され、順序が保持されている限り、プロデューサーはレコードがどの特定のパーティションにマップされるかを気にする必要はありません。同様に、コンシューマー ペアは、レコードが公開されたときと同じ順序で受信され、パーティションの割り当てがグループ内の他のコンシューマーと重複しない限り、どのパーティションに割り当てられるかを気にする必要はありません。

事例: 取引プラットフォーム

上場株式の特定の価格パターンを探し、特定のパターンが特定されたら取引シグナルを発行すると仮定します。在庫は大量にあるため、当然のことながら、それらを並行して処理したいという要望があります。ただし、特定の株価シンボルの時系列は、単一のコンシューマーで順番に処理する必要があります。

Kafka を使用すると、このユースケースやその他の同様のユースケースの実装がほぼ簡単になります。生の価格データを保存するために使用される「price」という 2 つのトピックを作成します。注文トピックは、によって生成された注文を保存するために使用されます。操作を完全に並列化できるように、さらにパーティションを作成できます。

株価シンボルをキーとして使用して、価格トピックの各価格のレコードを公開できます。 Kafka の自動パーティション割り当てにより、各株価シンボルがグループ内の 1 つのコンシューマーによって処理されるようになります。コンシューマー インスタンスは、処理負荷に合わせて自由にスケールアップおよびスケールダウンできます。コンシューマー グループには意味のある名前を付ける必要があります。理想的には、コンシューマー アプリケーションの目的を反映します。たとえば、trading-strategy.abc は「ABC」と呼ばれる仮想取引戦略です。

消費者が価格パターンを決定すると、注文トピックに別のメッセージ、つまり注文リクエストを公開できます。注文を読み取り、ブローカーに転送する責任を負う別のコンシューマー グループを Order Execution と呼びます。

この単純な例では、他のボトルネックがないことを前提として、完全にイベント駆動型で高度にスケーラブルなエンドツーエンドのトランザクション パイプラインを作成しました。必要な負荷の増加に対応するために、さまざまな段階で処理ノードを動的に追加できます。

共通のデータ ソースに基づいて、複数の取引戦略を同時に実行する必要があるとします。さらに、取引戦略はさまざまなチームによって開発されます。目的は、これらの実装を可能な限り分離し、チームが自律的に操作し、異なるプログラミング言語とツールチェーンを使用して独自のペースで開発および展開できるようにすることです。

Kafka の柔軟な多対多の pub-sub アーキテクチャは、状態の消費とブロードキャスト セマンティクスを組み合わせています。 Kafka では、異なるコンシューマー グループを使用することで、さまざまなアプリケーションが入力トピックを共有し、独自のペースでイベントを処理できるようになります。 2 番目の取引戦略には、専用のコンシューマー グループ trading-strategy.xyz が必要です。このグループは、特定のビジネス ロジックを汎用価格設定ストリームに適用し、結果として得られる注文を同じ注文トピックに公開します。このように、Kafka を使用すると、簡単に再利用および構成できる個別の要素からモジュール式のイベント処理パイプラインを構築できます。

要約する

トランザクション ストリーミング プラットフォームは、モジュール式で疎結合のイベント駆動型アプリケーションを構築するための効率的なビルディング ブロックです。トランザクション ストリーミングの世界では、Kafka は柔軟性と高パフォーマンスを兼ね備えた成功したオープン ソース ソリューションとしての地位を確立しています。同時実行性と並列性は Kafka のアーキテクチャの中核であり、スケーラブルな消費者エコシステム全体で負荷分散できる、部分的に順序付けられたトランザクション ストリームを形成します。コンシューマーとその周囲のグループを単純に再構成するだけで、イベントの配布と処理のセマンティクスが大きく変わる可能性があります。

もちろん、カフカには欠点がないわけではありません。 Kafka ツールは標準以下であると言えます。 Kafka 実践者のほとんどは、既製の CLI ユーティリティを放棄し、Kafdrop、Kafkacat などの他のオープンソース ツールや、Kafka Tool などのサードパーティの商用製品を採用してきました。 Kafka にはさまざまな設定オプションとデフォルト値があり、初心者には適していません。全体として、Kafka は複雑なシステムの構築方法におけるパラダイムシフトを表しており、その利点は明らかです。