[[312719]]

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

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

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

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

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

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

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

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

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

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

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

Fink批處理模型

Flink 通過一個(gè)底層引擎同時(shí)支持流處理和批處理

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

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

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

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

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

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

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

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

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

Flink批處理性能

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

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

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

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

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

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

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

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

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

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

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

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

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

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