Dataflow編程模型和spark streaming結合

主要介紹一下Dataflow編程模型的基本思想，後面再簡單比較一下Spark streaming的編程模型

== 是什麽 ==

為用戶提供以流式或批量模式處理海量數據的能力，該服務的編程接口模型（或者說計算框架）也就是下面要討論的dataflow model

流式計算框架處理框架很多，也有大量的模型／框架號稱能較好的處理流式和批量計算場景，比如Lambda模型，比如Spark等等，那麽dataflow模型有什麽特別的呢？

這就要要從流式（streaming） 和批量（ batch） 這兩個詞的語意說起，簡單的說，谷歌的同學認為目前的各種計算框架模型以及用戶在一定程度上對這兩個詞的語意的使用姿勢不夠恰當，或者說用這兩個詞來區分應用場景，進而給計算框架分類並不合適。而這種分類給框架的應用甚至設計帶來了一定的認識的誤解和偏差。

同學認為，當大家談論到流式計算，或者流式數據時，內心想表達的場景，實際上更準確的說法應該是unbounded data(processing)，也就是無邊界的連續的數據（的處理）；對應的批量（計算），更準確的說法是bounded data(processing)，亦即有明確邊界的數據的處理。而stream和batch實際只是這兩種數據集歷史上傳統使用的處理方式而已，這兩者並不完全等價。而隨著技術的發展，繼續用這種方式來分類和看待問題就顯得不夠高大上了。

一篇chanllenge前段時間熱門的Lambda模型 https://www.oreilly.com/ideas/questioning-the-lambda-architecture 的文章中所表達的觀點，一定程度上也是上面這種思想的一個體現。

而Dataflow模型則是谷歌的同學在處理無邊界數據的實踐中，總結的一套SDK級別的解決方案，其目標是做到在非有序的，無邊界的海量數據上，基於事件時間進行運算，並能根據數據自身的屬性進行window操作，同時數據處理過程的正確性，延遲，代價可根據需求進行靈活的調整配置。其底層計算引擎依托於 Millwheel 實時計算框架和FlumeJava批處理框架，在開源了相關SDK以後，發起了beam項目（http://beam.incubator.apache.org/），其底層計算引擎也可以替換適配成Spark／Flink等開源計算框架（進行中）

== 核心思想 ==

基本上Lambda模型被挑戰的點是，用一個流式＋批量的拼湊方案去解決海量無限數據的實時統計問題，看起來很美，但是出發點立意有些Low（亦即，認定了這種問題只能通過兩套截然不同的框架模型去協同處理），而維護兩套計算框架模型和處理邏輯的代價始終是這個模型無法克服的痛點。雖然有各種在上層封裝抽象統一的SDK編程接口方案的存在，企圖通過一套代碼，翻譯執行的方式，降低在兩套計算框架模型上開發和維護代碼的代價，但實際效果往往並不如意，翻譯執行層的存在，並不能抹平兩種計算框架模型的差異，到頭來真正能復用的代碼邏輯並不多，簡單的說就是Lambada框架本身並不解決用戶真正的痛點，只是一種無奈之舉。

而Dataflow計算模型，則是希望從編程模型的源頭上就統一解決傳統的流式和批量兩種計算語意希望處理的問題。

和spark通過micro batch來處理streaming場景（如前，更準確的說法是無邊界數據集）的出發點不同，Dataflow認為batch的處理模式只是streamming處理模式的一個子集。在無邊界數據集的處理過程中，要及時產出數據結果，必然需要對需要處理的數據劃定一個窗口區間，從而對數據及時的進行分段處理和產出，各種處理模式（stream，micro batch，session， batch）只是窗口的大小不同，窗口的劃分方式不同，比如，Batch的處理模式就只是一個窗口區間涵蓋了整個有邊界的數據集這樣的一種特例而已。一個設計良好的能處理無邊界數據集的系統，完全能在準確性和正確性上做到和“Batch”系統一樣甚至應該更好。而不是傳統的認為batch框架的正確性更好，streaming框架顧及了實時性，正確性天然就做不好，必須和batch框架配合走Lambada模型來補足。

那麽無邊界數據集的處理過程中，大家認為天然做不好的點，或者說最難處理的點在哪裏，Dataflow是怎麽解決的呢。

這裏又要先說一下在Dataflow模型裏強調的兩個時間概念：Event time和 process time，Event time 事件時間就是數據真正發生的時間，比如用戶瀏覽了一個頁面，或者下了一個訂單等等，這時候通暢就會有一些數據會被生產出來，比如前者可能會產生一條用戶的瀏覽日誌，而process time則是這條日誌數據真正到達計算框架中被處理的時間點。現實情況下，由於各種原因，數據采集，傳輸到達處理系統的時間可能會有長短不同的延遲，在分布式應用場景環境下，不僅是延遲，數據亂序到達往往也是常態。這些問題，在有邊界數據的處理場景過程中往往並不存在，或者無關緊要。

