引言

隨著大資料時代的來臨，大資料產品層出不窮。我們最近也對一款業內非常火的大資料產品 - Apache Flink做了調研，今天與大家分享一下。Apache Flink(以下簡稱flink) 是一個旨在提供‘一站式’ 的分散式開源資料處理框架。是不是聽起來很像spark？沒錯，兩者都希望提供一個統一功能的計算平臺給使用者。雖然目標非常類似，但是flink在實現上和spark存在著很大的區別，flink是一個面向流的處理框架，輸入在flink中是無界的，流資料是flink中的頭等公民。說到這裡，大家一定覺得flink和storm有幾分相似，確實是這樣。那麼有spark和storm這樣成熟的計算框架存在，為什麼flink還能佔有一席之地呢？今天我們就從流處理的角度將flink和這兩個框架進行一些分析和比較。1本文的流框架基於的實現方式本文涉及的流框架基於的實現方式分為兩大類。第一類是Native Streaming，這類引擎中所有的data在到來的時候就會被立即處理，一條接著一條（HINT： 狹隘的來說是一條接著一條，但流引擎有時會為提高效能快取一小部分data然後一次性處理），其中的代表就是storm和flink。第二種則是基於Micro-batch，資料流被切分為一個一個小的批次， 然後再逐個被引擎處理。這些batch一般是以時間為單位進行切分，單位一般是‘秒‘，其中的典型代表則是spark了，不論是老的spark DStream還是2.0以後推出的spark structured streaming都是這樣的處理機制；另外一個基於Micro-batch實現的就是storm trident，它是對storm的更高層的抽象，因為以batch為單位，所以storm trident的一些處理變的簡單且高效。

2流框架比較的關鍵指標

從流處理的角度將flink與spark和storm這兩個框架進行比較，會主要關注以下幾點，後續的對比也主要基於這幾點展開：

功能性（Functionality）- 是否能很好解決流處理功能上的痛點 , 比如event time和out of order data。容錯性（Fault Tolerance）- 在failure之後能否恢復到故障之前的狀態，並輸出一致的結果；此外容錯的代價也是越低越好，因為其直接影響效能。

吞吐量(throughputs)& 延時(latency)- 效能相關的指標，高吞吐和低延遲某種意義上是不可兼得的，但好的流引擎應能兼顧高吞吐&低延時。

功能性（Functionality）01.Event time&Window Operation①Event time

event time - 指資料或者事件真正發生時間 , 比如使用者點選網頁時產生一條點選事件的資料，點選時間就是這條資料固有的event time。

processing time - 指計算框架處理這條資料的時間。

（具體關於時間的定義可以參看flink文件http://t.cn/RaTnsdy。）spark DStream和storm 1.0以前版本往往都折中地使用processing time來近似地實現event time相關的業務。顯然，使用processing time模擬event time必然會產生一些誤差， 特別是在產生資料堆積的時候，誤差則更明顯，甚至導致計算結果不可用。在使用event time時，自然而然需要解決由網路延遲等因素導致的遲到或者亂序資料的問題。為了解決這個問題， spark、storm及flink都參考streaming 102(http://t.cn/RbQCUmJ)引入了watermark和lateness的概念。watermark: 是引擎處理事件的時間進度，代表一種狀態，一般隨著資料中的event time的增長而增長。比如 watermark(t)代表整個流的event time處理進度已經到達t， 時間是有序的，那麼streaming不應該會再收到timestamp t’ < t的資料，而只會接受到timestamp t’ >= t的資料。 如果收到一條timestamp t’ < t的資料， 那麼就說明這條資料是遲到的。lateness: 表示可以容忍遲到的程度，在lateness可容忍範圍內的資料還會參與計算，超過的會被丟棄。②Window Operation下面主要比較在使用window的操作中，spark structured streaming 和flink對event time處理機制的不同。flink

首先，我們結合圖來看flink， 時間軸從左往右增大。當watermark WM處於時 間視窗區間內時，即WM ∈ [start, end] , event time落在視窗範圍內的任何亂序資料都會被接受；隨著WM的增長並超過了視窗的結束時間，但還未超過可容忍的lateness時間範圍，即WM ∈ (window_end,window_end+ lateness]， 這時亂序資料仍然可以被接受； 只有當WM超過 window_end+lateness, 即WM ∈ (window_end+ lateness, ∞)， 遲到的資料將會被丟棄。

fiink中watermark的計算也比較靈活，可以選擇build-in的（如最大時間戳），也可以通過繼承介面自定義實現。此外，使用者可以選擇週期性更新或者事件觸發更新watermark。spark

