Flink中怎么實現批流一體

這篇文章將為大家詳細講解有關Flink中怎么實現批流一體，文章內容質量較高，因此小編分享給大家做個參考，希望大家閱讀完這篇文章后對相關知識有一定的了解。

實現批處理的技術許許多多，從各種關系型數據庫的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中怎么實現批流一體就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。