基於這種無邊界數據集的特性，在Dataflow模型中，數據的處理過程被概括為以下4個方面的問題的解決：

– What results are being computed. ： 計算邏輯是什麽

– Where in event time they are being computed. ： 計算什麽時候（事件時間）的數據

– When in processing time they are materialized. ： 在什麽時候（處理時間）進行計算

– How earlier results relateto later refinements. ： 後續數據的處理結果如何影響之前的處理結果

清晰的定義這些問題，並針對性的在模型框架層面加以解決，正是Dataflow區別於其它流式計算模型的核心關鍵所在。通常的流式計算框架往往模糊或者無法有效的區別對待數據的事件時間和處理時間，對於第4個問題，也可能缺乏直接的支持。這些問題通常需要開發人員自行在代碼業務邏輯上想辦法解決，因而也就加大了這類數據處理業務的開發難度，甚至成為一個不可能完成的任務。

而更重要的是，針對同一或類似數據集，各種數據處理需求，其核心計算邏輯往往可能是一致的，比如計算活躍用戶數，核心計算邏輯就是一個去重邏輯。 但是根據應用目標場景，統計口徑可能各有不同，比如可能要求計算過去一個小時的活躍用戶，也可能是計算全天的累計的活躍用戶，可能基於實際時間計算也可能基於數據采集時間計算，可能要求更新歷史數據（有數據晚到），也可能處於效率，性能考慮，直接放棄晚到的數據。 Dataflow計算模型的目標是把上述4方面的問題，用明確的語意清晰的拆分出來，更好的模塊化，快速適應各種業務邏輯開發需求。

例如在 https://cloud.google.com/dataflow/blog/dataflow-beam-and-spark-comparison 一文中，就用實際的例子比較了dataflow和spark在處理這類數據業務邏輯時，所需要進行的開發工作，總體的意思就是用dataflow模型開發，代碼更簡潔更容易理解，開發效率更高，維護成本更低。不過，需要註意的是，spark2.0的structure streaming API也引入了和Dataflow類似的模型思想，這篇文章裏的很多比較已經不成立。

== 實現 ==

那麽Dataflow是如何解決上面4方面的問題的呢，基本上，是通過構建以下三個核心功能模型來做到的：

• 一個支持基於事件時間的窗口（window）模型，並提供簡易的API接口：支持固定窗口／滑動窗口／Session（以Key為維度，基於事件時間連續性進行劃分）等窗口模式
• 一個和數據自身特性綁定的計算結果輸出觸發模型，並提供靈活可描述的API接口
• 一個增量更新模型，可以將數據增量更新的能力融合進上述窗口和結果觸發模型中。

=== 窗口模型 ===

為了在計算框架級別實現基於事件時間的窗口模型，Dataflow系統中，將常見的流式計算框架中的［key，value］兩元組tuple形式的信息數據，變換成了［key,value, event time, window ］這樣的四元組模型，event time的引入原因顯而易見，必須要有相關載體承載這個信息（否則只能基於process time／batch time 劃分窗口），而window窗口標識信息的引入，個人認為，很重要的一個原因是要支持Session類型的窗口模型，而同時，要將流式和增量更新的支持融合進窗口的概念中，也勢必需要在數據中引入這樣一個顯示的窗口信息（否則，通常的做法就只能是用micro batch分組數據的方式，隱式的標識數據的窗口屬性）

在消息的四元組數據結構基礎上，Dataflow通過提供對消息進行窗口賦值，窗口合並，按key分組，按窗口分組等原子功能操作，來實現各種窗口模型。

=== 觸發模型 ===

多數的基於Process time的固定或滑動窗口模型，並沒有顯示的窗口計算結果觸發這樣一個概念的定義，因為不太需要，窗口的邊界時間點，也就是觸發結果輸出的時間點。而對於Dataflow來說，因為事件時間和處理時間的延遲，以及框架需要正確處理無序數據的需求，使得判斷窗口的邊界，觸發計算和結果的輸出變得困難起來。在這一點上，Dataflow部分借用了底層Millwheel提供的Low watermark低水位這樣一個概念來解決窗口邊界的判斷問題，當低水位對應的時間點超過設定的時間窗口邊界時間點時，完成窗口的計算和結果輸出。但是，低水位的概念理論上雖然是OK的，在實際場景中，通常是一個概率模型，並不能完全保證準確的判斷事件時間的延遲情況，而且有很多場合對窗口邊界的判斷，用戶自己有自己的需求。