首先,spark中watermark是通過上一個batch最大的timestamp再減去lateness得到的，即watermark = Max(last batch timestamps) - lateness。當資料的event time大於watermark時，資料會被接受，否則不論這條資料屬於哪個視窗都會被丟棄。細節請參考spark文件（http://t.cn/RaTnvVQ）。下面來比較一下兩者實現細節上的不同：①lateness定義: 在spark中，遲到被定義為data的event time和watermark的比較結果，當data的event time < watermark時，data被丟棄；flink中只有在watermark > window_end + lateness的時候，data才會被丟棄。②watermark更新: spark中watermark是上個batch中的max event time，存在延遲；而在flink中是可以做到每條資料同步更新watermark。③window觸發: flink中window計算會觸發一次或多次，第一次在watermark >= window_end後立刻觸發（main fire），接著會在遲到資料到來後進行增量觸發。spark只會在watermark（包含lateness）過了window_end之後才會觸發，雖然計算結果一次性正確，但觸發比flink起碼多了一個lateness的延遲。上面三點可見flink在設計event time處理模型還是較優的：watermark的計算實時性高，輸出延遲低，而且接受遲到資料沒有spark那麼受限。不光如此，flink提供的window programming模型非常的靈活，不但支援spark、storm沒有的session window，而且只要實現其提供的WindowAssigner、Trigger、Evictor就能創造出符合自身業務邏輯的window，功能非常強大。02.SQL API目前flink相比spark，對streaming sql的支援還是比較初級的。在當前最新1.2版本中，僅支援Selection、Projection、Union、Tumble，不支援Aggregation、 Join、Top N、 Sort。計劃中1.3版本將支援 Window Aggregation(sum、max、 min、avg), 但依然不支援Distinct。相比flink，當前最新版本的spark structured streaming僅僅不支援Top N、Distinct。03.Kafka Source Integrationflink對於kafka的相容性非常好，支援kafka 0.8、0.9、0.10；相反，spark structured streaming只支援kafka0.10或更高版本。04.Interoperation with Static Dataspark底層對static batch data和streaming data有共同的rdd抽象，完美相容互操作。而flink中DataSet 和 DataStream是完全獨立的，不可以直接互動。此外，flink還可以執行storm的topology，帶來較強的移植性。另外一個有趣的功能是可以自由調整job latency and throughputs的取捨關係，比如需要high throughputs的程式可以犧牲latency來獲得更大的throughputs。

容錯性（Fault Tolerance）spark依賴checkpoint機制來進行容錯，只要batch執行到doCheckpoint操作前掛了，那麼該batch就會被完整的重新計算。spark可以保證計算過程的exactly once（不包含sink的exactly once）。storm的容錯通過ack機制實現，每個bolt或spout處理完成一條data後會傳送一條ack訊息給acker bolt。當該條data被所有節點都處理過後，它會收到來自所有節點ack， 這樣一條data處理就是成功的。storm可以保證資料不丟失，但是隻能達到at least once語義。此外，因為需要每條data都做ack，所以容錯的開銷很大。storm trident是基於microbatched實現了exactly once語義。flink使用Chandy-Chandy-Lamport Algorithm 來做Asynchronous Distributed Snapshots（非同步分散式快照），其本質也是checkpoint。如下圖，flink定時往流裡插入一個barrier（隔欄），這些barriers把資料分割成若干個小的部分，當barrier流到某個operator時，operator立即會對barrier對應的一小部分資料做checkpoint並且把barrier傳給下游（checkpoint操作是非同步的，並不會打斷資料的處理），直到所有的sink operator做完自己checkpoint後，一個完整的checkpoint才算完成。當出現failure時，flink會從最新完整的checkpoint點開始恢復。

flink的checkpoint機制非常輕量，barrier不會打斷streaming的流動，而且做checkpoint操作也是非同步的。其次，相比storm需要ack每條data，flink做的是small batch的checkpoint，容錯的代價相對要低很多。最重要的是flink的checkpoint機制能保證exactly once。

吞吐量和延遲（Throughputs& Latency）01.吞吐量（throughputs）spark是mirco-batch級別的計算，各種優化做的也很好，它的throughputs是最大的。但是需要提一下，有狀態計算（如updateStateByKey運算元）需要通過額外的rdd來維護狀態，導致開銷較大，對吞吐量影響也較大。storm的容錯機制需要對每條data進行ack，因此容錯開銷對throughputs影響巨大，throughputs下降甚至可以達到70%。storm trident是基於micro-batch實現的，throughput中等。flink的容錯機制較為輕量，對throughputs影響較小，而且擁有圖和排程上的一些優化機制，使得flink可以達到很高 throughputs。下圖是flink官網給出的storm和flink的benchmark，我們可以看出storm在開啟ack容錯機制後，throughputs下降非常明顯。而flink在開啟checkpoint和關閉的情況下throughputs變化不大，說明flink的容錯機制確實代價不高。對比官網的benchmark，我們也進行了throughputs的測試，實測結果是flink throughputs是storm的3.5倍，而且在解除了kafka叢集和flink叢集的頻寬瓶頸後，flink自身又提高了1.6倍。

02.延遲（latency）spark基於micro-batch實現，提高了throughputs，但是付出了latency的代價。一般spark的latency是秒級別的。storm是native streaming實現，可以輕鬆的達到幾十毫秒級別的latency，在幾款框架中它的latency是最低的。storm trident是基於micro-batch實現的，latency較高。flink也是native streaming實現，也可以達到百毫秒級別的latency。下圖是flink官網給出的和storm的latency對比benchmark。storm可以達到平均5毫秒以內的latency，而flink的平均latency也在30毫秒以內。兩者的99%的data都在55毫秒latency內處理完成，表現都很優秀。

3總 結

綜合對比spark、storm和flink的功能、容錯和效能（總結如下圖）

不難發現， flink是一個設計良好的框架，它不但功能強大，而且效能出色。此外它還有一些比較好設計，比如優秀的記憶體管理和流控。但是，flink目前成熟度較低，還存在著不少問題，比如 SQL支援比較初級；無法像storm一樣在不停止任務的情況下動態調整資源；不能像spark一樣提供很好的streaming和static data的互動操作等。對於這些問題，flink社群還在積極的跟進，相信在更多公司和貢獻者的共同努力下，flink會發展的越來越好。