著書『Kafka Streams in Action. リアルタイム作業のためのアプリケーションとマイクロサービス」

ハブロ住民の皆さん、こんにちは！ この本は、スレッド処理を理解したいすべての開発者に適しています。 分散プログラミングを理解すると、Kafka と Kafka Streams をより深く理解できるようになります。 Kafka フレームワーク自体を知っていると良いですが、必須ではありません。必要なことはすべて説明します。 経験豊富な Kafka 開発者も初心者も、この本で Kafka Streams ライブラリを使用して興味深いストリーム処理アプリケーションを作成する方法を学びます。 シリアル化などの概念にすでに慣れている中級および上級の Java 開発者は、スキルを応用して Kafka Streams アプリケーションを作成する方法を学びます。 この本のソース コードは Java 8 で書かれており、Java 8 ラムダ式構文を多用しているため、(別のプログラミング言語であっても) ラムダ関数の操作方法を知っておくと役に立ちます。

抜粋。 5.3. 集計およびウィンドウ操作

このセクションでは、Kafka Streams の最も有望な部分を検討していきます。 これまで、Kafka ストリームの次の側面について説明してきました。

• 処理トポロジを作成する。
• ストリーミング アプリケーションでの状態の使用。
• データストリーム接続を実行します。
• イベント ストリーム (KStream) と更新ストリーム (KTable) の違い。

次の例では、これらすべての要素をまとめます。 また、ストリーミング アプリケーションのもう XNUMX つの優れた機能であるウィンドウ処理についても学習します。 最初の例は単純な集計です。

5.3.1. 業種別株式売上高の集計

集約とグループ化は、ストリーミング データを操作する場合に重要なツールです。 個々の記録を受け取った際の検査だけでは不十分なことがよくあります。 データから追加情報を抽出するには、データをグループ化して結合する必要があります。

この例では、いくつかの業界の企業の株式の販売量を追跡する必要があるデイトレーダーのコスチュームを着ます。 具体的には、各業界で最大のシェア売上高を誇る XNUMX 社に興味があります。

このような集計では、データを目的の形式に変換するために次のいくつかの手順が必要になります (一般的に言えば)。

1. 生の株取引情報を公開するトピックベースのソースを作成します。 StockTransaction 型のオブジェクトを ShareVolume 型のオブジェクトにマップする必要があります。 重要なのは、StockTransaction オブジェクトには販売メタデータが含まれていますが、必要なのは販売されている株式数に関するデータだけであるということです。
2. ShareVolume データを銘柄記号ごとにグループ化します。 シンボルごとにグループ化すると、このデータを株式販売量の小計に折りたたむことができます。 KStream.groupBy メソッドは KGroupedStream 型のインスタンスを返すことに注意してください。 さらに、KGroupedStream.reduce メソッドを呼び出すことで、KTable インスタンスを取得できます。

KGroupedStream インターフェイスとは

KStream.groupBy メソッドと KStream.groupByKey メソッドは、KGroupedStream のインスタンスを返します。 KGroupedStream は、キーによってグループ化された後のイベントのストリームの中間表現です。 これは、直接作業することを目的としたものではありません。 代わりに、KGroupedStream が集計操作に使用され、その結果は常に KTable になります。 また、集計操作の結果は KTable であり、状態ストアを使用するため、結果としてのすべての更新がパイプラインのさらに下流に送信されるわけではない可能性があります。

KTable.groupBy メソッドは、同様の KGroupedTable (キーによって再グループ化された、更新ストリームの中間表現) を返します。

少し休憩して図を見てみましょう。 5.9 は、私たちが達成したことを示しています。 このトポロジはすでによく知られているはずです。

次に、このトポロジーのコードを見てみましょう (ファイル src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java にあります) (リスト 5.2)。

