KafkaおよびStream Processingのストリームとテーブルについて、パート1

* Michael G. Nollは、Apache KafkaやApache Stormを含むオープンソースプロジェクトに積極的に貢献しています。

この記事は主に、Apache Kafkaやストリーム処理[Stream Processing]に精通している人に役立ちます。

この記事では、おそらくミニシリーズの最初の記事で、ストリーミング処理、特にApache Kafkaでのストリームとテーブルの概念について説明したいと思います。 現在および将来の問題をより良く、かつ/またはより速く解決するのに役立つ、より良い理論的理解とアイデアがあればいいのですが。

内容：

*動機
*プレーン言語のストリームとテーブル
*図解された例
*シンプルな言語のカフカストリームとテーブル
* Kafka Streams、KSQL、およびScalaの同等物の詳細
*テーブルは巨人の肩の上に立つ（ストリーム上）
*データベースの裏返し
*結論

動機、またはなぜ気にする必要がありますか？

私の日々の仕事では、多くのApache Kafkaユーザーと、 Kafka StreamsおよびKSQL （KafkaのストリーミングSQL）を介してKafkaのストリーミング処理に関与しているユーザーと通信します。 ストリーミング処理やKafkaの使用経験のあるユーザーもいれば、OracleやMySQLなどのRDBMSの使用経験のあるユーザーもいません。

よくある質問：「StreamsとTablesの違いは何ですか？」この記事では、より深く理解できるように、短い（TL; DR）と長い両方の答えを提供したいと思います。 以下の説明の一部は、理解と暗記を単純化するため、若干単純化されます（たとえば、アトラクションのより単純なニュートンモデルは、ほとんどの日常的な状況に十分であり、幸いなことに、ストリーム処理は 、アインシュタインの相対論的モデルとても複雑です）。

