Triggers
- ウィンドウが「いつ」出力するかの記述
- 例: event timeで10分ごと
- （補足）ウィンドウは内部状態（e.g. joinなら行, 平均計算なら途中までの平均値と行数）を持つが、Triggerが発動したタイミングでその時点の状態に合わせた出力が成される
- （補足）Triggerが発生してもWindowは残り続ける場合も。例えば10分ごとのfixed windowのうちの一枚(?)が
[13:20:00, 13:30:00)のものだった時、「13:25:00に途中状態から一旦出力する」というトリガも設定できる
Watermarks
- event time軸における時計のようなもの
- watermarkが t を指す時、「t よりも小さい event time が打刻されたデータは全て観測された」ことを（大雑把には）表す
- （補足）ただし、watermarkが t になった後で
t' < tな event time を持つデータがやってくることも現実的にはある。これはlate dataと呼ぶ
- （補足）ただし、watermarkが t になった後で
- watermarkは Chapter3 でみっちりやる
- （補足）event time を持つストリームデータ列を観測した時、watermarkをどの時刻に設定するかは任意性がある。これをいい感じにやればskewを押さえつつもlate dataの発生を抑えられる
- （補足）watermarkはストリーム処理系の中で唯一である必要はない。パイプラインのウィンドウオペレーションごとに持たせたり、分散処理してるノードごとに持たせたりしても良い
Accumulation
- Triggersのところに「1つのwindowから複数の出力が出ることがある」と書いた。
- accumulationは、2, 3, …個目の結果を出力する時のモード
- p.51に出てくるが、Beam Modelでは以下の3つがある。括弧内は「こっちの名称のほうがわかりやすいでしょ？」とp.51の脚注に書いてあるもの
- discarding mode (delta mode)
- accumulating mode (value mode)
- accumulating and retracting mode (value and retraction mode)
What / Where / When / How idiom
Akidau先生イチオシの、ストリーム処理の分析フレームワーク。
この疑問に答えていくと、とあるストリーム処理のパイプライン（またはその部分）が何をしているものかわかるという代物。
（ぶっちゃけ分かりづらくない…？）
上述の5つの概念のうち、
- event time & processing time
- windowing
は全てにカスる。残りの
- Triggers
- Watermarks
- Accumulation
がどう絡むか記載する。
What result
- 計算結果は何になるか。すなわちパイプラインで定義された演算のこと
- ストリーム処理に限らずバッチ処理でも規定される
Where in event time
- これをwhereは苦しくないっすか…？
- event time軸で、結果はいつ出力されるか
- event time と windowing が絡むと書かれてる。triggersも絡むのでは・・・？
- 「event timeが HH:10:00, HH:20:00, … の時に結果を出すウィンドウ」みたいな定義がこれに該当すると思われる
- 文中の
time-agnostic processingとかmore complex types of windowsがどういう意図なのかは掴みかねた- time-agnostic processingはただのフィルタリングとか行ごとに適用される演算。こういうものには
Where in event timeという質問の解は未定義に思うのだが、文ではそう言ってない気もする
- time-agnostic processingはただのフィルタリングとか行ごとに適用される演算。こういうものには
When in processing time
- processing time軸で、ウィンドウからの結果はいつ出力されるか
- すなわちtriggerのこと。以下の理由でwatermarkも絡む:
- triggerがprocessing time軸で定義されているケースはストリーム処理系が参照する時計（通常はシステムクロック）との比較でtriggerが発火する
- triggerがevent time軸で定義されているケースは、watermarkとの比較でtriggerが発火する
- ウィンドウから結果を出力することを materialize と呼ぶ
How do refinements of results relate
- 苦しすぎだって
- Accumulationそのもののことらしい
Batch Foundations: What and Where (p.28~34)
ここからしばらくバッチ処理の話。What resultとWhere in event timeは（無限データ列が対象の）ストリーム処理持ち出さずとも（有限データ列が対象の）バッチ処理で語れるので。
逆に、以下はバッチの話だと通常出てこない。
- Triggers
- When
- Watermarks
- When
- Accumulation
- How
これらは全て「途中経過」に関するものであり、有限データを扱うバッチ処理の場合は最後に一発結果を出すだけで基本的に十分なので。
Beam Modelは無限データも有限データも統一的に扱えるので、ここからBeam Modelのコードスニペットが出てくる。
What: Transformations (p.28~32)
本書の至るところで使われるテーブルと:
> SELECT * FROM UserScores ORDER BY EventTime;
------------------------------------------------
| Name | Team | Score | EventTime | ProcTime |
------------------------------------------------
| Julie | TeamX | 5 | 12:00:26 | 12:05:19 |
| Frank | TeamX | 9 | 12:01:26 | 12:08:19 |
| Ed | TeamX | 7 | 12:02:26 | 12:05:39 |
| Julie | TeamX | 8 | 12:03:06 | 12:07:06 |
| Amy | TeamX | 3 | 12:03:39 | 12:06:13 |
| Fred | TeamX | 4 | 12:04:19 | 12:06:39 |
| Naomi | TeamX | 3 | 12:06:39 | 12:07:19 |
| Becky | TeamX | 8 | 12:07:26 | 12:08:39 |
| Naomi | TeamX | 1 | 12:07:46 | 12:09:00 |
------------------------------------------------
計算が出てくる:
computing keyed integer sums over a simple dataset consisting of nine values
この9行のレコードのScoreカラムの合計値を色んな軸で区切って計算するという話。
上記の9行をevent time v.s. processing time なグラフにプロット（値はScore）すると↓
- Score=5 はだいたい5分くらいのラグがある
- Score=9 はだいぶ遅れてやってくるデータ
とか、そういうことが見て取れる
Beam Modelにおいて、計算対象（入出力）は PCollections とモデリングされ、計算は PTransforms とモデリングされる。
接頭辞の P は Parallel のP。Beam Modelは分散並列前提なので、テーブルやらキューみたいなもの（PCollections）も分散して置かれるし、それに対する処理（PTransforms）も分散並列で行われる。
ここからBeamのコードスニペットが出てくる。書籍のは適宜擬似コードになっているが、本当に動くコードはこちらにあるとのこと: https://github.com/takidau/streamingbook/blob/master/src/main/java/net/streamingbook/BeamModel.java
先に挙げた9行テーブルの例のサブセットとして、Beam Modelで次のような計算を記述している。
黄色いのは “What result” に対応。Beam語で言うと PTransforms に当たる。
- まず、ファイルだかネットワーク越しにやってくる文字列を
IO.read(...)で読み、単一StringなPCollection<String> rawに格納 - 次に
rawにParseFnというパース計算を適用。どうやら文字列にはチーム情報とスコア情報が含まれていたようで、それをkey-value形式でPCollection<KV<Team, Integer>> inputに変換 inputにSum.integersPerKey()を適用。名前からして、「同一キーのバリューの合計値を算出」するPTransform。その結果をPCollection<KV<Team, Integer>> totalsに格納
先の例だと9行全て Team = "TeamX" となっているので、キーは単一となり、合計値も単一となる。
Processing timeが経過していくとスコアが5, 7なレコードが順に到着し、合計値が更新されていく。今回はバッチ計算ということなので、全データが到着したことが保証されるタイミング（上図だと12:10の設定）に1回切り、結果が出力される（上図の右下）。
ちなみに、 http://streamingbook.net/fig のサイトにニュルニュル動く図が公開されているのでこれ見ながらじっくり悩むと便利なときがある
上図は http://streamingbook.net/fig/2-3
Where: Windowing (p.32~34)
ウィンドウはデータ列が無限長でも有限長でも定義できる。
Ch.1でもやったが、よくあるウィンドウは↓の3つ。
Fixed windowのうち、2分おきなやつ（そしてウィンドウ内では同一キーのバリューの合計値を算出）をBeam Modelで記述すると↓
黄色ハイライトのせいで見えづらいが、 Window.into(FixedWindows.of(TWO_MINUTES)) の青字は “Where in event time” に対応している。その問の答えはここでは「event timeで2分ごと」となる。
前出の例でこのウィンドウ計算を行うと、結果は↓のようになる。バッチ処理であり、processing time = 12:10 で一気に結果が出てくる点に再度注意。
Going Streaming: When and How (p.34~)
いよいよストリーム処理特有の「無限データ列処理の途中経過」の話、