Flink常見的關鍵技術與特性詳解
Flink項目是大數據處理領域最近冉冉升起的一顆新星，其不同于其他大數據項目的諸多特性吸引了越來越多的人關注Flink項目。本文將深入分析Flink一些關鍵的技術與特性，希望能夠幫助讀者對Flink有更加深入的了解，對其他大數據系統的開發者也能有所裨益。
注：本文假設讀者對MapReduce，Spark及Storm等大數據處理系統有基本了解，同時熟悉流處理與批處理的基本概念。36大數據（http://www.36dsj.com/）
Flink簡介
Flink的核心是一個流式的數據流執行引擎，其針對數據流的分布式計算提供了數據分布，數據通信以及容錯機制等功能。基于流執行引擎，Flink提供了諸多更高抽象層的API以方便用戶編寫分布式任務：
1. DataSet API, 對靜態數據進行批處理操作，將靜態數據抽象成分布式的數據集，用戶可以方便的采用Flink提供的各種操作符對分布式數據集進行各種操作，支持Java，Scala和Python。
2. DataStream API，對數據流進行流處理操作，將流式的數據抽象成分布式的數據流，用戶可以方便的采用Flink提供的各種操作符對分布式數據流進行各種操作，支持Java和Scala。
3. Table API，對結構化數據進行查詢操作，將結構化數據抽象成關系表，并通過Flink提供的類SQL的DSL對關系表進行各種查詢操作，支持Java和Scala。
此外，Flink還針對特定的應用領域提供了領域庫，例如：
1. Flink ML，Flink的機器學習庫，提供了機器學習Pipelines API以及很多的機器學習算法實現。
2. Gelly，Flink的圖計算庫，提供了圖計算的相關API以及很多的圖計算算法實現。
Flink的技術棧如下圖所示：36大數據（http://www.36dsj.com/）
圖1 Flink技術棧
此外，Flink也可以方便地和其他的Hadoop生態圈的項目集成，例如，Flink可以讀取存儲在HDFS或HBase中的靜態數據，以Kafka作為流式的數據源，直接重用MapReduce/Storm代碼，或是通過YARN申請集群資源等等。
統一的批處理與流處理系統
在大數據處理領域，批處理任務與流處理任務一般被認為是兩種不同的任務，一個大數據項目一般會被設計為只能處理其中一種任務，例如Apache Storm，Apache Smaza只支持流處理任務，而Aapche MapReduce， Apache Tez，Apache Spark只支持批處理任務。
Spark Streaming是Apache Spark之上支持流處理任務的子系統，看似一個特例，實則不然。Spark Streaming采用了一種micro-batch的架構，即將輸入的數據流切分成細粒度的batch數據，對于每一個batch數據，以此為輸入提交一個批處理Spark任務，所以Spark Streaming本質上還是基于Spark批處理系統對流式數據進行處理，和Apache Storm，Apache Smaza等完全流式的數據處理方式完全不同。Flink能夠同時處理批處理任務與流處理任務，其靈活的執行引擎支持完全原生的批量的數據處理和流式的數據處理。
在執行引擎這一層，流處理系統與批處理系統***的不同在于節點間數據傳輸的方式。對于一個流處理系統，其節點間數據傳輸的標準模型是：當一條數據被處理完成后，序列化到緩存中，然后立刻通過網絡傳輸到下一個節點，由下一個節點繼續處理。而對于一個批處理系統，其節點間數據傳輸的標準模型是：當一條數據被處理完成后，序列化到緩存中，并不會立刻通過網絡傳輸到下一個節點，當緩存寫滿，就持久化到本地硬盤上，當所有數據都被處理完成后，才開始將處理后的數據通過網絡傳輸到下一個節點。36大數據（http://www.36dsj.com/）
這兩種數據傳輸模式是兩個極端，對應的是流處理系統對低延遲的要求和批處理系統對高吞吐量的要求。Flink的執行引擎采用了一種十分靈活的方式，同時支持了這兩種數據傳輸模型。Flink以固定的緩存塊為單位進行網絡數據傳輸，用戶可以通過緩存塊超時值指定緩存塊的傳輸時機。如果緩存塊的超時值為0，則Flink的數據傳輸方式類似上面提到的流處理系統的標準模型，此時系統可以獲得***的處理延遲。
如果緩存塊的超時值為***大，則Flink的數據傳輸方式類似上面提到的批處理系統的標準模型，此時系統可以獲得***的處理吞吐量。同時緩存塊的超時值也可以設置為0到***大之間的任意值。緩存塊的超時閾值越小，則Flink流處理執行引擎的數據處理延遲越低，但吞吐量也會越低，緩存塊的超時閾值越大時，則反之。通過調整緩存塊的超時閾值，用戶可根據自己的需要靈活的權衡Flink的延遲和吞吐量。
圖2 Flink執行引擎數據傳輸模式
在統一的流式執行引擎的基礎上，Flink同時支持了流處理系統與批處理系統，并且保證了其流處理系統與批處理系統的性能(延遲，吞吐量等)，相對于其他原生的流處理與批處理系統，并沒有因為統一的執行引擎而受到影響。用戶可以在Flink上同時執行批處理任務與流處理任務，這大大減輕了用戶安裝，部署，監控，維護等成本。36大數據（http://www.36dsj.com/）
Flink流處理的容錯機制
對于一個分布式系統來說，單個進程或是節點崩潰導致整個Job失敗是經常發生的事情，在異常發生的時候不會丟失用戶數據，并能夠自動恢復是分布式系統的需要支持的特性之一。本節主要介紹Flink流處理系統對于任務級別的容錯機制。
批處理系統比較容易實現容錯機制，由于文件可以重復訪問，當某個任務失敗后，重啟該任務即可。但是在流處理系統中，由于數據源是***的數據流，一個流處理任務甚至可能會執行幾個月，將所有數據緩存或是持久化，留待以后重復訪問基本上是不可行的。Flink基于分布式快照與可部分重發的數據源實現了容錯，用戶可自定義對整個Job進行快照的時間間隔，當出現任務失敗時，Flink將整個Job恢復到最近一次快照的狀態，并從數據源重發快照之后的數據。
Flink的分布式快照的實現借鑒了Chandy和Lamport在1985年發表的一篇關于分布式快照的論文，其實現的主要思想如下：
按照用戶自定義的分布式快照間隔時間，Flink會在定時在所有數據源中插入一種特殊的快照標記消息，這些快照標記消息和其他消息一樣在DAG中流動，但是不會被用戶定義的業務邏輯所處理，每一個快照標記消息都將其所在的數據流分成兩部分：本次快照數據和下次快照數據。36大數據（http://www.36dsj.com/）
圖3 Flink包含快照標記消息的消息流
快照標記消息沿著DAG流經各個操作符，當操作符處理到快照標記消息時，會對自己的狀態進行快照，并存儲起來。當一個操作符有多個輸入的時候，Flink會將先抵達的快照標記消息及其之后的消息緩存起來，當所有的輸入中對應該次快照的快照標記消息全部抵達后，操作符對自己的狀態快照并存儲，之后處理所有快照標記消息之后的已緩存消息。操作符對自己的狀態快照并存儲可以是異步與增量的操作，并不需要阻塞消息的處理。分布式快照的流程如下圖所示：
圖4 Flink分布式快照流程圖
當所有的Data Sink(終點操作符)都收到快照標記信息并對自己的狀態快照和存儲后，整個分布式快照就完成了，同時通知數據源釋放該快照標記消息之前的所有消息。若之后發生節點崩潰等異常情況時，只需要恢復之前存儲的分布式快照狀態，并從數據源重發該快照以后的消息就可以了。
Exactly-Once是流處理系統需要支持的一個非常重要的特性，它保證每一條消息被流處理系統處理一次，且僅被處理一次，許多流處理任務的業務邏輯都依賴于Exactly-Once特性。相對于At-Least-Once或是At-Most-Once, Exactly-Once特性對流處理系統的要求更嚴格，實現也更困難。Flink基于分布式快照實現了Exactly-Once特性。36大數據（http://www.36dsj.com/）
相對于其他流處理系統的容錯方案，Flink基于分布式快照的方案在功能和性能方面都具有很多優點，包括：
1. 低延遲。由于操作符狀態的存儲可以是異步的，所以進行快照的過程基本上不會阻塞消息的處理，對消息的延遲不會產生負面的影響。
2. 高吞吐量。當操作符狀態較少時，對吞吐量基本沒有影響。當操作符狀態較多時，相對于其他的容錯機制，分布式快照的時間間隔是用戶自定義的，所以用戶可以權衡錯誤恢復時間和吞吐量的要求，調整分布式快照的時間間隔。
3. 與業務邏輯的隔離。Flink的分布式快照機制與用戶的業務邏輯是完全隔離的，用戶的業務邏輯不會依賴或是對分布式快照產生任何影響。
4. 錯誤恢復代價。分布式快照的時間間隔越短，錯誤恢復的時間越少，與吞吐量負相關。
Flink流處理的時間窗口
對于流處理系統來說，流入的消息是***的，所以對于聚合或是連接等操作，流處理系統需要對流入的消息進行分段，然后基于每一段數據進行聚合或是連接等操作。消息的分段即稱為窗口，流處理系統支持的窗口有很多類型，最常見的就是時間窗口，基于時間間隔對消息進行分段處理。本節主要介紹Flink流處理系統支持的各種時間窗口。
對于目前大部分流處理系統來說，時間窗口一般是根據Task所在節點的本地時鐘來進行切分，這種方式實現起來比較容易，不會阻塞消息處理。但是可能無法滿足某些應用的要求，例如：
1. 消息本身帶有時間戳，用戶希望按照消息本身的時間特性進行分段處理。
2. 由于不同節點的時鐘可能不同，以及消息在流經各個節點時延遲不同，在某個節點屬于同一個時間窗口處理的消息，流到下一個節點時可能被切分到不同的時間窗口中，從而產生不符合預期的結果。
Flink支持三種類型的時間窗口，分別適用于用戶對于時間窗口不同類型的要求：
1. Operator Time。根據Task所在節點的本地時鐘來進行切分的時間窗口。
2. Event Time。消息自帶時間戳，根據消息的時間戳進行處理，確保時間戳在同一個時間窗口的所有消息一定會被正確處理。由于消息可能是亂序流入Task的，所以Task需要緩存當前時間窗口消息處理的狀態，直到確認屬于該時間窗口的所有消息都被處理后，才可以釋放其狀態。如果亂序的消息延遲很高的話，會影響分布式系統的吞吐量和延遲。
3. Ingress Time。有時消息本身并不帶有時間戳信息，但用戶依然希望按照消息而不是節點時鐘劃分時間窗口(例如，避免上面提到的第二個問題)。此時可以在消息源流入Flink流處理系統時，自動生成增量的時間戳賦予消息，之后處理的流程與Event Time相同。Ingress Time可以看成是Event Time的一個特例，由于其在消息源處時間戳一定是有序的，所以在流處理系統中，相對于Event Time，其亂序的消息延遲不會很高，因此對Flink分布式系統的吞吐量和延遲的影響也會更小。
Event Time時間窗口的實現
Flink借鑒了Google的MillWheel項目，通過WaterMark來支持基于Event Time時間窗口。
當操作符通過基于Event Time的時間窗口來處理數據時，它必須在確定所有屬于該時間窗口的消息全部流入此操作符后，才能開始處理數據。但是由于消息可能是亂序的，所以操作符無法直接確認何時所有屬于該時間窗口的消息全部流入此操作符。36大數據（http://www.36dsj.com/）
WaterMark包含一個時間戳，Flink使用WaterMark標記所有小于該時間戳的消息都已流入，Flink的數據源在確認所有小于某個時間戳的消息都已輸出到Flink流處理系統后，會生成一個包含該時間戳的WaterMark，插入到消息流中輸出到Flink流處理系統中，Flink操作符按照時間窗口緩存所有流入的消息，當操作符處理到WaterMark時，它對所有小于該WaterMark時間戳的時間窗口的數據進行處理并發送到下一個操作符節點，然后也將WaterMark發送到下一個操作符節點。
為了保證能夠處理所有屬于某個時間窗口的消息，操作符必須等到大于這個時間窗口的WaterMark之后，才能開始對該時間窗口的消息進行處理，相對于基于Operator Time的時間窗口，Flink需要占用更多的內存，且會直接影響消息處理的延遲時間。對此，一個可能的優化措施是，對于聚合類的操作符，可能可以提前對部分消息進行聚合操作，當有屬于該時間窗口的新消息流入時，基于之前的部分聚合結果繼續計算，這樣的話，只需緩存中間計算結果即可，無需緩存該時間窗口的所有消息。
對于基于Event Time時間窗口的操作符來說，流入WaterMark的時間戳與當前節點的時鐘一致是最簡單理想的狀況了，但是在實際環境中是不可能的，由于消息的亂序以及前面節點處理效率的不同，總是會有某些消息流入時間大于其本身的時間戳，真實WaterMark時間戳與理想情況下WaterMark時間戳的差別稱為Time Skew，如下圖所示：
圖5 WaterMark的Time Skew圖
Time Skew決定了該WaterMark與上一個WaterMark之間的時間窗口所有數據需要緩存的時間，Time Skew時間越長，該時間窗口數據的延遲越長，占用內存的時間也越長，同時會對流處理系統的吞吐量產生負面影響。
基于時間戳的排序
在流處理系統中，由于流入的消息是***的，所以對消息進行排序基本上被認為是不可行的。但是在Flink流處理系統中，基于WaterMark，Flink實現了基于時間戳的全局排序。
Flink基于時間戳進行排序的實現思路如下：排序操作符緩存所有流入的消息，當其接收到WaterMark時，對時間戳小于該WaterMark的消息進行排序，并發送到下一個節點，在此排序操作符中釋放所有時間戳小于該WaterMark的消息，繼續緩存流入的消息，等待下一個WaterMark觸發下一次排序。
由于WaterMark保證了其之后不會出現時間戳比它小的消息，所以可以保證排序的正確性。需要注意的是，如果排序操作符有多個節點，只能保證每個節點的流出消息是有序的，節點之間的消息不能保證有序，要實現全局有序，則只能有一個排序操作符節點。
通過支持基于Event Time的消息處理，Flink擴展了其流處理系統的應用范圍，使得更多的流處理任務可以通過Flink來執行。
定制的內存管理
略，請參考上篇文章：脫離JVM? Hadoop生態圈的掙扎與演化
總結
本文主要介紹了Flink項目的一些關鍵特性，Flink是一個擁有諸多特色的項目，包括其統一的批處理和流處理執行引擎，通用大數據計算框架與傳統數據庫系統的技術結合，以及流處理系統的諸多技術創新等，因為篇幅有限，Flink還有一些其他很有意思的特性沒有詳細介紹，比如DataSet API級別的執行計劃優化器，原生的迭代操作符等，感興趣的讀者可以通過Flink的官網了解更多Flink的詳細內容。希望通過本文的介紹能夠讓讀者對Flink項目能有更多的了解，也讓更多的人使用甚至參與到Flink項目中去。