指定されたコードは、その簡潔さと、数行で実行される大量のアクションによって区別されます。 builder.stream メソッドの最初のパラメーターに新しい点があることに気づくかもしれません。列挙型 AutoOffsetReset.EARLIEST (LATEST もあります) の値は、Consumed.withOffsetResetPolicy メソッドを使用して設定されます。 この列挙型は、各 KStream または KTable のオフセット リセット戦略を指定するために使用でき、構成のオフセット リセット オプションよりも優先されます。

GroupByKey と GroupBy

KStream インターフェイスには、レコードをグループ化するための XNUMX つのメソッド、GroupByKey と GroupBy があります。 どちらも KGroupedTable を返すので、両者の違いは何なのか、いつどちらを使用すればよいのか疑問に思われるかもしれません。

GroupByKey メソッドは、KStream 内のキーがすでに空でない場合に使用されます。 そして最も重要なことは、「再パーティション化が必要」フラグが設定されていないことです。

GroupBy メソッドは、グループ化キーが変更されたことを前提としているため、再分割フラグが true に設定されています。 GroupBy メソッドの後に結合や集計などを実行すると、自動的に再パーティション化されます。
概要: 可能な限り、GroupBy ではなく GroupByKey を使用する必要があります。

mapValues メソッドと groupBy メソッドが何を行うかは明らかなので、sum() メソッド (src/main/java/bbejeck/model/ShareVolume.java にあります) を見てみましょう (リスト 5.3)。

ShareVolume.sum メソッドは在庫販売量の累計を返し、一連の計算全体の結果が KTable オブジェクトになります。 。 これで、KTable が果たす役割が理解できました。 ShareVolume オブジェクトが到着すると、対応する KTable オブジェクトに最新の更新が保存されます。 すべての更新は前の shareVolumeKTable に反映されますが、すべてがさらに送信されるわけではないことに注意することが重要です。

次に、この KTable を使用して (取引株式数ごとに) 集計し、各業界で最も取引高の多い XNUMX 社を導き出します。 この場合のアクションは、最初の集計の場合と同様になります。

1. 別の groupBy 操作を実行して、個々の ShareVolume オブジェクトを業界ごとにグループ化します。
2. ShareVolume オブジェクトの要約を開始します。 今回の集約オブジェクトは固定サイズの優先キューです。 この固定サイズのキューでは、売却された株式の量が最も多い XNUMX 社だけが保持されます。
3. 前の段落のキューを文字列値にマップし、業界ごとに最も取引されている上位 XNUMX つの銘柄を数値で返します。
4. 結果を文字列形式でトピックに書き込みます。

図では、 図 5.10 にデータ フロー トポロジ グラフを示します。 ご覧のとおり、XNUMX 回目の処理は非常に単純です。

この 5 回目の処理の構造を明確に理解したので、ソース コードに移ります (ファイル src/main/java/bbejeck/chapter_5.4/AggregationsAndReducingExample.java にあります) (リスト XNUMX)。 。

このイニシャライザには、fixedQueue 変数が含まれています。 これは、取引された株式の降順で上位 N 件の結果を追跡するために使用される java.util.TreeSet のアダプターであるカスタム オブジェクトです。

groupBy 呼び出しと mapValues 呼び出しについてはすでに説明したので、それらについては説明しません (KTable.print メソッドは非推奨であるため、KTable.toStream メソッドを呼び出しています)。 ただし、aggregate() の KTable バージョンをまだ見ていないので、少し時間をかけて説明します。

覚えているとおり、KTable が異なるのは、同じキーを持つレコードが更新と見なされることです。 KTable は古いエントリを新しいエントリに置き換えます。 集計も同様の方法で行われ、同じキーを持つ最新のレコードが集計されます。 レコードが到着すると、加算器 (集約メソッド呼び出しの XNUMX 番目のパラメーター) を使用して、FixedSizePriorityQueue クラスのインスタンスに追加されますが、同じキーを持つ別のレコードが既に存在する場合は、減算器 (集約メソッド呼び出しの XNUMX 番目のパラメーター) を使用して古いレコードが削除されます。集約メソッド呼び出し)。

