[[312719]]

實現(xiàn)批處理的技術(shù)許許多多，從各種關(guān)系型數(shù)據(jù)庫的sql處理，到大數(shù)據(jù)領(lǐng)域的MapReduce，Hive，Spark等等。這些都是處理有限數(shù)據(jù)流的經(jīng)典方式。而Flink專注的是無限流處理，那么他是怎么做到批處理的呢?

無限流處理：輸入數(shù)據(jù)沒有盡頭;數(shù)據(jù)處理從當前或者過去的某一個時間 點開始，持續(xù)不停地進行

另一種處理形式叫作有限流處理，即從某一個時間點開始處理數(shù)據(jù)，然后在另一個時間點結(jié)束。輸入數(shù)據(jù)可能本身是有限的(即輸入數(shù)據(jù)集并不會隨著時間增長)，也可能出于分析的目的被人為地設(shè)定為有限集(即只分析某一個時間段內(nèi)的事件)。

顯然，有限流處理是無限流處理的一種特殊情況，它只不過在某個時間點停止而已。此外，如果計算結(jié)果不在執(zhí)行過程中連續(xù)生成，而僅在末尾處生成一次，那就是批處理(分批處理數(shù)據(jù))。

批處理是流處理的一種非常特殊的情況。在流處理中，我們?yōu)閿?shù)據(jù)定義滑 動窗口或滾動窗口，并且在每次窗口滑動或滾動時生成結(jié)果。批處理則不同，我們定義一個全局窗口，所有的記錄都屬于同一個窗口。舉例來說， 以下代碼表示一個簡單的Flink 程序，它負責(zé)每小時對某網(wǎng)站的訪問者計數(shù)，并按照地區(qū)分組。

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

如果知道輸入數(shù)據(jù)是有限的，則可以通過以下代碼實現(xiàn)批處理。

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

Flink 的不尋常之處在于，它既可以將數(shù)據(jù)當作無限流來處理，也可以將它當作有限流來處理。Flink 的 DataSet API 就是專為批處理而生的，如下所示。

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

如果輸入數(shù)據(jù)是有限的，那么以上代碼的運行結(jié)果將與前一段代碼的相同， 但是它對于習(xí)慣使用批處理器的程序員來說更友好。

Fink批處理模型

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

在流處理引擎之上，F(xiàn)link 有以下機制：

• 檢查點機制和狀態(tài)機制：用于實現(xiàn)容錯、有狀態(tài)的處理;
• 水印機制：用于實現(xiàn)事件時鐘;
• 窗口和觸發(fā)器：用于限制計算范圍，并定義呈現(xiàn)結(jié)果的時間。

在同一個流處理引擎之上，F(xiàn)link 還存在另一套機制，用于實現(xiàn)高效的批處理。

• 用于調(diào)度和恢復(fù)的回溯法：由 Microsoft Dryad 引入，現(xiàn)在幾乎用于所有批處理器;
• 用于散列和排序的特殊內(nèi)存數(shù)據(jù)結(jié)構(gòu)：可以在需要時，將一部分數(shù)據(jù)從內(nèi)存溢出到硬盤上;
• 優(yōu)化器：盡可能地縮短生成結(jié)果的時間。

兩套機制分別對應(yīng)各自的API(DataStream API 和 DataSet API);在創(chuàng)建 Flink 作業(yè)時，并不能通過將兩者混合在一起來同時 利用 Flink 的所有功能。

在最新的版本中，F(xiàn)link 支持兩種關(guān)系型的 API，Table API 和 SQL。這兩個 API 都是批處理和流處理統(tǒng)一的 API，這意味著在無邊界的實時數(shù)據(jù)流和有邊界的歷史記錄數(shù)據(jù)流上，關(guān)系型 API 會以相同的語義執(zhí)行查詢，并產(chǎn)生相同的結(jié)果。Table API 和 SQL 借助了 Apache Calcite 來進行查詢的解析，校驗以及優(yōu)化。它們可以與 DataStream 和 DataSet API 無縫集成，并支持用戶自定義的標量函數(shù)，聚合函數(shù)以及表值函數(shù)。

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

DataStream API 是數(shù)據(jù)驅(qū)動應(yīng)用程序和數(shù)據(jù)管道的主要API。

從長遠來看，DataStream API應(yīng)該通過有界數(shù)據(jù)流完全包含DataSet API。

Flink批處理性能

MapReduce、Tez、Spark 和 Flink 在執(zhí)行純批處理任務(wù)時的性能比較。測試的批處理任務(wù)是 TeraSort 和分布式散列連接。

第一個任務(wù)是 TeraSort，即測量為 1TB 數(shù)據(jù)排序所用的時間。

TeraSort 本質(zhì)上是分布式排序問題，它由以下幾個階 段組成：

(1) 讀取階段：從 HDFS 文件中讀取數(shù)據(jù)分區(qū);

(2) 本地排序階段：對上述分區(qū)進行部分排序;

(3) 混洗階段：將數(shù)據(jù)按照 key 重新分布到處理節(jié)點上;

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

(5) 寫入階段：將排序后的分區(qū)寫入 HDFS 文件。

Hadoop 發(fā)行版包含對 TeraSort 的實現(xiàn)，同樣的實現(xiàn)也可以用于 Tez，因為 Tez 可以執(zhí)行通過MapReduce API 編寫的程序。Spark 和 Flink 的 TeraSort 實現(xiàn)由 Dongwon Kim 提供.用來測量的集群由 42 臺機器組成，每臺機器 包含 12 個 CPU 內(nèi)核、24GB 內(nèi)存，以及 6 塊硬盤。

結(jié)果顯示，F(xiàn)link 的排序時間比其他所有系統(tǒng)都少。 MapReduce 用了2157 秒，Tez 用了1887 秒，Spark 用了2171 秒，F(xiàn)link 則 只用了 1480 秒。

第二個任務(wù)是一個大數(shù)據(jù)集(240GB)和一個小數(shù)據(jù)集(256MB)之間的分布式散列連接。結(jié)果顯示，F(xiàn)link 仍然是速度最快的系統(tǒng)，它所用的時間分別是 Tez 和 Spark 的 1/2 和 1/4.

產(chǎn)生以上結(jié)果的總體原因是，F(xiàn)link 的執(zhí)行過程是基于流的，這意味著各個處理階段有更多的重疊，并且混洗操作是流水線式的，因此磁盤訪問操作更少。相反，MapReduce、Tez 和 Spark 是基于批的，這意味著數(shù)據(jù)在通過網(wǎng)絡(luò)傳輸之前必須先被寫入磁盤。該測試說明，在使用Flink 時，系統(tǒng)空閑時間和磁盤訪問操作更少。

值得一提的是，性能測試結(jié)果中的原始數(shù)值可能會因集群設(shè)置、配置和軟件版本而異。

因此，F(xiàn)link 可以用同一個數(shù)據(jù)處理框架來處理無限數(shù)據(jù)流和有限數(shù)據(jù)流，并且不會犧牲性能。