504
The Dataflow Model 論文
A Practical Approach to Balancing Correctness, Latency, and Cost in MassiveScale, Unbounded, OutofOrder Data Processing
這篇論文的副標題很長，說明幾點:
1. 這篇文章的主要工作是，Balancing Correctness, Latency, and Cost，故它仍然不能突破CAP定理，仍然是在做tradeoff
2. Unbounded, OutofOrder，針對的對象是無限的，亂序的數據，尤其是亂序的數據，這個點在之前的model無法得到較好的處理
並且這篇論文討論的是，抽象的計算模型和算子，類似mapreduce的論文，設計和實現並不是它的重點
要解決的問題
簡單說，
對於batch，latency太長，而且隻能針對bounded數據
所以現在的主流是Streaming，但是Streaming在保證latency的時候，如何保證Correctness，或Completeness
答案是，根據CAP定理，是不可能的
那麼當前的方案就是balancing，balancing的方式大致就是backfill
無論是Lamda， 還是linkedin的kappa，還是這篇文章的思路可以是說都是backfill的一種表現形式，所以這篇paper的題目也是Practical Approach
即它通過設計做的比之前的方案更精細一些，尤其對於windows的場景，更通用一些
提出的方法
用文章的話說，從概念上看，他的contribution為，
1. Allows for the calculation of event-time ordered results, windowed by features of the data themselves, over an unbounded, unordered data source, with correctness, latency, and cost tunable across a broad spectrum of combinations.
首先，在對無限，無序數據的處理上，尤其是基於event-time的windowed聚合計算，達到latency和correctness的balancing
2. Decomposes pipeline implementation across four related dimensions, providing clarity, composability, and exibility:
What results are being computed.
Where in event time they are being computed.
When in processing time they are materialized.
How earlier results relate to later refinements.
對於流式計算，簡單的one-by-one無狀態模式，沒啥好說的
這篇論文要解決複雜的有狀態模式，比如典型的就是基於windowed的聚合操作
這篇文章把這類操作抽象成4個階段，
what，你要算什麼
where，在什麼範圍內聚合，globe的？在某個時間window中？
when，什麼時候輸出實時統計結果
how，如何修正修正前麵輸出的結果
這樣你把這4個問題解決了，ok，這個問題也就解了，這篇文章後續就是來回答這4個問題
3. Separates the logical notion of data processing from the underlying physical implementation, allowing the choice of batch, micro-batch, or streaming engine to become one of simply correctness, latency, and cost.
這篇文章提出的模型是獨立於物理實現的，可以適用於batch，micro-batch，或streaming，這個是對lamda架構的優化，不用寫兩份代碼了
但注意，這裏說抽象模型可以獨立於物理實現，但並不是說用一個物理engine可以解決所有問題
Scalable implementations of the above atop the MillWheel streaming engine and the FlumeJava batchengine, with an external reimplementation for Google Cloud Dataflow
作者也是基於兩個engine，MillWheel streaming engine and the FlumeJava batchengine，來擴展實現了Dataflow
具體的來說，這篇文章的貢獻是提出3個模型，
A windowing model which supports unaligned event-time windows, and a simple API for their creation and use (Section 2.2).解決Where問題
A triggering model that binds the output times of results to runtime characteristics of the pipeline, with a powerful and exible declarative API for describing desired triggering semantics (Section 2.3). 解決when問題
An incremental processing model that integrates retractions and updates into the windowing and triggering models described above (Section 2.3). 解決how問題
概念
為了能理解這3個模型，先理清一些概念
Unbounded/Bounded vs Streaming/Batch
一句話，Streaming/Batch往往表示execution engine，而unbounded/bounded表示數據的infinite/ finite
Windowing
統計窗口，對於unbounded data，隻能基於windowing做處理
windowing有如下3種，
前兩種很簡單，Sessions Windowing，這個比較新鮮，這個是在google實踐中很重要的一種windowing形式
Session，即當連續出現key1時形成session windowing窗口，沒有key1出現是就不存在窗口，典型應用異常檢測，當出現持續異常時就是session windowing，沒有異常是不需要統計
Time Domains
時間域，分為兩種，
Event Time, which is the time at which the event itself actually occurred，發生時間
Processing Time, which is the time at which an event is observed at any given point during processing within the pipeline，處理時間
顯然處理時間一定是晚於發生時間的，我們可以用下麵的watermark圖來visualize他們的skew關係
我們可以用heuristically established的方式來build這個圖形，用於監控係統的狀況
DATAFLOW MODEL
In this section, we will de ne the formal model for the system and explain why its semantics are general enough to subsume the standard batch, micro-batch, and streaming models, as well as the hybrid streaming and batch semantics of the Lambda Architecture.
Core Primitives
dataflow提供兩種基本原語，分別對應於無狀態和有狀態
ParDo for generic parallel processing. Each input element to be processed (which itself may be a nite collection) is provided to a user-defined function (called a DoFn in Dataflow), which can yield zero or more output elements per input.
基本的無狀態原語
可以等同於flatMap，和map的不同是，可以輸出0到多個結果
GroupByKey for key-grouping (key; value) pairs.
有狀態的原語
Windowing
現在開始介紹windowing模型，這要解決的where問題，即在infinite的數據流中，我們要處理哪部分數據
首先，dataflow將window信息放入tuple內，
所以dataflow的tuple是4元組，(key; value; event time; window)
同時，支持兩種windows操作，
AssignWindows，
可以看到通過AssignWindows，可以將原始數據，轉換為帶windowing信息的數據
在例子給出的case下，一條raw數據會產生兩條帶windowing信息的數據
這樣做的好處就將，where信息固化在原始數據中了，你不用再在代碼裏麵記著
問題是，這樣可能會帶來數據膨脹，如果Sliding（60m，1m），豈不是一條raw tuple，要產生60條帶windowing信息的tuple
WindowMerging，
這個過程，可以用來消除前麵帶來的數據膨脹，
這個過程還是比較清晰的
Triggers & Incremental Processing
開始解決when和how的問題
核心問題，我們麵對的時候無序的數據，那麼我們怎麼知道，這個windowing裏麵的數據已經到全了，可以emit產生結果了？
是不是可以依賴我們上麵給出的watermark圖來預估，是可以的，但這個方案不完善；會有too fast和too slow問題
too fast，即，通過watermark你是無法保證100%數據完整性的，因為watermark是啟發式生成的
too slow，即，latency問題，watermark反映的是大部分數據到全的時間點，必然不會有好的latency
所以可見，這個方案挺廢的，即保證不了一致性，也保證不了latency
那麼回到那個問題，我們怎麼知道什麼時候該emit結果了？
答案是，你無法準確知道
所以這邊的思路和lamda是一致的，先輸出實時數據滿足latency需要，並且用batch數據來backfill，修正數據的正確性
這就是這裏提到的trigger和增量更新模型，
trigger模型解決when的問題，你可以定義各種不同的trigger，已滿足你對latency和correctness的balancing的需求
增量模型解決how的問題，即如何修正數據的正確性，這裏分為3種，
Discarding: Upon triggering, window contents are discarded, and later results bear no relation to previous results.
trigger觸發時，會丟棄當前window的數據，這樣要求various trigger fires to be independent，比如說sum操作
這樣的好處，減小mem的負擔；問題是，會產生碎片化數據，需要後續再次combine和merge
Accumulating: Upon triggering, window contents are left intact in persistent state, and later results become a refinement of previous results.
trigger觸發時，會保留當前window的數據，後續可以繼續refine數據
這樣的場景，適用於downstream consumer支持overwrites操作，比如數據庫
這樣的問題就是，當數據量比較大的時候，你無法在mem裏麵保留長時間數據，那麼需要寫入存儲，那麼backfill可能需要offline來完成
Accumulating & Retracting: 比上麵那種多了retracting
這個隻是用於不同的場景，比如downstream consumer是在做sum統計，那麼必須先把上次的減去，才能加上這次的數據
Examples
對於下麵的input，
Batch Model
Batch的方式，等所有數據都來全了，計算一遍解決，問題就是latency高達接近10分鍾 （對於最早的數據）
基於windowing的batch方式，和普通batch區別，增加windows聚合的結果
Micro-Batch Model
和batch比，兼顧latency
incremental的方式不同，下麵是discarding，看看區別
基於windowing的micro-batch，
基於流的Windowing Model
采用watermark的trigger，
這個的問題上麵說過，
too fast，9在依據watermark觸發時，還沒到
too late, 7的數據要等到8到達的時候才能輸出，
在watermark trigger的基礎上增加micro-batch trigger，這樣的好處還是提高latency，
基於Session Windowing Model
最後更新：2017-04-07 21:05:52