これは、アグリゲーターであるFixedSizePriorityQueueが、すべての値をXNUMXつのキーで集計するのではなく、最も取引されているN種類の株の数量の移動合計を保存することを意味します。 受信した各エントリには、これまでに販売された株式の総数が含まれています。 KTable は、各更新のローリング集計を必要とせずに、現在どの企業の株が最も取引されているかに関する情報を提供します。

私たちは XNUMX つの重要なことを行うことを学びました。

• KTable 内の値を共通キーでグループ化します。
• これらのグループ化された値に対してロールアップや集計などの便利な操作を実行します。

これらの操作の実行方法を知ることは、Kafka Streams アプリケーションを介して移動するデータの意味を理解し、データがどのような情報を運ぶかを理解するために重要です。

また、本書の前半で説明した重要な概念のいくつかもまとめました。 第 4 章では、ストリーミング アプリケーションにとってフォールト トレラントなローカル状態がいかに重要であるかについて説明しました。 この章の最初の例では、ローカル状態がなぜ非常に重要であるかを説明しました。ローカル状態により、すでに見た情報を追跡できるようになります。 ローカル アクセスによりネットワーク遅延が回避され、アプリケーションのパフォーマンスが向上し、エラー耐性が向上します。

ロールアップまたは集計操作を実行するときは、状態ストアの名前を指定する必要があります。 ロールアップおよび集計操作は KTable インスタンスを返し、KTable は状態ストレージを使用して古い結果を新しい結果に置き換えます。 これまで見てきたように、すべての更新がパイプラインに送信されるわけではありません。集計操作は概要情報を生成するように設計されているため、これは重要です。 ローカル状態を適用しない場合、KTable はすべての集計結果とロールアップ結果を転送します。

次に、特定の期間内での集計などの操作、いわゆるウィンドウ操作の実行について見ていきます。

5.3.2. ウィンドウ操作

前のセクションでは、スライディング畳み込みと集計について紹介しました。 このアプリケーションは、株式販売量の継続的なロールアップを実行し、その後、取引所で最も取引されている XNUMX つの株式を集計しました。

場合によっては、このような継続的な集計と結果のロールアップが必要になることがあります。 また、特定の期間にのみ操作を実行する必要がある場合もあります。 たとえば、過去 10 分間に特定の企業の株式に対して行われた為替取引の数を計算します。 または、過去 15 分間に新しい広告バナーをクリックしたユーザーの数。 アプリケーションはこのような操作を複数回実行する場合がありますが、その結果は指定された期間 (タイム ウィンドウ) にのみ適用されます。

買い手ごとの為替取引のカウント

次の例では、複数のトレーダー (大規模な組織または賢明な個人投資家) にわたる株式取引を追跡します。

この追跡には XNUMX つの理由が考えられます。 その XNUMX つは、市場リーダーが何を売買しているかを知る必要性です。 これらの大手企業や洗練された投資家がチャンスを見出したのであれば、彼らの戦略に従うのは理にかなっています。 XNUMX 番目の理由は、違法なインサイダー取引の可能性のある兆候を見つけたいという欲求です。 これを行うには、大規模な売上の急増と重要なプレス リリースの相関関係を分析する必要があります。

このような追跡は次の手順で構成されます。

• 株式取引トピックから読み取るためのストリームを作成します。
• 受信レコードを購入者 ID と銘柄記号ごとにグループ化します。 groupBy メソッドを呼び出すと、KGroupedStream クラスのインスタンスが返されます。
• KGroupedStream.windowedBy メソッドは、時間ウィンドウに限定されたデータ ストリームを返します。これにより、ウィンドウ集計が可能になります。 ウィンドウの種類に応じて、TimeWindowedKStream または SessionWindowedKStream が返されます。
• 集計操作のトランザクション数。 ウィンドウ化されたデータ フローによって、このカウントで特定のレコードが考慮されるかどうかが決まります。
• 開発中に結果をトピックに書き込むか、コンソールに出力します。

