一文秒懂大資料流實時處理框架Flink與Spark的異同點

閱讀本文約需要10分鐘，您可以先關注我們，避免下次無法找到。

01 簡介

Spark Streaming是目前相對流行的實時流處理框架，但準確來說spark底層是透過一個微批處理來模擬實時處理，相對來說，某些場景下實時性欠缺，無法對應一些實時性要求很高的流處理場景，譬如雙十一場景下的銷售額統計等。

這是因為 Spark的流處理是基於所謂微批處理（ Micro- batch processing）的思想，即它把流處理看作是批處理的一種特殊形式，每次接收到一個時間間隔的資料才會去處理，底層微批設計導致很難在表層實時資料處理上有質的提升。

雖然Spark2。3中提出了連續處理模型（ Continuous Processing Model），但目前只支援很有限的功能，並不能在大的專案中使用，業界也無成熟的應用案例。Spark還需要做出很大的努力才能改進現有的準實時流處理模型。想要在流處理的實時性上提升，就不能繼續用微批處理的模式，而要想辦法實現真正的流處理即每當有一條資料輸入就立刻處理，不做等待。

Flink最核心的資料結構是Stream，它代表一個執行在多分割槽上的並行流。在 Stream 上同樣可以進行各種轉換操作（Transformation）。與 Spark 的 RDD 不同的是，Stream 代表一個數據流而不是靜態資料的集合。

所以，它包含的資料是隨著時間增長而變化的。而且 Stream 上的轉換操作都是逐條進行的，即每當有新的資料進來，整個流程都會被執行並更新結果。這樣的基本處理模式決定了 Flink 會比 Spark Streaming 有更低的流處理延遲性。

02 Spark與Flink的異同點

（1）相同點

Spark和Flink都支援批處理和流處理，兩者在資料處理框架上具有一定的一致性。

1）都基於記憶體計算；

2）都有統一的批處理和流處理APl，都支援類似SQL的程式設計介面；

3）都支援很多相同的轉換操作，程式設計都是用類似於Scala Collection APl的函數語言程式設計模式；

4）都有完善的錯誤恢復機制；

5）都支援Exactly once的語義一致性。

（2）不同點

在具體實時處理細節上，兩者仍存在一定的差異。從流處理的角度來講，Spark基於微批次處理，把流資料看成是一個個小的批處理資料塊分別處理，所以延遲性只能做到秒級。而Flink基於每個事件處理，每當有新的資料輸入都會立刻處理，是真正的流式計算，支援毫秒級計算。由於相同的原因，Spark只支援基於時間的視窗操作（處理時間或者事件時間），而Flink支援的視窗操作則非常靈活，不僅支援時間視窗，還支援基於資料本身的視窗（另外還支援基於time、count、session，以及data-driven的視窗操作），更是支援基於資料時間及水位線的視窗設定，自由度更大，時間視窗變數考量更充分，開發者可以自由定義想要的視窗操作。

03 Flink功能特徵

（1）處理特徵

Flink具有如下的處理特徵：

1）支援高吞吐、低延遲、高效能的流處理

2）支援帶有事件時間的視窗（Window）操作

3）支援有狀態計算的Exactly-once語義

4）支援高度靈活的視窗（Window）操作，支援基於time、count、session，以及data-driven的視窗操作

5）支援具有Backpressure功能的持續流模型

6）支援基於輕量級分散式快照（Snapshot）實現的容錯

7）一個執行時同時支援Batch on Streaming處理和Streaming處理

8）Flink在JVM內部實現了自己的記憶體管理

9）支援迭代計算

10）支援程式自動最佳化：避免特定情況下Shuffle、排序等昂貴操作，中間結果有必要進行快取

（2）資料處理API

Flink主要支援如下的兩個資料處理API：

1）DataStream API （流處理）

2）DataSet API（批處理）

（3）場景支援

Flink的主要使用場景如下：

1）支援機器學習（FlinkML）

2）支援圖分析（Gelly）

3）支援關係資料處理（Table）

4）支援複雜事件處理（CEP）

04 Flink簡要安裝使用示例

1）要執行Flink，要求必須安裝好Java 8。x。使用如下命令檢查Java是否已經正確安裝：

2）下載Flink安裝包，可以選擇任何喜歡的Scala組合，如下所示

3）將安裝包下載到指定目錄，解壓縮後進入Flink應用目錄

1。 $ tar -xzf flink-1。12。0-bin-scala_2。12。tgz 2。 $ cd flink-1。12。0-bin-scala_2。12

4）以本地模式啟動Flink

1。 $ 。/bin/start-cluster。sh 2。 Starting cluster。 3。 Starting standalonesession daemon on host。 4。 Starting taskexecutor daemon on host。

5）瀏覽器中訪問8081埠，檢視Flink WebUI，如下圖所示

6）透過Flink實現單詞數的統計（WordCount）的Pom。xml檔案配置如下

1。 <？xml version=“1。0” encoding=“UTF-8”？> 2。 5。 4。0。0 6。 helloword 7。 helloword 8。 1。0-SNAPSHOT 9。 10。 11。 org。apache。flink 12。 flink-clients_2。12 13。 1。11。2 14。 15。 16。 org。apache。flink 17。 flink-connector-kafka-0。11_2。12 18。 1。11。2 19。 20。 21。 org。apache。flink 22。 flink-scala_2。12 23。 1。11。2 24。 25。 26。 org。apache。flink 27。 flink-streaming-scala_2。12 28。 1。11。2 29。 30。 31。 org。slf4j 32。 slf4j-log4j12 33。 1。7。21 34。 test 35。 36。 37。 log4j 38。 log4j 39。 1。2。17 40。 41。 42。 43。 44。 45。 net。alchim31。maven 46。 scala-maven-plugin 47。 4。4。0 48。 49。 50。 51。 compile 52。 53。 54。 55。 56。 57。 org。apache。maven。plugins 58。 maven-assembly-plugin 59。 3。0。0 60。 61。 62。 jar-with-dependencies 63。 64。 65。 66。 67。 make-assembly 68。 package 69。 70。 single 71。 72。 73。 74。 75。 76。 77。

7）實現WordCount的Scala檔案的配置如下

1。 import org。apache。flink。api。scala。_ 2。 import org。apache。flink。api。scala。ExecutionEnvironment 3。 object WordCount { 4。 def main（args： Array［String］）： Unit = { 5。 val env： ExecutionEnvironmentExecutionEnvironment = ExecutionEnvironment。getExecutionEnvironment 6。 val inputPath： String = “E：\\IdeaProjects\\helloword\\src\\main\\resources\\hello。txt” 7。 val inputDataSet： DataSet［String］ = env。readTextFile（inputPath） 8。 val resultDataSet： DataSet［（String，Int）］ = inputDataSet。flatMap（_。split（“ ”）） 9。。map（（_， 1）） 10。。groupBy（0） 11。。sum（1） 12。 resultDataSet。print（） 13。 } 14。 }

8）將測試資料寫入hello。txt 檔案中，執行WordCount類，列印輸出計算結果，如圖所示：