因此，Dataflow提供了可自定義的窗口觸發模型，可以使用低水位做觸發，也可以使用比如：定時觸發，計數觸發，計量觸發，模式匹配觸發或其它外部觸發源，甚至各種觸發條件的邏輯運算組合等不同等機制來應對可能的需求。

=== 增量更新 ===

當窗口被觸發以後，對於後續晚到的數據，對已經觸發過的窗口，如何處理，Dataflow在框架層面也提供了直接的支持，基本上包括三種策略：

• 丟棄：一旦特定窗口觸發過，對應窗口的數據就丟棄，晚到的數據也丟棄。
• 累積：觸發過的窗口對應的數據保留（保留時間策略也可調整），晚到的數據更新對應窗口的輸出結果
• 累計並更正：和累積模式類似，區別在於會先對上一次窗口觸發的結果發送一個反相修正的信息，再輸出新的結果，便於有需要的下遊更正之前收到的信息。

== 相關研究，項目等 ==

=== spark 2.0 ===

Spark 2.0版本，新增的structured streaming API，針對原先的streaming編程接口DStream的問題進行了改進，Dstream的問題包括：

• 框架自身只能針對Batch time進行處理，很難處理event time，很難處理延遲，亂序的數據
• 流式和批量處理的API還是不完全一致，兩種使用場景中，程序代碼還是需要一定的轉換
• 端到端的數據容錯保障邏輯需要用戶自己小心構建，增量更新和持久化存儲等一致性問題處理難度較大

通過Structured Streaming API，Spark一方面支持了和Dataflow類似的概念，如Event time based的窗口策略，自定義的觸發邏輯，對輸出（sink）模塊的更新模式（追加，全量覆蓋，更新）的builtin支持，更加統一的處理無邊界數據和有邊界數據等。

總體看來，Spark 2.0的structured streaming 模型和Dataflow有異曲同工之處，設計的目標看起來很遠大，甚至給出了一份功能比較表格來證明其優越性

不過在2.0的版本所支持的類Dataflow模型的功能還相對簡單，比如session window，water flow等概念都還需要在2.1或者後續的版本中保證，也還不支持輸出的更新模式，追加模式更新只能支持無聚合操作的場景，還有各種功能還停留在設想階段，對於join等操作還有各種各樣的限制等等，這些部分和dataflow業已實現的功能還有較大的差距。

對於exactly once發送的保障，spark2.0要求外部數據源具備offset定位的能力，再加上snapshot等機制來實現，而dataflow是通過對消息在框架內部進行持久化來實現replay，不依賴外部數據源的能力。

另外，個人理解像 prefix integrity， Transactional sink等概念，實際上是對上下遊讀寫接口的一個封裝，幫用戶實現了一些業務邏輯（比如prefix integrity 的實現依托於於per key有序性的保證，這是由外部source源提供的保障，比如 file／kafka等；而Transactional sinks等則是比如對jdbc接口邏輯的封裝），整體上偏外圍功能一點，用這些特性來和其它框架比較不一定客觀，因為設計理念不太不一樣。Dataflow的模型設計中，用戶能更加細化的定義每個環節的步驟和設置，所以不會把一些邏輯替用戶實現，更多的是以模塊化的方式，留給用戶去自己選擇，而Structured steaming則把很多事情包辦了，定制的余地較小，靈活性應該會差一些，不過這也給程序的自動優化帶來了一些便利。當然，這是我個人初步粗淺的理解，不見得準確。

=== beam ===

Beam http://beam.incubator.apache.org/ 是一個由谷歌發起的apache 項目，目前還處於incubator狀態，基本來說就是實現dataflow編程模型的SDK項目，目標是提供一個high level的統一API編程接口，後端的執行引擎計劃對接spark／flink／cloud dataflow。目前的編程語言支持Java，計劃加入Python。這個項目的前景如何，不太好說，單就適配各個後端的角度來說，就spark後端來說，在spark 1.x時代，這種high level的編程模型抽象是對spark編程模型的一種add on，有一定的附加價值，但是按照spark 2.0 structured streaming的發展路線來說，這一層抽象就稍微顯得有些多余了。而基於Java的語法，在表達的簡潔性上，相比Scala也會帶來一些額外的代價。

== 參考資料 ==

• dataflow論文 ： The Dataflow Model- A Practical Approach to Balancing Correctness, Latency, and Cost in Massive-Scale, Unbounded, Out-of-Order Data Processing
• https://cloud.google.com/dataflow/blog/dataflow-beam-and-spark-comparison
• https://spark-summit.org/2016/events/structuring-spark-dataframes-datasets-and-streaming/
• https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html
• https://databricks.com/blog/2016/07/28/continuous-applications-evolving-streaming-in-apache-spark-2-0.html
• https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-101
• https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-102
• https://cloud.google.com/dataflow
• http://beam.incubator.apache.org/

Dataflow編程模型和spark streaming結合