このアプリケーションのトポロジは単純ですが、明確な図があると役立ちます。 図を見てみましょう。 5.11。

次に、ウィンドウ操作の機能と対応するコードを見ていきます。

窓の種類

Kafka Streams には XNUMX 種類のウィンドウがあります。

• セッション的;
• 「タンブリング」（タンブリング）。
• スライディング/ホッピング。

どちらを選択するかは、ビジネス要件によって異なります。 タンブリング ウィンドウとジャンピング ウィンドウは時間制限がありますが、セッション ウィンドウはユーザーのアクティビティによって制限されます。セッションの継続時間はユーザーがどれだけアクティブであるかによってのみ決まります。 覚えておくべき主な点は、すべてのウィンドウ タイプはシステム時間ではなく、エントリの日付/タイム スタンプに基づいているということです。

次に、各ウィンドウ タイプでトポロジを実装します。 完全なコードは最初の例でのみ示されており、他のタイプのウィンドウの場合は、ウィンドウ操作のタイプ以外は何も変わりません。

セッションウィンドウ

セッション ウィンドウは、他の種類のウィンドウとは大きく異なります。 これらは、時間ではなく、ユーザーのアクティビティ (または追跡するエンティティのアクティビティ) によって制限されます。 セッション ウィンドウは、非アクティブな期間によって区切られます。

図 5.12 にセッション ウィンドウの概念を示します。 小さいセッションは、その左側のセッションとマージされます。 右側のセッションは、長期間非アクティブな状態が続くため、分離されます。 セッション ウィンドウはユーザー アクティビティに基づいていますが、エントリの日付/時刻スタンプを使用して、エントリがどのセッションに属しているかを判断します。

セッションウィンドウを使用して株式取引を追跡する

セッション ウィンドウを使用して、為替トランザクションに関する情報を取得してみましょう。 セッションウィンドウの実装をリスト 5.5 に示します (これは src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java にあります)。

このトポロジのほとんどの操作はすでに見てきたので、ここで再度確認する必要はありません。 ただし、ここにはいくつかの新しい要素もありますので、それについてはこれから説明します。

通常、groupBy 操作は、ある種の集計操作 (集計、ロールアップ、またはカウント) を実行します。 現在までの合計を使用した累積集計、または指定した時間枠内のレコードを考慮するウィンドウ集計のいずれかを実行できます。

リスト 5.5 のコードは、セッション ウィンドウ内のトランザクションの数をカウントします。 図では、 5.13 これらのアクションは段階的に分析されます。

windowedBy(SessionWindows.with(twentySeconds).until(fifteenMinutes)) を呼び出すことで、非アクティブ期間が 20 秒、持続期間が 15 分のセッション ウィンドウを作成します。 20 秒のアイドル間隔は、アプリケーションが現在のセッションの終了または開始から 20 秒以内に到着したエントリを現在の (アクティブな) セッションに含めることを意味します。

次に、セッション ウィンドウでどの集計操作を実行する必要があるかを指定します (この場合は count)。 受信エントリが非アクティブ期間 (日付/タイムスタンプのどちらかの側) の外側にある場合、アプリケーションは新しいセッションを作成します。 保持間隔とは、セッションを一定時間維持することを意味し、セッションの非アクティブ期間を超えても接続できる遅延データを許可します。 さらに、マージの結果として得られる新しいセッションの開始と終了は、最も古い日付/タイム スタンプと最新の日付/タイム スタンプに対応します。

count メソッドのいくつかのエントリを見て、セッションがどのように機能するかを見てみましょう (表 5.1)。

レコードが到着すると、同じキー、現在の日付/タイムスタンプ - 非アクティブ期間よりも短い終了時刻、および現在の日付/タイムスタンプ + 非アクティブ期間よりも大きい開始時刻を持つ既存のセッションが検索されます。 これを考慮すると、表から 5.1 つのエントリが得られます。 XNUMX は次のように XNUMX つのセッションにマージされます。

1. レコード 1 が最初に到着するため、開始時刻は終了時刻と等しく、00:00:00 になります。

