Apache Flink 简介和编程模型
????Apache Flink是一個同時支持分布式數據流處理和數據批處理的大數據處理系統。 Flink可以表達和執行許多類別的數據處理應用程序，包括實時數據分析，連續數據管道，歷史數據處理（批處理）和迭代算法（機器學習，圖表分析）以及容錯的數據流。
1. 簡介
????實時流處理系統（Stream Processing System）和歷史數據處理（BatchProcessing System）傳統上被認為是兩個非常不同的類型的應用。他們使用不同的編程模型和API進行編程，并且由不同的系統執行。例如ApacheStorm、Spark Streaming等專有的實時流處理系統與Hadoop、Spark等專用的批處理系統。
????然而目前越來越明顯的是，當今大量的大規模數據處理應用的數據實際上是隨著時間的推移而不斷產生的，也就是實時流處理。這些連續的數據流來自如Web日志、應用程序日志，傳感器或則一些事務日志記錄。然而目前的解決方案不是把實時流數據當作流來處理，而是忽略了數據產生的連續性和及時性。數據記錄（通常是人為的）分批次存入靜態數據集中（例如，小時，每日或每月），然后以時間不可知的方式進行處理。像“lambda架構”這樣的架構模式結合了批處理和流處理系統，以實現多種計算方式：為及時的近似結果提供了一個快速的流處理方法，并為后期的精確結果提供了一個批量離線處理方法。所有這些方法都有很高的延遲(批處理)、高復雜性(連接和協調多個系統，并實現業務邏輯兩次)，以及任意時間的不準確，因為時間維度沒有被應用程序代碼顯式地處理。
????Apache Flink 很好的解決了這個問題。Apache Flink遵循一種模式，將數據流處理作為統一模型，同時支持數據實時流處理和數據批處理。并且它與持久消息隊列相結合，允許數據流的準任意重放(如ApacheKafka或Amazon Kinesis)。Apache Flink 將批處理程序看作是有邊界的數據集，實時流數據是無邊界的數據集。使用者只需要使用一個系統，它既能處理實時流數據，也能處理靜態歷史數據集。ApacheFlink為了使它的批處理更加具有競爭性和性能優越，Apache Flink使用一個專用的API用于處理靜態數據集，并且使用專門的數據結構和算法優化批處理操作，如join或group。
2. 系統架構
圖5.26 Flink的生態圈[19]
?
????圖5.26顯示了Apache Flink的軟件生態圈。Flink中有兩個核心API:用于處理有限數據集DataSet API(通常稱為批處理)和用于處? ?理無界數據流的DataStream API(通常稱為實時流處理)。Flink的核心是DistributedStreaming Dataflow引擎，它用來執行dataflow程序。Flink的核心運行引擎可以看作是Streaming Dataflow引擎，DataSetAPI和DataStreamAPI都可以通過該引擎創建運行時程序。在核心API的基礎上，Flink還綁定了用于特定于領域的庫和API，目前是用于機器學習的FlinkML, 用于圖處理的Gelly和用于sql的操作的Table API。從部署模式上講，Flink支持local模式、集群模式（standalone集群或者Yarn集群）、Cloud端部署。
????圖5.27顯示Apache Flink的分布式運行環境架構。Flink分布式程序包含2個主要的進程：JobManager和TaskManager。當程序運行時，不同的進程就會參與其中，包括JobManager、TaskManager和JobClient。
JobManager：也叫做Master進程，負責Job的管理和資源的協調。包括任務調度，監控任務的執行狀態，協調任務的執行，檢查點管理，失敗恢復等。
TaskManager：Task Manager是具體執行tasks的worker節點，執行任務運行在一個JVM中的一個或多個線程中。TaskManager就是運行在不同節點上的JVM進程，這個進程會擁有一定的量的資源。比如內存，CPU，網絡，磁盤等等。可以將進程的內存劃分到多個slot上去。每個TaskSlot包括JVM進程中的一部分內存。
TaskSlots：TaskSlot是分布式程序真正執行Task的地方。通過調整TaskSlot的數量，用戶可以定義子任務是如何相互隔離的。假如每個TaskManager有一個TaskSlot，意味著每個task在單獨的JVM中運行(例如，可以在單獨的容器中啟動)。TaskManager擁有多個TaskSlot意味著更多的子任務共享相同的JVM內存。相同JVM中的任務共享TCP連接(通過多路復用)和心跳消息。它們還可以共享數據集和數據結構，從而減少了每個任務的開銷。
JobClinet：JobClient是程序執行的入口。Job Client負責接收用戶提交的程序，并將用戶提交的程序通過優化器和GraphBuilder轉換成Dataflow graph。然后將生成的Data flow提交給Job Manager。一旦執行完成，Job Client將返回給用戶執行結果。
圖5.27 Apache Flink分布式運行環境[20]
?
????Flink具體執行流程如下。首先，Flink程序提交給JobClient，JobClient再提交到JobManager，JobManager負責資源的協調和Job的執行。一旦資源分配完成，task就會分配到不同的TaskManager，TaskManager會初始化線程去執行task，并根據程序的執行狀態向JobManager反饋，執行的狀態包括starting、in progress、finished以及canceled和failing等。當Job執行完成，結果會返回給客戶端。
3. 一致性語義和容錯
????Apache Flink提供容錯機制來持續恢復數據流應用程序的狀態。該機制確保即使在出現故障時，程序的狀態最終也會恢復到之前正常運行的狀態。程序中的數據流也只會被處理一次，也就是我們常常說的ExactlyOnce。ApacheFlink容錯機制不斷的生成分布式數據流快照。對于程序產生狀態較小的流式傳輸應用程序，這些快照是非常輕量級的，可以頻繁的異步生成，并且對整個系統性能不會產生太大的影響。流應用程序的狀態存儲一般都放置在分布式文件系統中（比如HDFS）。如果程序失敗（由于機器，網絡或軟件故障），Flink會停止發送分布式流式數據流。然后系統重新啟動所有的Operator并將其重置為最新的checkpoints。輸入流被重置為state snapshot中記錄的 offset。由于實時流處理系統的輸入流都是可以回播的輸入源，比如Kafka這樣的分布式消息發布訂閱系統，系統恢復時，可以從最新的checkpoint所在的輸入數據偏移量開始發送數據。
1）Barriers
?????? Flink的分布式快照的核心元素是streambarriers。這些barriers被注入到數據流中，作為數據流的一部分和其他數據一同流動（正如InfoSphere的punctuation），barriers不會超過其他數據到達（亂序到達）。一個Barrier將數據流中的數據分割成兩個數據集，即進入當前快照的數據和進入下一次快照的數據。每個Barrier帶有一個ID，該ID為將處于該Barrier之前的數據歸入快照的檢查點的ID。Barrier不會打斷數據流的流動，所以它是十分輕量級的。來自不同的快照的多個Barrier可以同一時間存在于同一個流中，也就是說，不同的快照可以并行同時發生。
圖5.28 Apache Flink Barrier 容錯機制[24]
?????? 數據流中的Barrier是在數據流的source處被插入到并行數據流圖的。快照n的barrier被插入的點（成為Sn），就是在源數據流中快照n能覆蓋到的數據的最近位置，如在Apache Kafaka中，這個位置就是上一個數據（record）在分區（partition）中的偏移量（offset）。這個位置Sn將會交給checkpointcoordinator（Flink的JobManager中）。
?????? 這些Barrier隨數據流流動向下游，當一個中間Operator在其輸入流接收到快照n的barrier時，它在其所有的輸出流中都發送一個快照n的Barrier。當一個sinkoperator（流DAG的終點）從其輸入流接收到n的Barrier，它將快照n通知給checkpointcoordinator。在所有Sink都通知了一個快照后，這個快照就完成了。當快照n完成后，由于數據源中先于Sn的的數據已經通過了整個data flowtopology，我們就可以確定不再需要這些數據了。
2）Apache Flink 語義
Apache Flink基于State的異步快照機制，可以做到Exactly Once的語義。也就是說Apache Flink在失敗恢復的時候，也能保證輸入流的數據，當且僅當被處理一次。
?????? 對齊操作可能會給流應用增加延遲（latency），通常這些額外時延都僅是毫秒級的，但也有在一些異常情況下延遲明顯增長的情況。一些應用對所有數據都嚴格要求極低延遲（幾毫秒），在這些應用中，Flink提供一個可以跳過檢查點中對齊操作的開關接口。檢查點快照依然將在Operator在所有輸入流接收到檢查點Barrier時生成。當選擇跳過對齊操作時，即使Operator在一些輸入流中接收到檢查點n的Barrier，它仍將繼續處理所有輸入數據。在這種情況下，Operator在檢查點n快照生成之前，也會處理屬于快照n+1的數據。在恢復時，這些數據將會重復出現，因為它們既屬于檢查點n的狀態快照，也會在檢查點n之后的數據重放（replay）中出現。
?????? 由于一些特殊的應用需要極低的延遲(幾毫秒)，這個時候Apache Flink可以通過禁用對齊操作以提高性能，這種情況下，Apache Flink的一致性語義就是At Least Once。
4. ApacheFlink編程模型
4.1數據模型
???????? Flink 數據以record是作為它的處理單元。每個record又是由一個Event產生，每個record再實時流處理系統一般都是和時間相綁定。Record一般會有以下三種時間：
1. 事件時間（EventTime），是指事件創建時的時間。這種類型時間一般會表示為事件的時間戳，再通過事件生成傳感器或者事件生成服務等附到事件中。Flink 通過時間戳指定器獲取事件的時間戳。
2. 攝入時間（IngestionTime），是指事件在源運算符中進入Flink的數據流的時間。
3. 處理時間（ProcessingTime），是指運算符在執行時間類操作時的本地時間。
4.2計算模型
1）Flink DataStream
????Flink程序的基礎構建模塊是流與轉換。其中Stream是一個中間結果數據，而Transformation是一個操作，它對一個或多個輸入Stream進行計算處理，輸出一個或多個結果Stream。執行時，Flink程序映射到數據流，由流以及相應的Transformation構成。每一個數據流起始于一個或多個 source，并終止于一個或多個 sink。數據流類似于任意的有向無環圖（DAG）。如下圖所示：
圖5.29 Apache Flink Streaming Dataflow[21]
????Flink程序本質上是并行分布的。在執行過程中，一個流包含一個或多個流分區 ，而每一個Transformations操作包含一個或多個Transformations子任務 。操作子任務間彼此獨立，以不同的線程執行，甚至有可能運行在不同的機器或容器上。
????Flink包括許多Transformations操作。比如Map、FlatMap、Reduce、Window等等。Flink也由許多Sink操作，比如writeAsText、writeAsCsv、print等Sink操作。需要了解的讀者可以通過閱讀ApacheFlink的官方API。
2）并行數據流
????本質上說，Flink 程序是分布式、并發執行的。在程序運行過程中，一個數據流可能會有一個或多個流分區，而一個運算符也可能會有一個或多個運算子任務。每個運算子任務與另外一個運算子任務之間都是相互獨立的，他們是在不同的線程中運行的，甚至有可能所運行的機器或者容器都完全不同。運算子任務的數量由運算符的并發數確定。數據流的并發數就是它所生成的運算符的個數。程序中不同的運算符可以有不同等級的并發量。
圖5.30 Apache Flink Programing Model[21]
????streams可以在兩個算子之間以one-to-one模式或者redistributing模式傳遞數據。
1.???????one-to-one模式的數據流（例如上圖中Source和map()運算符之間的數據流）中元素的分組和順序會保持不變，也就是說，map()運算符的子任務所看見的元素與 Source 運算符的子任務所生成的元素的順序完全一致。
2.???????redistributing模式的數據流（例如上圖中map()和keyBy/window運算符之間的數據流，以及keyby/window和Sink運算符之間的數據流）會改變數據流所在的分區。根據所選的變換的不同，每個運算子任務會將數據發送到不同的目標子任務中去。keyBy()（通過對key進行哈希計算來重分區）、boradcast()和rebalance()（隨機重分區）就是重分發模式的幾個例子。在重分發模式下，元素之間的先后次序在每對發送——接收子任務（例如map()的子任務和keyBy/window的子任務）中是保持不變的。因此，在上圖的例子中，盡管在子任務之間每個key的順序都是確定的，但是由于程序的并發過程引入了不確定性，最終到達 Sink 的元素順序就不能保證與一開始的元素順序完全一致。
4.3 基本操作
????????
Transformation | Transform Operator描敘 |
Map | 取一個元素并生成一個元素 |
FlatMap | 獲取一個元素并生成一個或多個元素。 |
Filter | 為每一個元素進行過濾操作 |
KeyBy | 邏輯上將一條流分解成不同的分區，所有具有相同key的記錄都被分配到相同的分區。 |
Reduce | 將相同分區的元素進行聚合操作 |
Window | Windows根據某些特性（例如，在最近5秒內到達的數據）對每個key中的數據進行分組， |
Split | 根據一些標準將流分成兩個或更多流。 |
Select | 從一個split流中選擇一個分流 |
Fold | 在具有初始值的key數據流上“滾動”折疊。 |
Aggregation | 在key數據流上滾動聚合。 |
WindowAll | Windows根據某些特征（例如，在最近5秒內到達的數據）對所有流事件進行分組。 |
Union | 兩個或更多數據流的聯合創建一個包含所有流中所有元素的新流。 |
Window Apply | 將整體函數應用于窗口。 |
Window Reduce | 在窗口中應用函數進行reduce并返回reduce的值。 |
Connect | “連接”保留其類型的兩個數據流。 |
?
Sink | Sink Operator 描敘 |
WriteAsText | 將數據輸出為文本文件 |
WriteAsCsv | 將數據輸出為Csv文件 |
打印標準輸出流上每個元素的toString（）值。 | |
WriteToSocket | 根據SerializationSchema將元素寫入Socket |
WriteUsingOutputFormat | 自定義文件輸出的方法和基類。 支持自定義對象到字節的轉換。 |
4.4 Window機制
?????? 計數（counts）、求和（sums）等聚合事件和批處理過程的工作模式完全不同。舉個例子，由于數據流在理論上是無限的，因此直接計算數據流中的所有元素的個數基本上是無法實現的。因此，數據流的聚合操作（計數、求和等）都是由窗口（window）限定了范圍的，例如“計算前五分鐘的元素個數”，“對前100個元素求和”等。
?????? 窗口可以通過時間（例如以30秒為單位）或者數據（例如以100個元素為單位）來定義。有多種不同類型的窗口，例如數據不重疊的滾動窗口（tumbling window）、數據重疊的滑動窗口（slidingwindow），以及以非活動狀態為間隔的會話窗口（session window）。
圖5.31 Apache Flink window mechanism [21]
4.5 Flink DataSet
????Flink將批處理程序作為流處理程序的特殊情況來執行，只是這種特殊的“流”是有界的，然而批處理程序這種特殊的“流”是無邊界的。?DataSet在Flink系統內部被執行時視為DataFlow。上述適用于流處理程序的概念同樣適用于批處理程序。Flink對于處理批處理程序，它也同樣通過構建上述介紹的Dataflow數據模型。只不過Flink將批處理程序中的靜態數據集視為有邊界的數據集，對于實時流處理程序，Flink將實時流數據視為無邊界的數據集。對于批處理程序，ApacheFlink使用一個專用的API，對于處理靜態數據集，使用專門的數據結構和算法來進行批處理操作，如join或group，并且使用專用的調度策略進行優化。
?
總結
以上是生活随笔為你收集整理的Apache Flink 简介和编程模型的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 内存映射IO (MMIO) 简介
- 下一篇： Apache SparkStreamin