別の一般的な質問：「良い、しかしなぜそれが私を悩ませる必要がありますか？ これは日々の仕事でどのように役立つのでしょうか？」要するに、多くの理由からです！ ストリーミング処理の使用を開始すると、実際にはほとんどの場合、ストリームとテーブルの両方が必要であることがすぐにわかります。 後で説明するように、表は状態を表します。 結合[たとえば（たとえば、ファクトのフローをディメンションテーブル[ ディメンションテーブル ]と組み合わせてリアルタイムでデータを強化 [ データエンリッチメント ] ）または集約[集約] （たとえば、リアルタイムで5分間で主要なビジネスインジケータの平均値を計算する）、その後、テーブルにストリーミング画像が導入されます。 そうでなければ、それはあなたが自分でそれをしなければならないことを意味します[多くのDIYの苦痛] 。

おそらくこのエリアの最初の「Hello World」である悪名高いWordCountの例でさえ、「状態」カテゴリに分類されます。これは、単語カウントのために連続的に更新されるテーブル/マップに行のストリームを集約する状態処理の例です。 したがって、単純なWordCountストリーミングを実装しているか、 不正検出などのより複雑なものを実装しているかに関係なく、基本的なデータ構造と内部に必要なすべてのものを使用したスト​​リーム処理のための使いやすいソリューションが必要です（ヒント：ストリームとテーブル） もちろん、CassandraやMySQLなどのリモートストレージとストリーミング処理テクノロジーを組み合わせて（必要に応じて）、必要に応じてHadoop / HDFSを追加してフォールトトレランス処理を提供する複雑で不要なアーキテクチャを構築する必要はありません[フォールトトレランス処理] （3つのことが多すぎます）。

プレーン言語のストリームとテーブル

ここに私が思いつく最高の例えがあります：

• カフカのストリームは、 時間の初めから現在までの世界で起こったすべてのイベント（または単にビジネスイベント）の完全な履歴です。 彼は過去と現在を代表しています。 今日から明日に移行するにつれて、新しいイベントが世界の歴史に絶えず追加されています。
• カフカのテーブルは今日の世界の状態です 。 彼女は現在を代表しています。 これは世界のすべてのイベントの集合体であり 、今日から明日へと移行するにつれて常に変化しています。

また、将来の投稿の食前酒として：世界（ストリーム）のイベントの全履歴にアクセスできる場合、いつでも世界の状態、つまりストリーム内の任意の時点tテーブルを復元できます。ここで、 t t= 。 言い換えれば、世界の状態（テーブル）の「スナップショット」をいつでも作成できます。たとえば、ギザに大ピラミッドが建設された紀元前2560年、またはヨーロッパの連合。

図解された例

最初の例は、ユーザーのジオロケーションを含むストリームを示しています。これは、各ユーザーの現在の（最後の）位置を固定するテーブルに集約されます 。 後で説明しますが、Kafka [トピック]トピックをテーブルに直接読み込むと、これがテーブルのデフォルトのセマンティクスになります。



2番目のユースケースは、同じユーザージオロケーション更新ストリームを示していますが、ストリームは、各ユーザーが訪れた場所の数を記録するテーブルに集約されます 。 集計関数が異なるため（ここでは数量のカウント）、テーブルの内容も異なります。 より正確には、他のキー値。

簡単な言語のKafkaのストリームとテーブル

詳細を説明する前に、簡単なものから始めましょう。

Kafka のトピックは、キーと値のペアの無制限のシーケンスです。 キーと値は、通常のバイト配列です。 <byte[], byte[]> 。

ストリームは、 [スキーマ]スキームを使用したトピックです。 キーと値はバイト配列ではなくなりましたが、特定のタイプがあります。

例： <byte[], byte[]>トピックは、ユーザーの地理的位置の<User, GeoLocation>ストリームとして読み取ります。

テーブルとは、通常の意味でのテーブルです（すでにRDBMSに精通していて、Kafkaだけを知っている皆さんの喜びを感じます）。 しかし、ストリーム処理のプリズムを通して見ると、テーブルも集約されたストリームであることがわかります （「テーブルはテーブルです」の定義で停止することを本当に期待していなかったでしょう？）。

例：ジオデータの更新を伴う<User, GeoLocation>ストリームは、ユーザーの最後の位置を追跡する<User, GeoLocation>テーブルに集約されます。 集計段階で、テーブル内の[UPSERT]値は、入力ストリームのキーによって更新されます。 これは、上記の最初の例で見ました。

例： <User, GeoLocation>ストリームは<User, Long>テーブルに集約され、各ユーザーの訪問済みロケーションの数を追跡します。 集計段階では、テーブル内のキーの値が継続的に計算（および更新）されます。 これは、上記の2番目の例で見ました。

合計：



トピック、ストリーム、およびテーブルには、Kafkaで次のプロパティがあります。

トピック、ストリーム、およびテーブルがKafka Streams APIおよびKSQLにどのように関連するかを見てみましょう。また、プログラミング言語との類似性を引き出します（たとえば、トピック/ストリーム/テーブルをパーティション化できるなどは無視します）。

しかし、このレベルでのこの履歴書は、ほとんど役に立たないかもしれません。 それでは、詳しく見てみましょう。

Kafka Streams、KSQL、およびScalaの同等物の詳細

以下の各セクションは、Scala（ストリーミング処理が同じマシンで行われることを想像してください）とScala REPLの類推から始めて、コードをコピーして自分でプレイできるようにします。次に、Kafka StreamsとKSQLで同じことを行う方法を説明します（分散マシンでの柔軟でスケーラブルで耐障害性のあるストリーミング処理）。 最初に述べたように、以下の説明を少し簡略化します。 たとえば、Kafkaでのパーティション分割の影響は考慮しません。

Scalaを知らない場合：恥ずかしがらないでください！ Scalaの類似物を詳細に理解する必要はありません。 どの操作（たとえばmap() ）が結合されているか、それらが何であるか（たとえばreduceLeft()が集約）、およびストリームの「チェーン」とテーブルの「チェーン」との相関関係に注意を払うだけで十分です。

トピックス

Kafkaのトピックは、キーと値のメッセージで構成されています。 このトピックは、メッセージのシリアル化形式や「タイプ」に依存しません。メッセージのキーと値は、通常のbyte[]配列のように扱われます。 言い換えれば、この観点からは、データの中に何があるのか​​分かりません。

Kafka StreamsおよびKSQLには「トピック」の概念はありません。 彼らはストリームとテーブルについてのみ知っています。 したがって、ここではScalaのアナログトピックのみを示します。

 // Scala analogy scala> val topic: Seq[(Array[Byte], Array[Byte])] = Seq((Array(97, 108, 105, 99, 101),Array(80, 97, 114, 105, 115)), (Array(98, 111, 98),Array(83, 121, 100, 110, 101, 121)), (Array(97, 108, 105, 99, 101),Array(82, 111, 109, 101)), (Array(98, 111, 98),Array(76, 105, 109, 97)), (Array(97, 108, 105, 99, 101),Array(66, 101, 114, 108, 105, 110))) 

ストリーム

ここで、トピックをストリームに読み込み、スキームに関する情報を追加します（スキーム[schema-on-read]を読み取ります ）。 言い換えれば、生の、型付けされていないトピックを「型付きトピック」またはストリームに変換します。

読み取りスキームと書き込みスキーマ[schema-on-write] ： Kafkaとそのトピックは、データのシリアル化形式に依存しません。 したがって、ストリームまたはテーブルのデータを読み取る場合はスキームを指定する必要があります。 これは、読み取りスキームと呼ばれます。 読み取り方式には、長所と短所の両方があります。 幸いなことに、アプリケーションとサービスでAPIコントラクトを定義するのと同じように、データのコントラクトを定義することで、読み取りスキームと書き込みスキームの中間を選択できます。 これは、 Apache Avroなどの構造化された拡張可能なデータ形式を選択し、 Confluent Schema RegistryなどのAvroスキーム用のレジストリ展開で実現できます。 そして、もし興味があれば、Kafka StreamsとKSQLはAvroをサポートしています。

Scalaでは、これは以下のmap()操作を使用して実現されます。 この例では、ペア<String, String>からストリームを取得します。 データの内部を確認する方法に注目してください。

 // Scala analogy scala> val stream = topic | .map { case (k: Array[Byte], v: Array[Byte]) => new String(k) -> new String(v) } // => stream: Seq[(String, String)] = // List((alice,Paris), (bob,Sydney), (alice,Rome), (bob,Lima), (alice,Berlin)) 

Kafka Streamsでは、 StreamsBuilder#stream()介してStreamsBuilder#stream()トピックを読みます。 ここでは、トピックからデータを読み取るときにConsumed.with()パラメーターを使用して目的のスキーマを決定する必要があります。

 StreamsBuilder builder = new StreamsBuilder(); KStream<String, String> stream = builder.stream("input-topic", Consumed.with(Serdes.String(), Serdes.String())); 

KSQLでは、次のようなことをして、トピックをSTREAMとして読む必要があります。 ここでは、トピックからデータを読み取るときに列の名前とタイプを指定することにより、目的のスキームを定義します。

 CREATE STREAM myStream (username VARCHAR, location VARCHAR) WITH (KAFKA_TOPIC='input-topic', VALUE_FORMAT='...') 

テーブル

現在、表の同じトピックを読んでいます。 まず、回線に関する情報を追加する必要があります（回線の読み取り）。 次に、ストリームをテーブルに変換する必要があります。 Kafkaのテーブルのセマンティクスでは、最終的なテーブルでは、このキーの最後の値のトピックからの各メッセージキーを表示する必要があると述べています。

最初の例を使用してみましょう。ここでは、サマリーテーブルが各ユーザーの最後の場所を追跡します。



Scalaの場合：

 // Scala analogy scala> val table = topic | .map { case (k: Array[Byte], v: Array[Byte]) => new String(k) -> new String(v) } | .groupBy(_._1) | .map { case (k, v) => (k, v.reduceLeft( (aggV, newV) => newV)._2) } // => table: scala.collection.immutable.Map[String,String] = // Map(alice -> Berlin, bob -> Lima) 

スキームに関する情報の追加は、上記のストリームの例と同様に、最初のmap()操作を使用して実現されます。 ストリームは、集計ステップ （詳細は後述）を使用して[ストリームからテーブル]テーブルに変換されますが、この場合はテーブルのUPSERT操作（状態なし）で、これはgroupBy().map()ステップgroupBy().map()各キーのreduceLeft()操作が含まれます。 集約とは、キーごとに多くの値を1つに圧縮することを意味します。 この特定のreduceLeft()集約-以前のaggV値は、指定されたキーの新しい値の計算に使用されないことに注意してください。

ストリームとテーブルの関係について興味深いのは、上記のコマンドが以下の短いオプションに相当するテーブルを作成することです（ 参照透過性を思い出してください）。ストリームから直接テーブルを構築し、タイプ/スキームの指定をスキップできますストリームは既に入力されているためです。 テーブルは、ストリームの派生 、集約であることがわかります 。

 // Scala analogy, simplified scala> val table = stream | .groupBy(_._1) | .map { case (k, v) => (k, v.reduceLeft( (aggV, newV) => newV)._2) } // => table: scala.collection.immutable.Map[String,String] = // Map(alice -> Berlin, bob -> Lima) 

Kafka Streamsでは、通常StreamsBuilder#table()を使用KTable単純なワンライナーでKTableのKafkaトピックを読み取ります。

 KTable<String, String> table = builder.table("input-topic", Consumed.with(Serdes.String(), Serdes.String())); 

ただし、わかりやすくするために、最初にKStreamでトピックを読んでから、上記に示したのと同じ集計手順を実行してKStreamすることもできます。

 KStream<String, String> stream = ...; KTable<String, String> table = stream .groupByKey() .reduce((aggV, newV) -> newV); 

KSQLでは、トピックをTABLEとして読むために次のようなことを行う必要があります。 ここでは、トピックから読み取るときに列の名前とタイプを指定して、目的のスキームを決定する必要があります。

 CREATE TABLE myTable (username VARCHAR, location VARCHAR) WITH (KAFKA_TOPIC='input-topic', KEY='username', VALUE_FORMAT='...') 

これはどういう意味ですか？ これは、最初に述べたように、テーブルが実際に集約されたストリームであることを意味します。 これは、テーブルがトピックから直接作成されたとき、上記の特別なケースで直接見ました。 ただし、これは実際には一般的なケースです。

テーブルは巨人の肩の上に立つ（ストリーム上）

概念的には、Kafkaの1次データ構造はストリームのみです。 一方、テーブルは、（1） キーごとの集約を通じて既存のストリームから派生するか、（2）常に集約されたストリームに拡張される既存のストリームから削除されます（最後のテーブルを「proto-streams」と呼ぶことができます[ 「Ur-stream」] ）。

テーブルは、ストリームの具体化されたビューとしても説明されます。 このコンテキストでは、ストリーミングは集約にすぎません。

2つのケースのうち、（1）は議論のためにより興味深いので、これに注目しましょう。 それはおそらく、Kafkaで集計がどのように機能するかを最初に把握する必要があることを意味します。

カフカの集計

集約は、ストリーム処理の一種です。 他のタイプには、たとえば、フィルタリング[filters]および結合[joins]が含まれます。

前にわかったように、Kafkaのデータはキーと値のペアとして表示されます。 さらに、Kafkaの集計の最初の特性は、それらがすべてkeyによって計算されることです。 そのため、 groupBy() Streamsの集約フェーズの前にgroupBy()またはgroupByKey()介してgroupBy()グループKStreamする必要があります。 同じ理由で、上記のScalaの例ではgroupBy()を使用するgroupBy()ました。

パーティション化とメッセージキー： Kafkaの等しく重要な側面は、この記事では無視しますが、トピック、ストリーム、およびテーブルがパーティション化されることです。 実際、データはパーティションごとにキーごとに処理および集約されます。 デフォルトでは、メッセージ/レコードはキーに基づいてパーティション間で分散されます。したがって、実際には、技術的に複雑でより正確な「キーパーティション分割」ではなく、「キー集約」を簡素化することは完全に受け入れられます。 ただし、カスタムパーティショニング割り当てアルゴリズムを使用する場合は、処理ロジックでこれを考慮する必要があります。

Kafkaの集計の2番目の特性は、新しいデータが着信ストリームに到着すると、集計が継続的に更新されることです。 キーによる計算のプロパティと共に、これにはテーブルが必要です。より正確には、結果として、したがって返される集計のタイプが変更可能なテーブルが必要です。 キーの以前の値（集計結果）は、常に新しい値で上書きされます。 Kafka StreamsとKSQLの両方で、集計は常にテーブルを返します。

2番目の例に戻り、ストリーム内の各ユーザーが訪れた場所の数を計算します。



カウントは集計の一種です。 値を計算するには、前のセクションの集計ステージを.reduce((aggV, newV) -> newV)から.map { case (k, v) => (k, v.length) }です。 戻り値の型はテーブル/マップであることに注意してください（Scalaではデフォルトで不変map使用するため、Scalaコードではmap不変[不変マップ]であるという事実を無視してください）。

 // Scala analogy scala> val visitedLocationsPerUser = stream | .groupBy(_._1) | .map { case (k, v) => (k, v.length) } // => visitedLocationsPerUser: scala.collection.immutable.Map[String,Int] = // Map(alice -> 3, bob -> 2) 

上記のScalaの例に相当するKafka Streamsコード：

 KTable<String, Long> visitedLocationsPerUser = stream .groupByKey() .count(); 

KSQLの場合：

 CREATE TABLE visitedLocationsPerUser AS SELECT username, COUNT(*) FROM myStream GROUP BY username; 

テーブル-集約されたストリーム（入力ストリーム→テーブル）

上で見たように、テーブルは入力ストリームの集約であり、要するに、テーブルは集約ストリームです。 Kafka StreamsまたはKSQLで集計を実行すると、結果は常にテーブルになります。

集約段階の特性により、状態なしでUPSERTセマンティクスを介してストリームからテーブルを直接取得するかどうかが決定されます（テーブルは、ストリームの最後の値にキーを表示します。これは、Kafkaトピックを直接テーブルに読み込むときに集約されます）。 ステートフルカウント （最後の例を参照）、または加算、平均化などのより複雑な集計。 Kafka StreamsおよびKSQLを使用する場合、タンブリングウィンドウ、ホッピングウィンドウ、およびセッションウィンドウを使用したウィンドウ化された集計を含む、集計の多くのオプションがあります。

テーブルに変更ストリームがあります（テーブル→出力ストリーム）

テーブルは入力ストリームの集合体ですが、独自の出力ストリームもあります！ データベースに変更データ（CDC）を記録するように、Kafkaのテーブルのすべての変更は、テーブルのchangelogストリームと呼ばれる内部変更ストリームに記録されます。 Kafka StreamsおよびKSQLでの計算の多くは、実際にはchangelogストリームで行われます 。 これにより、たとえば、Kafka StreamsおよびKSQLは、 イベント時処理のセマンティクスのセマンティクスに従って履歴データを正しく処理できます。ストリームは現在と過去の両方を表しますが、表は現在のみを表すことができることに注意してください正確に、固定された時間[スナップショットin time] ）。

注： Kafka Streamsでは、 KTable#toStream()使用して、明示的にテーブルを変更ストリーム[changelog stream]に変換できます。

最初の例を次に示しますが、変更ログストリームを使用します 。



テーブルのchangelogストリームは、このテーブルの入力ストリームのコピーであることに注意してください。 これは、対応する集約関数（UPSERT）の性質によるものです。 「これはディスクスペースを消費する1対1のコピーではありませんか？」-Kafka StreamsとKSQLは、不要なデータコピーとローカル/ネットワークIOを最小限に抑える最適化を実行します。 基本的に何が起こっているかをよりよく説明するために、上の図のこれらの最適化を無視します。

そして最後に、 changelog streamを含む2番目のユースケース。 ここでは、 キーごとのカウントを実行する別の集計関数があるため、テーブルの変更のストリームは異なります。



ただし、これらの内部変更ログストリームは、アーキテクチャおよび運用上の影響も受けます。 変更ストリームは継続的にバックアップされ、Kafkaのトピックとして保存されます。テーマ自体は、Kafka StreamsおよびKSQLの弾力性と復元力を提供する魔法の一部です。 これは、 [stateful]状態または[stateless]なしの処理に関係なく、データを失うことなく、すべての操作を通じて、マシン/仮想マシン/コンテナ間で処理タスクを移動できるためです。 テーブルはアプリケーション（[Kafka Streams]）またはクエリ（KSQL）の[状態]状態の一部であるため、Kafkaは処理コード（簡単）だけでなく、テーブルを含む処理ステータスもマシン間で高速かつ信頼性の高い方法で転送する必要があります（はるかに複雑です）。 テーブルをクライアントマシンAからマシンBに移動する必要があるときはいつでも、新しい宛先Bで、テーブルはマシンAとまったく同じ状態の変更ログストリームからKafka（サーバー側）に再構築されます。上記の最後の図では、入力ストリームを処理することなく、変更ログストリームから「 カウントテーブル」を簡単に取得できます。

双対ストリーム表

ストリームテーブルの二重性という用語は、ストリームとテーブルの間の上記の関係を指します。 これは、たとえば、ストリームをテーブルに、このテーブルを別のストリームに、結果のストリームを別のテーブルに、というように変換できることを意味します。 詳細については、Confluent： Introducing Kafka Streams：Stream Processing Made Simpleブログ投稿を参照してください。

データベースの裏返し

前のセクションで説明した内容に加えて、データベースの裏返しに関する記事をご覧になった方もいらっしゃるかもしれませんが、この全体を見てみたいと思うかもしれません。 ここでは詳しく説明しませんので、Kafkaとストリーミング処理の世界をデータベースの世界と簡単に比較してみましょう。 警戒してください：以降、深刻な単純化[白黒の単純化] 。

データベースでは、 テーブルは1次構造です。 これがあなたの仕事です。 「ストリーム」は、たとえばMySQLのbinlogやOracleのGoldenGateのようなデータベースにも存在しますが、通常はそれらと直接やり取りできないという意味であなたから隠されています。 データベースは現在については知っていますが、過去については知りません（過去が必要な場合は、 バックアップテープからデータを復元します。これは単なるハードウェアストリームです）。

Kafkaおよびストリーム処理では、 ストリームは1次構造です。 前に見たように、テーブルはストリームの派生物です 。 ストリームは現在と過去を知っています。 例として、 New York Timesは、1850年代から160年にわたるジャーナリズムのすべての公開された記事を 、信頼できるデータのソースであるカフカに保管しています。

要するに、データベースは最初にテーブルとして、次にストリームとして考えます。 カフカは最初にストリームで考え、次にテーブルで考えます。 , Kafka- , , — , WordCount, , . , Kafka , Kafka Streams KSQL, ( ). Kafka , , - [stream-relational] , [stream-only].

, — . Kafka , — .

おわりに

, , Kafka . , , « » « Kafka ».

- Kafka, Kafka Streams KSQL, :

• , KSQL , SQL- Kafka, Kafka - . , , Kafka , . KSQL clickstream ( Docker), Kafka, KSQL, Elasticsearch Grafana real-time dashboard .
• , Java Scala Kafka Streams API .
• , , , , KSQL, Kafka Streams, KSQL.

, Kafka Streams KSQL, Kafka , , (, , , , , , ). , . :-)

, « , 1». , , . ? !

— , , .