2. 次に、エントリ 2 が到着し、23:59:55 以降に終了し、00:00:35 以降に開始するセッションを探します。 レコード 1 を見つけて、セッション 1 と 2 を結合します。セッション 1 の開始時刻 (早い) とセッション 2 の終了時刻 (遅い) を取得し、新しいセッションが 00:00:00 に開始し、00 に終了するようにします。 00:15。

3. レコード 3 が到着すると、00:00:30 と 00:01:10 の間のセッションを探しますが、見つかりません。 キー 123-345-654,FFBE の 00 番目のセッションを追加し、00:50:XNUMX に開始して終了します。

4. レコード 4 が到着し、23:59:45 から 00:00:25 までのセッションを探しています。 今回はセッション 1 と 2 の両方が見つかり、開始時刻が 00:00:00、終了時刻が 00:00:15 の XNUMX つのセッションが XNUMX つに結合されます。

このセクションの説明から、次の重要なニュアンスを覚えておく価値があります。

• セッションは固定サイズのウィンドウではありません。 セッションの継続時間は、指定された期間内のアクティビティによって決まります。
• データ内の日付/時刻スタンプによって、イベントが既存のセッション内に該当するか、アイドル期間内に該当するかが決まります。

次に、次のタイプの窓である「タンブリング」窓について説明します。

「タンブリング」ウィンドウ

タンブリング ウィンドウは、特定の期間内に発生するイベントをキャプチャします。 特定の企業のすべての株式取引を 20 秒ごとにキャプチャする必要があり、その期間中のすべてのイベントを収集するとします。 20 秒の間隔が終了すると、ウィンドウが切り替わり、新しい 20 秒の観察間隔に移動します。 図 5.14 はこの状況を示しています。

ご覧のとおり、過去 20 秒間に受信したすべてのイベントがウィンドウに含まれています。 この期間が終了すると、新しいウィンドウが作成されます。

リスト 5.6 は、タンブリング ウィンドウを使用して 20 秒ごとに株式取引をキャプチャするコードを示しています (src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java にあります)。

TimeWindows.of メソッド呼び出しに対するこの小さな変更により、タンブリング ウィンドウを使用できるようになります。 この例では until() メソッドを呼び出していないため、デフォルトの保持間隔である 24 時間が使用されます。

最後に、ウィンドウ オプションの最後の「ホッピング」ウィンドウに進みます。

スライド (「ジャンピング」) ウィンドウ

スライディング/ホッピング ウィンドウはタンブリング ウィンドウに似ていますが、若干の違いがあります。 スライディング ウィンドウは、最近のイベントを処理するための新しいウィンドウを作成する前に、時間間隔が終了するまで待機しません。 ウィンドウ期間よりも短い待機間隔の後に、新しい計算が開​​始されます。

タンブリング ウィンドウとジャンピング ウィンドウの違いを説明するために、証券取引所の取引をカウントする例に戻りましょう。 私たちの目標は依然としてトランザクション数をカウントすることですが、カウンターを更新するまでずっと待機する必要はありません。 代わりに、より短い間隔でカウンターを更新します。 たとえば、図に示すように、トランザクション数は引き続き 20 秒ごとにカウントしますが、カウンタは 5 秒ごとに更新されます。 5.15。 この場合、最終的にデータが重複する XNUMX つの結果ウィンドウが表示されます。

リスト 5.7 は、スライディング ウィンドウを定義するコードを示しています (src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java にあります)。

タンブリング ウィンドウは、advancedBy() メソッドへの呼び出しを追加することでホッピング ウィンドウに変換できます。 示されている例では、保存間隔は 15 分です。

このセクションでは、集計結果を時間枠に制限する方法を説明しました。 このセクションで特に次の XNUMX つのことを覚えておいてください。

