2.3 編程模型

2.3.1 分層組件棧

Flink的組件分為4層，各個模塊之間的層次關(guān)系如圖2-5所示。

（1）Deploy層：Flink支持多種部署模式，如本地（Local）單機版、Standalone集群、YARN集群及云（Cloud）部署模式。

（2）Core 層：本層是 Flink 分布式數(shù)據(jù)處理引擎的核心實現(xiàn)層，包括計算圖的所有底層實現(xiàn)，例如時間與窗口機制、一致性語義、任務(wù)管理與調(diào)度、物理執(zhí)行計劃。應(yīng)用程序通常不需要調(diào)用本層 API，而是調(diào)用流處理 API、批處理API或構(gòu)建在這兩層API基礎(chǔ)之上的Library API。

（3）API層：該層包括流處理API和批處理API，F(xiàn)link的批處理是建立在流式架構(gòu)上的，而不是用批處理模擬流處理，這種技術(shù)基因決定了 Flink 更適用于流處理的場合。

（4）Library層：該層是Flink的應(yīng)用框架層，構(gòu)建在流處理API和批處理API之上，因此同一應(yīng)用框架庫有兩種版本選擇，如流式關(guān)系型 API（Table/SQL）。此外，本層還包括CEP、FlinkML和Gelly。

2.3.2 流式計算模型

一個典型的流處理應(yīng)用程序（命名為Programm 2.1）如下：

這段程序的邏輯計算圖形式如圖2-6所示。

圖 2-6 中 Stream 為傳輸通道中的數(shù)據(jù)，Operator 為計算圖的節(jié)點，Streaming Dataflow為計算圖。

計算圖的物理形式由計算節(jié)點的多個并行實例組成，其中并行實例的含義是：在分布式環(huán)境中，同一計算節(jié)點有多個功能相同的物理部署實例，如圖2-7中邏輯形式中的map（）節(jié)點會有兩個部署實例map（）[1]和map（）[2]。

在并行模式時：

（1）每個Operator的實例數(shù)為并行度，任意兩個Operator的并行度之間是獨立的。例如，圖2-7中Source Operator的并行度為2，而Sink Operator的并行度為1；每個Operator稱為一個任務(wù)，Operator的每個實例稱為子任務(wù)（subtask），子任務(wù)這個概念來自其和JVM線程之間的關(guān)系。

（2）Stream有一個或多個分區(qū)（partition）。Stream有兩種模式：

● 直連（One-to-One）模式，即一個實例的輸出是另一個實例的輸入。在Programm 2.1 的物理計算圖形式中，Source 的 subtask[1]（即 Source[1]）和map的subtask[1]（即map [1]）直接相連，Source[1]的輸出全部傳輸給map [1]，沒有被拆分成多個分區(qū)。

● 分區(qū)（Redistribution）模式，即一個實例的輸出被拆分成多個部分傳輸給多個下級實例。在Programm 2.1的物理計算圖形式中，map [1]被拆分成兩部分，分別輸入給不同的下級實例。

2.3.3 流處理編程

1.DataStream與DataSet

Flink用DataStream表示無界數(shù)據(jù)集，用DataSet表示有界數(shù)據(jù)集，前者用于流處理應(yīng)用程序，后者用于批處理程序。根據(jù)所處理事件數(shù)據(jù)結(jié)構(gòu)類型的不同，應(yīng)用程序可以定義不同類型的 DataStream對象和 DataSet對象。以下程序定義事件類型為String的DataStream對象和事件類型為LabeledVector（帶標(biāo)簽的訓(xùn)練樣本，每個樣本用向量表示）的DataSet對象：

從操作形式上看，DataStream 和 DataSet 與集合（Collection）有些相似，但是兩者有著本質(zhì)不同：

（1）DataStream 和 DataSet 是不可變的數(shù)據(jù)集合，因此不可以像操縱集合那樣增加或刪除 DataStream和 DataSet中的元素，也不可以通過諸如下標(biāo)等方式訪問某個元素。這里重申之前定義的概念，事件、元素、數(shù)據(jù)等都是用于指代流處理或批處理所處理的數(shù)據(jù)對象的，具體使用哪個稱呼依賴語境。

（2）Flink應(yīng)用程序通過Source創(chuàng)建DataStream對象和DataSet對象，通過轉(zhuǎn)換操作產(chǎn)生新的DataStream對象和DataSet對象。

2.程序結(jié)構(gòu)

Flink按照數(shù)據(jù)處理流程編寫應(yīng)用程序，共分為5個步驟。

1）獲取運行時

運行時分為兩類：StreamingExecutionEnvironment和ExecutionEnvironment，分別對應(yīng)流處理和批處理程序：

運行時是應(yīng)用程序被調(diào)度執(zhí)行時的上下文環(huán)境，上述方法根據(jù)當(dāng)前環(huán)境自動選擇本地或集群運行時環(huán)境。以流處理為例，創(chuàng)建方法如下：

（1）通過createLocalEnvironment方法創(chuàng)建運行時，基于這種運行時的應(yīng)用程序會運行在同一個 JVM 進程中，本地調(diào)試時通常采用這種運行時。createLocalEnvironment有三種接口形式：

從上面的接口可以看出，通過 createLocalEnvironment 方法創(chuàng)建的運行仍是StreamingExecutionEnvironment。

（2）通過 createRemoteEnvironment 創(chuàng)建運行時，基于這種運行時的應(yīng)用程序會被提交到集群中運行，連接集群調(diào)試通常用這種運行時。createRemoteEnvironment有兩種接口形式：

2）添加外部數(shù)據(jù)源

可以添加外部數(shù)據(jù)源，如 Kafka和文件，也可以由應(yīng)用創(chuàng)建 DataStream或DataSet，后一種方法常用于測試環(huán)境。

3）定義算子轉(zhuǎn)換函數(shù)

下面的代碼將input元素值轉(zhuǎn)換成整型，轉(zhuǎn)換后得到DataStream[Int]：

4）定義Sink

Sink的功能是將數(shù)據(jù)處理結(jié)果寫入外部系統(tǒng)：

除了上述兩種常用的Sink，應(yīng)用程序還可以將處理結(jié)果寫入Kafka：

5）啟動程序

調(diào)用運行時的execute（）方法：

3.指定鍵（key）

可以通過Scala Case類（或Java元組）的位置索引、對象屬性名稱、key的選擇器（selector）三種方式指定key，定義如下：

4.并行度設(shè)置

有4種設(shè)置Flink并行度的方式。

（1）通過緊跟在Operator之后的setParallelism方法設(shè)置并行度，這種并行度只影響對應(yīng)的Operator：

（2）通過運行時設(shè)置作業(yè)級并行度：

（3）通過客戶端設(shè)置并行度，這種并行度也是作業(yè)級的：

（4）通過 Flink 的配置文件設(shè)置系統(tǒng)級并行度，這種并行度對集群上的所有作業(yè)都起作用：