本記事の位置付け

下記読書会のための要約です。

課題図書

今回の範囲

Chapter8 Queries, Modeling, and Transformation より
Queries on Streaming Data
(Data Modeling の手前まで）

要約

ストリーミングデータに対するクエリ

ストリーミングデータは絶え間なく飛んでいる。(constantly in flight)
想像に難くないように、ストリーミングデータに対するクエリはバッチのクエリとは異なる。ストリーミングの優位性をフルに活かすためには、ストリーミングのもつリアルタイムな性質(its real-time nature) を反映したクエリのパターンを採用する必要がある。

例：Kafka, Pulsar

基本的なクエリのパターンとストリーム

7章でやった継続的CDCについて思い出してみよう。

この読書会ではまだ7章をやっていなかった気がするので CDC について要約。

CDCとは Change Data Capture (変更データキャプチャ)の略で、大きく分けてバッチ指向のCDCと継続的CDCがある。

バッチ指向のCDCは、DBのテーブル上の update_at 的な項目を見て、その項目が指定した期間内にあるレコードを取得する手法だが、指定した期間の最新情報しか取得できないなどの制約もある。
（例：ある顧客が、1日の間に3回の取引をしたとする。バッチ指向のCDCを日次の夜間バッチで走らせているシステムの場合、この顧客の情報は、1回目の取引が行われる前（前日の状態）と、3回目の取引が終わった状態（この日の最新の状態）としてしか保持することができない）

継続的CDCは、ほぼリアルタイムで、すべての変更をキャプチャする。
手法としては、ログベースでの変更キャプチャや、DBに変更があるたびにサーバレス関数を起動して変更を記録する方法などがある。

ここでいう（継続的）CDCがやっていることは、本質的には、分析用のDBを本番のDBに対する「速い追従者」として設定するということだ。

ストリーミングのクエリとして最も長く使われている手法は、単純に、本番のDBに対する処理の少し後で、分析用のDBをクエリして、統計的な結果や集計情報を取得する処理をくっつけること（entail）である。

「速い追従者」（fast-follower）アプローチ

なんでこんなアプローチがストリーミングのクエリパターンになっているのか？
同じことを、本番のDBにクエリをかけてやっちゃいけないのか？

原理的には可能だが、実務上はやらない方がいい。
（In principle, yes; in practice, no.）

本番のDBは、本番のワークロードを処理しつつ、同時に分析用の大量データを扱うクエリを処理するようには作られていないから。
そういうクエリを本番用DBに対してかけると、本番処理が遅くなったり、クラッシュしたりする。

（本文脚注より）とある食料品チェーンで、新人のアナリストが本番DBに対して SELECT * でクエリをかけたところ、業務上なくてはならない在庫DBが3日間落ちた

（感想）OLTPとOLAPは分けた方がいいという主張には一応賛成なのですが、上記の事例は本番DBの設計がさすがに脆弱すぎるように思う。 SELECT * を許可しないとか、適度なタイムアウトを設定するとか、本番側の設計で防げた部分もあるのでは。

ベーシックなCDCのクエリパターン（＝変更部分を分析用のDBにちょこまか書きに行くアプローチ）を採用すれば、リアルタイムな分析を実施しながら、本番処理への影響を最小限にとどめることができる。

「速い追従者」のパターンでは、従来型のトランザクションDBを追従者として使うこともできるが、適切なOLAP指向のシステムを使うことに著しい優位性がある。

Druid と Big Queryはいずれも、ラムダアーキテクチャと似た形で、ストリーミング用のバッファと長期にわたる列指向のストレージを併せ持つ。

「速い追従者」アプローチの制約

「速い追従者」アプローチのには致命的な制約がある。
このアプローチは、本質的にバッチのクエリパターンのことを顧みていない。
現在のテーブルの状態に対して SELECT のクエリをかけていながら、ストリーミング中の変化に関する(??)イベントを動的にトリガーする機会を逃すことになる。

下線部分自信なし。

原文：You’re still running SELECT queries against the current table state, and missing the opportunity to dynamically trigger events off changes in the stream.
→ 最後の changes in the stream はどこにかかる？

という英語に関する疑問と、現在のテーブルの状態に対してクエリをかける行為がストリーミングに対するイベントの軌道をミスるという話なら、顧みられていないのはバッチじゃなくてストリーミングじゃないのかという疑問。

カッパ・アーキテクチャ

※この名称から誰もが一度は思い浮かべるであろう姿

※カッパ・アーキテクチャと上記の妖怪は、おそらくあまり（全然）関係がない。この読書会で、ラムダ・アーキテクチャとあわせてカッパ・アーキテクチャを扱った回のブログ↓があるので詳細はそちらを参照

カッパ・アーキテクチャでは、すべてのデータをイベントのように扱い、それらのイベントをテーブルとしてではなく、stream として保管する。

stream と store の訳語に自信なし。
stream はそのまま stream , store は保管と訳しています。

本番のDBがソースの場合、カッパ・アーキテクチャはCDCのイベントを保管する。

イベントの stream はアプリのバックエンドから直接流れてくることもある。
（IoT のデバイス群や、発生したイベントをネットワーク越しにプッシュしてくるあらゆるシステム）

カッパ・アーキテクチャでは、ストリーミングの保管システムを単純にバッファとして扱うのではなく、より長い期間にわたってイベントを保持することができ、かつ、そのストレージから直接データをクエリすることができる。

下線部わかったようでわからない。
具体的にどういうツールのどういう機能をイメージしているのか。

こんな絵を描かれても上記の疑問は解決しない

データの保持期間は月、年単位などの長い期間にも設定できる。
純粋なリアルタイム指向のシステムでは、通常は長くて１週間しかデータを保持しないのに比べると、これはかなり長い期間である。

カッパ・アーキテクチャの大まかなアイディアは、ストリーミングのストレージをリアルタイムの移送レイヤーとして扱い、データベースを履歴データの取得・クエリに使うというものである。

これは、ストリーミングのストレージに対して直接クエリをかけられるという能力によっても、外部ツールの助けを使う場合であっても起こりうる（This can happen either ...）。

例：Kafka の KSQL
データの集計・集約、統計的な計算、セッション化をサポート。

クエリの要求がもっと複雑だったり、データを他のデータソースからのデータと組み合わせる必要がある場合は、Sparkのような外部ツールが特定の時間の範囲のデータを Kafka から読み出して、クエリの結果を計算する。

ストリーミングのデータ保管システムは、Flink や Beam のような他のアプリケーションにデータを与える(feed)こともできる。

ウィンドウ処理、トリガー、（データが自発的に）生み出す統計、遅れて到着するデータ

伝統的なバッチクエリの根源的な制約は、このパラダイムはクエリエンジンを外部の観察者として扱うということである。

例：時間単位の cron のように、データからみて外部の行為者がクエリをかけるという仕組み。

一方、より広く使われているストリーミングのシステムでは、データそのものが計算の要求（notion of computation）を起動する仕組みをサポートしている。

データそのもの（？They）が、一定の数のレコードがバッファに集められるたびに平均値や中央値などの統計情報を吐き出したり（emit）、ユーザーがセッションを閉じるたびにサマリー情報を出力したりする。

ウィンドウとその類型

ウィンドウは、ストリーミングのクエリと処理において欠かすことのできない特徴である。
ウィンドウとは、動的なトリガーによって処理される小さいバッチのことである。

ウィンドウは動的に生成されるが、その方法はいくつかある。例を見てみましょう。

セッションウィンドウ

ブラウザなどで使う「セッション」と似た概念と理解
キーごとのまとまり（文中ではユーザーIDが例示されている）で、近い時間内に起こったイベントのまとまりをセッションとして扱う
一定時間操作が行われないと、セッションが閉じられる

セッションにおける遅延データの扱い

たとえば遅延時間を5分と設定すると、5分以内の遅れで到着したデータがあれば、ウィンドウを（再び）オープンする。

固定時間ウィンドウ

固定された時間ごとにウィンドウを作っては閉じる方式
伝統的なバッチETL処理に似ている（例：日次バッチ、時間ごとのバッチなど）

この図になぜ「One-minute interval」というタイトル（？）がついているのかが謎。。。
20秒のかたまりが3つで1分だから？？

スライディング・ウィンドウ

固定時間ウィンドウの、オーバーラップした部分があるバージョン

スライディング（の時間？処理内容？）は様々に設定できる。（The sliding can vary）

例えば、30秒ごとにスライディング・ウィンドウを作るが、統計データの計算を行うのはユーザーがあるバナーをクリックしたときだけ、とすることもできる。
この場合、多数のユーザーがバナーを一斉にクリックした場合はアウトプットが極端に多くなるが、何も起こらなければ計算も行わない。

ウォーターマーク

第7章（ingest data の章）でもとりあげたとおり、データは生成された順番どおりに到着するとは限らない。
（ネットワーク遅延など様々な要因による）

遅延データを扱うための法式のひとつがウォーターマークである。

ウォーターマーク：遅延を判断するための基準となる時間
あるウィンドウに新しいデータが到着したが、タイムスタンプがウォーターマークより古ければ、遅延データとみなす

遅延データとみなした後の話には触れられていない。
単純に破棄するのか、古いウィンドウに足し込むのか、このあたりは要件によって設計しろということなのか。

ストリーミングデータを他のデータと組み合わせる

データはしばしば、他のデータと組み合わせることによって価値が出る。
ストリーミングデータも同様である。

以下はストリーミングデータを組み合わせる手段の例。

従来型のテーブル結合

テーブルにしたあとで Joinする。（ストリーミングの意味・・・）

エンリッチメント

ストリーミングデータに、ストリーミングじゃないデータの情報を加えること。

例：

小売店のシステムで、ユーザーIDと商品のIDがストリーミングとして流れてくる
商品IDをキーに、商品の詳細情報（←ストリーミングじゃないデータ）をクエリして、それをストリーミングデータにのせる

実務上、エンリッチメントのソース（ストリーミングじゃない側のデータ）はあらゆる場所になりうる。
例：クラウド上のDWH、RDBMS、オブジェクトストレージ上のファイル　など。

ストリーミングとストリーミングの結合

ストリーミングデータに、別のストリーミングデータを結合する需要が高まっている。

例として、あるオンライン小売業者が、Web上のイベントのデータ（ストリーミングデータ①）と、広告プラットフォームのデータ（ストリーミングデータ②）を組み合わせたいとする。

この会社は、両方のデータを Spark に食わせる（feed）こともできるが、諸々の複雑な考慮事項が浮上する。

その代表例が、データの到着時間の遅れである。

ストリーミングデータ同士の結合におけるレイテンシの考慮

ストリーミングにおいて、データは様々な遅延時間（レイテンシ）を持ちうる。

データソースによるレイテンシの違い
- 例：広告プラットフォームは、必ず5分遅れでデータを送ってくる
イベントによるレイテンシ
- 例：セッションのクローズや、電話などのイベントの情報は、ユーザーが次にモバイルアプリを起動するまでフィードされない

そんなわけで、典型的なストリーミング同士の結合はストリーミングバッファを利用する。

ストリーミングデータは一定時間、バッファに溜められ、結合処理のプロセッサー（join processor）はバッファを参照してデータを結合しようとする。

データの保持期間を過ぎると、データは破棄される（evicted）。

バッファの時間は設定により変更可能（configurable）だが、インターバルを長くとるほど多くのストレージその他の資源を必要とする。