• セッション ウィンドウのサイズは、期間ではなく、ユーザーのアクティビティによって制限されます。
• 「タンブリング」ウィンドウは、一定期間内のイベントの概要を提供します。
• ジャンプ ウィンドウの期間は固定されていますが、頻繁に更新されるため、すべてのウィンドウに重複するエントリが含まれる場合があります。

次に、接続のために KTable を KStream に変換する方法を学びます。

5.3.3. KStream オブジェクトと KTable オブジェクトの接続

第 4 章では、XNUMX つの KStream オブジェクトの接続について説明しました。 次に、KTable と KStream を接続する方法を学習する必要があります。 これは次のような単純な理由で必要になる場合があります。 KStream はレコードのストリームであり、KTable はレコード更新のストリームですが、場合によっては、KTable からの更新を使用してレコード ストリームに追加のコンテキストを追加したい場合があります。

証券取引所の取引数に関するデータを取得し、関連する業界の証券取引所のニュースと組み合わせてみましょう。 すでに持っているコードを考慮して、これを達成するために何をする必要があるかを次に示します。

1. 株式取引数のデータを含む KTable オブジェクトを KStream に変換し、キーをこの銘柄記号に対応する業種を示すキーに置き換えます。
2. 証券取引所ニュースのトピックからデータを読み取る KTable オブジェクトを作成します。 この新しい KTable は産業分野ごとに分類されます。
3. ニュースの更新情報を業界セクターごとの証券取引所の取引数に関する情報と結び付けます。

次に、このアクションプランを実行する方法を見てみましょう。

KTable を KStream に変換する

KTable を KStream に変換するには、次の手順を実行する必要があります。

1. KTable.toStream() メソッドを呼び出します。
2. KStream.map メソッドを呼び出して、キーを業界名に置き換えて、Window インスタンスから Transactionsummary オブジェクトを取得します。

これらの操作を次のように連鎖させます (コードはファイル src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java にあります) (リスト 5.8)。

KStream.map 操作を実行しているため、返された KStream インスタンスは、接続で使用されるときに自動的に再パーティション化されます。

変換プロセスは完了しました。次に、株式ニュースを読み取るための KTable オブジェクトを作成する必要があります。

株式ニュース用のKTableの作成

幸いなことに、KTable オブジェクトの作成には 5 行のコードしかかかりません (コードは src/main/java/bbejeck/chapter_5.9/CountingWindowingAndKtableJoinExample.java にあります) (リスト XNUMX)。

文字列 Serdes が設定で使用されるため、Serde オブジェクトを指定する必要がないことに注意してください。 また、EARLIEST 列挙を使用すると、テーブルの最初からレコードが入力されます。

ここで、最後のステップである接続に進むことができます。

ニュース更新とトランザクション数データを結び付ける

つながりを作ることは難しくありません。 関連する業界の株式ニュースがない場合に備えて、左結合を使用します (必要なコードはファイル src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java にあります) (リスト 5.10)。

この leftJoin 演算子は非常に単純です。 第 4 章の結合とは異なり、KStream-KTable 結合を実行する場合、KTable にはキーごとにエントリが XNUMX つだけ存在するため、JoinWindow メソッドは使用されません。 このような接続には時間の制限がありません。レコードは KTable に存在するか、存在しません。 主な結論: KTable オブジェクトを使用すると、更新頻度の低い参照データで KStream を強化できます。

次に、KStream からイベントを強化するより効率的な方法を見ていきます。

5.3.4. GlobalKTable オブジェクト

ご覧のとおり、イベント ストリームを強化するか、イベント ストリームにコンテキストを追加する必要があります。 第 4 章では 4 つの KStream オブジェクト間の接続について説明し、前のセクションでは KStream と KTable 間の接続について説明しました。 これらのすべてのケースで、キーを新しいタイプまたは値にマッピングするときに、データ ストリームを再分割する必要があります。 再パーティション化は明示的に行われる場合もあれば、Kafka Streams によって自動的に行われる場合もあります。 キーが変更され、レコードが新しいセクションに配置される必要があるため、再パーティション化が必要です。そうしないと接続が不可能になります (これについては、第 4.2.4 章のセクション XNUMX の「データの再パーティション化」セクションで説明しました)。

