實現批處理的技術許許多多，從各種關係型資料庫的sql處理，到大資料領域的MapReduce，Hive，Spark等等。這些都是處理有限資料流的經典方式。而Flink專注的是無限流處理，那麼他是怎麼做到批處理的呢？

無限流處理：輸入資料沒有盡頭；資料處理從當前或者過去的某一個時間 點開始，持續不停地進行

另一種處理形式叫作有限流處理，即從某一個時間點開始處理資料，然後在另一個時間點結束。輸入資料可能本身是有限的（即輸入資料集並不會隨著時間增長），也可能出於分析的目的被人為地設定為有限集（即只分析某一個時間段內的事件）。

顯然，有限流處理是無限流處理的一種特殊情況，它只不過在某個時間點停止而已。此外，如果計算結果不在執行過程中連續生成，而僅在末尾處生成一次，那就是批處理（分批處理資料）。

批處理是流處理的一種非常特殊的情況。在流處理中，我們為資料定義滑 動視窗或滾動視窗，並且在每次視窗滑動或滾動時生成結果。批處理則不同，我們定義一個全域性視窗，所有的記錄都屬於同一個視窗。舉例來說， 以下程式碼表示一個簡單的Flink 程式，它負責每小時對某網站的訪問者計數，並按照地區分組。

val counts = visits   
.keyBy("region")   
.timeWindow(Time.hours(1))   
.sum("visits")

如果知道輸入資料是有限的，則可以通過以下程式碼實現批處理。

val counts = visits   
.keyBy("region")   
.window(GlobalWindows.create)   
.trigger(EndOfTimeTrigger.create)   
.sum("visits")
 

Flink 的不尋常之處在於，它既可以將資料當作無限流來處理，也可以將它當作有限流來處理。Flink 的 DataSet API 就是專為批處理而生的，如下所示。

val counts = visits   
.groupBy("region")   
.sum("visits")

如果輸入資料是有限的，那麼以上程式碼的執行結果將與前一段程式碼的相同， 但是它對於習慣使用批處理器的程式設計師來說更友好。

Fink批處理模型

Flink 通過一個底層引擎同時支援流處理和批處理

在流處理引擎之上，Flink 有以下機制：

• 檢查點機制和狀態機制：用於實現容錯、有狀態的處理；

• 水印機制：用於實現事件時鐘；

• 視窗和觸發器：用於限制計算範圍，並定義呈現結果的時間。

在同一個流處理引擎之上，Flink 還存在另一套機制，用於實現高效的批處理。

• 用於排程和恢復的回溯法：由 Microsoft Dryad 引入，現在幾乎用於所有批處理器；
• 用於雜湊和排序的特殊記憶體資料結構：可以在需要時，將一部分資料從記憶體溢位到硬碟上；
• 優化器：儘可能地縮短生成結果的時間。

兩套機制分別對應各自的API（DataStream API 和 DataSet API）；在建立 Flink 作業時，並不能通過將兩者混合在一起來同時 利用 Flink 的所有功能。

在最新的版本中，Flink 支援兩種關係型的 API，Table API 和 SQL。這兩個 API 都是批處理和流處理統一的 API，這意味著在無邊界的實時資料流和有邊界的歷史記錄資料流上，關係型 API 會以相同的語義執行查詢，併產生相同的結果。Table API 和 SQL 藉助了 Apache Calcite 來進行查詢的解析，校驗以及優化。它們可以與 DataStream 和 DataSet API 無縫整合，並支援使用者自定義的標量函式，聚合函式以及表值函式。

Table API / SQL 正在以流批統一的方式成為分析型用例的主要 API。

DataStream API 是資料驅動應用程式和資料管道的主要API。

從長遠來看，DataStream API應該通過有界資料流完全包含DataSet API。

Flink批處理效能

MapReduce、Tez、Spark 和 Flink 在執行純批處理任務時的效能比較。測試的批處理任務是 TeraSort 和分散式雜湊連線。

第一個任務是 TeraSort，即測量為 1TB 資料排序所用的時間。

TeraSort 本質上是分散式排序問題，它由以下幾個階 段組成：

(1) 讀取階段：從 HDFS 檔案中讀取資料分割槽；

(2) 本地排序階段：對上述分割槽進行部分排序；

(3) 混洗階段：將資料按照 key 重新分佈到處理節點上；

(4) 終排序階段：生成排序輸出；

(5) 寫入階段：將排序後的分割槽寫入 HDFS 檔案。

Hadoop 發行版包含對 TeraSort 的實現，同樣的實現也可以用於 Tez，因為 Tez 可以執行通過MapReduce API 編寫的程式。Spark 和 Flink 的 TeraSort 實現由 Dongwon Kim 提供.用來測量的叢集由 42 臺機器組成，每臺機器 包含 12 個 CPU 核心、24GB 記憶體，以及 6 塊硬碟。

結果顯示，Flink 的排序時間比其他所有系統都少。 MapReduce 用了2157 秒，Tez 用了1887 秒，Spark 用了2171 秒，Flink 則 只用了 1480 秒。

第二個任務是一個大資料集（240GB）和一個小資料集（256MB）之間的分散式雜湊連線。結果顯示，Flink 仍然是速度最快的系統，它所用的時間分別是 Tez 和 Spark 的 1/2 和 1/4.

產生以上結果的總體原因是，Flink 的執行過程是基於流的，這意味著各個處理階段有更多的重疊，並且混洗操作是流水線式的，因此磁碟訪問操作更少。相反，MapReduce、Tez 和 Spark 是基於批的，這意味著資料在通過網路傳輸之前必須先被寫入磁碟。該測試說明，在使用Flink 時，系統空閒時間和磁碟訪問操作更少。

值得一提的是，效能測試結果中的原始數值可能會因叢集設定、配置和軟體版本而異。

因此，Flink 可以用同一個資料處理框架來處理無限資料流和有限資料流，並且不會犧牲效能。

更多Flink相關文章：

穿梭時空的實時計算框架——Flink對時間的處理

Flink快速入門--安裝與示例執行

大資料實時處理的王者-Flink

Flink,Storm,SparkStreaming效能對比

更多實時計算，Flink,Kafka的技術文章歡迎關注實時流式計算

相關推薦