再パーティション化にはコストがかかる

再パーティション化にはコストがかかります。中間トピックを作成したり、重複したデータを別のトピックに保存したりするための追加のリソース コストです。 また、このトピックからの書き込みと読み取りによりレイテンシが増加することも意味します。 さらに、複数のアスペクトまたはディメンション間で結合する必要がある場合は、結合をチェーンし、レコードを新しいキーでマップし、再パーティション化プロセスを再度実行する必要があります。

より小さなデータセットへの接続

場合によっては、接続される参照データの量が比較的小さいため、その完全なコピーを各ノードにローカルに簡単に適合させることができます。 このような状況のために、Kafka Streams は GlobalKTable クラスを提供します。

GlobalKTable インスタンスは、アプリケーションがすべてのデータを各ノードにレプリケートするため、一意です。 また、すべてのデータが各ノードに存在するため、すべてのパーティションで使用できるように、参照データ キーによってイベント ストリームを分割する必要はありません。 GlobalKTable オブジェクトを使用してキーレス結合を行うこともできます。 この機能を説明するために、前の例の XNUMX つに戻ってみましょう。

KStream オブジェクトを GlobalKTable オブジェクトに接続する

セクション 5.3.2 では、買い手による為替取引のウィンドウ集計を実行しました。 この集計の結果は次のようになります。

{customerId='074-09-3705', stockTicker='GUTM'}, 17
{customerId='037-34-5184', stockTicker='CORK'}, 16

これらの結果は目的を果たしましたが、顧客の名前と完全な会社名も表示されていれば、さらに便利になったでしょう。 顧客名と会社名を追加するには、通常の結合を実行できますが、XNUMX つのキー マッピングと再パーティション化を行う必要があります。 GlobalKTable を使用すると、そのような操作のコストを回避できます。

これを行うには、リスト 5.11 の countStream オブジェクト (対応するコードは src/main/java/bbejeck/chapter_5/GlobalKTableExample.java にあります) を使用し、それを XNUMX つの GlobalKTable オブジェクトに接続します。

これについてはすでに説明したので繰り返しません。 ただし、読みやすくするために、toStream().map 関数のコードはインライン ラムダ式ではなく関数オブジェクトに抽象化されていることに注意してください。

次のステップでは、GlobalKTable の 5 つのインスタンスを宣言します (示されているコードは、ファイル src/main/java/bbejeck/chapter_5.12/GlobalKTableExample.java にあります) (リスト XNUMX)。

トピック名は列挙型を使用して記述されることに注意してください。

すべてのコンポーネントの準備ができたので、あとは接続用のコードを記述するだけです (コードはファイル src/main/java/bbejeck/chapter_5/GlobalKTableExample.java にあります) (リスト 5.13)。

このコードには XNUMX つの結合がありますが、どちらの結果も個別に使用されないため、結合されています。 結果は操作全体の終了時に表示されます。

上記の結合操作を実行すると、次のような結果が得られます。

{customer='Barney, Smith' company="Exxon", transactions= 17}

本質は変わっていませんが、これらの結果はより明確に見えます。

第 4 章まで遡ってみると、すでにいくつかのタイプの接続が実際に動作していることがわかります。 それらは表にリストされています。 5.2. この表は、Kafka Streams のバージョン 1.0.0 の時点での接続機能を反映しています。 将来のリリースでは何かが変更される可能性があります。

最後に、基本を復習しましょう。ローカル状態を使用してイベント ストリーム (KStream) を接続し、ストリーム (KTable) を更新できます。 または、参照データのサイズが大きすぎない場合は、GlobalKTable オブジェクトを使用できます。 GlobalKTables は、すべてのパーティションを各 Kafka Streams アプリケーション ノードにレプリケートし、キーがどのパーティションに対応するかに関係なく、すべてのデータが利用できるようにします。

次に、Kafka ストリーム機能について説明します。この機能のおかげで、Kafka トピックからのデータを消費せずに状態の変化を観察できます。

5.3.5. クエリ可能な状態

すでに状態に関係するいくつかの操作を実行しており、結果は常にコンソールに出力されるか (開発目的)、トピックに書き込まれます (運用目的)。 結果をトピックに書き込む場合、結果を表示するには Kafka コンシューマーを使用する必要があります。

これらのトピックからのデータの読み取りは、マテリアライズド ビューの一種とみなすことができます。 この目的のために、Wikipedia のマテリアライズド ビューの定義を使用できます。「...クエリの結果を含む物理データベース オブジェクト。 たとえば、リモート データのローカル コピー、テーブルの行や列のサブセット、結合結果、集計によって得られた概要テーブルなどです。」 (https://en.wikipedia.org/wiki) /マテリアライズド_ビュー)。

Kafka Streams を使用すると、状態ストアに対して対話型のクエリを実行できるため、これらの実体化されたビューを直接読み取ることができます。 状態ストアへのクエリは読み取り専用操作であることに注意することが重要です。 これにより、アプリケーションがデータを処理しているときに誤って状態の不整合が発生することを心配する必要がなくなります。

状態ストアを直接クエリできる機能は重要です。 これは、最初に Kafka コンシューマーからデータをフェッチしなくても、ダッシュボード アプリケーションを作成できることを意味します。 また、データを再度書き込む必要がないため、アプリケーションの効率も向上します。

• データの局所性のおかげで、すぐにアクセスできます。
• データは外部ストレージに書き込まれないため、データの重複が排除されます。

覚えておいていただきたい主な点は、アプリケーション内から状態を直接クエリできることです。 これによって得られるチャンスはいくら強調してもしすぎることはありません。 Kafka からのデータを消費し、アプリケーションのデータベースにレコードを保存する代わりに、状態ストアにクエリを実行して同じ結果を得ることができます。 ステート ストアへの直接クエリは、コードが少なくなり (コンシューマーが不要)、ソフトウェアも少なくなります (結果を保存するデータベース テーブルが必要なくなります)。

この章ではかなりの内容を説明したので、ステート ストアに対する対話型クエリの説明はここまでにしておきます。 しかし、心配しないでください。第 9 章では、対話型クエリを備えた単純なダッシュボード アプリケーションを作成します。 この章と前の章の例の一部を使用して、対話型クエリと、それを Kafka Streams アプリケーションに追加する方法を示します。

サマリー

• KStream オブジェクトは、データベースへの挿入に相当するイベントのストリームを表します。 KTable オブジェクトは更新ストリームを表し、データベースの更新に似ています。 KTable オブジェクトのサイズは増加せず、古いレコードは新しいレコードに置き換えられます。
• KTable オブジェクトは集計操作に必要です。
• ウィンドウ操作を使用すると、集計されたデータをタイム バケットに分割できます。
• GlobalKTable オブジェクトのおかげで、パーティショニングに関係なく、アプリケーション内のどこからでも参照データにアクセスできます。
• KStream、KTable、および GlobalKTable オブジェクト間の接続が可能です。

これまで、高レベルの KStream DSL を使用して Kafka Streams アプリケーションを構築することに焦点を当ててきました。 高レベルのアプローチを使用すると、きちんとした簡潔なプログラムを作成できますが、これを使用するとトレードオフが発生します。 DSL KStream を使用するということは、制御の度合いを減らしてコードの簡潔性を高めることを意味します。 次の章では、低レベルのハンドラー ノード API を見て、他のトレードオフを試します。 プログラムは以前より長くなりますが、必要なほぼすべてのハンドラー ノードを作成できるようになります。

→本書の詳細については、こちらをご覧ください。 出版社のウェブサイト

→ ハブロジテリの場合 クーポン利用で25％割引 - カフカストリーム

→紙版の書籍の代金をお支払いいただくと、電子書籍がメールで送信されます。

出所： habr.com