Apache Flink是一個面向分散式資料流處理和批量資料處理的開源計算平臺，它能夠基於同一個Flink執行時（Flink Runtime），提供支援流處理和批處理兩種型別應用的功能。現有的開源計算方案，會把流處理和批處理作為兩種不同的應用型別，因為他們它們所提供的SLA是完全不相同的：流處理一般需要支援低延遲、Exactly-once保證，而批處理需要支援高吞吐、高效處理，所以在實現的時候通常是分別給出兩套實現方法，或者通過一個獨立的開源框架來實現其中每一種處理方案。例如，實現批處理的開源方案有MapReduce、Tez、Crunch、Spark，實現流處理的開源方案有Samza、Storm。
Flink在實現流處理和批處理時，與傳統的一些方案完全不同，它從另一個視角看待流處理和批處理，將二者統一起來：Flink是完全支援流處理，也就是說作為流處理看待時輸入資料流是無界的；批處理被作為一種特殊的流處理，只是它的輸入資料流被定義為有界的。基於同一個Flink執行時（Flink Runtime），分別提供了流處理和批處理API，而這兩種API也是實現上層面向流處理、批處理型別應用框架的基礎。

基本特性 關於Flink所支援的特性，我這裡只是通過分類的方式簡單做一下梳理，涉及到具體的一些概念及其原理會在後面的部分做詳細說明。
流處理特性

• 支援高吞吐、低延遲、高效能的流處理
• 支援帶有事件時間的視窗（Window）操作
• 支援有狀態計算的Exactly-once語義
• 支援高度靈活的視窗（Window）操作，支援基於time、count、session，以及data-driven的視窗操作
• 支援具有Backpressure功能的持續流模型
• 支援基於輕量級分散式快照（Snapshot）實現的容錯
• 一個執行時同時支援Batch on Streaming處理和Streaming處理
• Flink在JVM內部實現了自己的記憶體管理
• 支援迭代計算
• 支援程式自動優化：避免特定情況下Shuffle、排序等昂貴操作，中間結果有必要進行快取
  

API支援

• 對Streaming資料類應用，提供DataStream API
• 對批處理類應用，提供DataSet API（支援Java/Scala）
  

Libraries支援

• 支援機器學習（FlinkML）
• 支援圖分析（Gelly）
• 支援關係資料處理（Table）
• 支援複雜事件處理（CEP）
  

整合支援

• 支援Flink on YARN
• 支援HDFS
• 支援來自Kafka的輸入資料
• 支援Apache HBase
• 支援Hadoop程式
• 支援Tachyon
• 支援ElasticSearch
• 支援RabbitMQ
• 支援Apache Storm
• 支援S3
• 支援XtreemFS
  

基本概念 Stream & Transformation & Operator 使用者實現的Flink程式是由Stream和Transformation這兩個基本構建塊組成，其中Stream是一箇中間結果資料，而Transformation是一個操作，它對一個或多個輸入Stream進行計算處理，輸出一個或多個結果Stream。當一個Flink程式被執行的時候，它會被對映為Streaming Dataflow。一個Streaming Dataflow是由一組Stream和Transformation Operator組成，它類似於一個DAG圖，在啟動的時候從一個或多個Source Operator開始，結束於一個或多個Sink Operator。
下面是一個由Flink程式對映為Streaming Dataflow的示意圖，如下所示：


上圖中，FlinkKafkaConsumer是一個Source Operator，map、keyBy、timeWindow、apply是Transformation Operator，RollingSink是一個Sink Operator。
Parallel Dataflow 在Flink中，程式天生是並行和分散式的：一個Stream可以被分成多個Stream分割槽（Stream Partitions），一個Operator可以被分成多個Operator Subtask，每一個Operator Subtask是在不同的執行緒中獨立執行的。一個Operator的並行度，等於Operator Subtask的個數，一個Stream的並行度總是等於生成它的Operator的並行度。
有關Parallel Dataflow的例項，如下圖所示：


上圖Streaming Dataflow的並行檢視中，展現了在兩個Operator之間的Stream的兩種模式：

• One-to-one模式
  

比如從Source[1]到map()[1]，它保持了Source的分割槽特性（Partitioning）和分割槽內元素處理的有序性，也就是說map()[1]的Subtask看到資料流中記錄的順序，與Source[1]中看到的記錄順序是一致的。

• Redistribution模式
  

這種模式改變了輸入資料流的分割槽，比如從map()[1]、map()[2]到keyBy()/window()/apply()[1]、keyBy()/window()/apply()[2]，上游的Subtask向下遊的多個不同的Subtask傳送資料，改變了資料流的分割槽，這與實際應用所選擇的Operator有關係。
另外，Source Operator對應2個Subtask，所以並行度為2，而Sink Operator的Subtask只有1個，故而並行度為1。

Task & Operator Chain 在Flink分散式執行環境中，會將多個Operator Subtask串起來組成一個Operator Chain，實際上就是一個執行鏈，每個執行鏈會在TaskManager上一個獨立的執行緒中執行，如下圖所示：

上圖中上半部分表示的是一個Operator Chain，多個Operator通過Stream連線，而每個Operator在執行時對應一個Task；圖中下半部分是上半部分的一個並行版本，也就是對每一個Task都並行化為多個Subtask。


Time & Window Flink支援基於時間視窗操作，也支援基於資料的視窗操作，如下圖所示：

上圖中，基於時間的視窗操作，在每個相同的時間間隔對Stream中的記錄進行處理，通常各個時間間隔內的視窗操作處理的記錄數不固定；而基於資料驅動的視窗操作，可以在Stream中選擇固定數量的記錄作為一個視窗，對該視窗中的記錄進行處理。
有關視窗操作的不同型別，可以分為如下幾種：傾斜視窗（Tumbling Windows，記錄沒有重疊）、滑動視窗（Slide Windows，記錄有重疊）、會話視窗（Session Windows），具體可以查閱相關資料。
在處理Stream中的記錄時，記錄中通常會包含各種典型的時間欄位，Flink支援多種時間的處理，如下圖所示：

上圖描述了在基於Flink的流處理系統中，各種不同的時間所處的位置和含義，其中，Event Time表示事件建立時間，Ingestion Time表示事件進入到Flink Dataflow的時間 ，Processing Time表示某個Operator對事件進行處理事的本地系統時間（是在TaskManager節點上）。這裡，談一下基於Event Time進行處理的問題，通常根據Event Time會給整個Streaming應用帶來一定的延遲性，因為在一個基於事件的處理系統中，進入系統的事件可能會基於Event Time而發生亂序現象，比如事件來源於外部的多個系統，為了增強事件處理吞吐量會將輸入的多個Stream進行自然分割槽，每個Stream分割槽內部有序，但是要保證全域性有序必須同時兼顧多個Stream分割槽的處理，設定一定的時間視窗進行暫存資料，當多個Stream分割槽基於Event Time排列對齊後才能進行延遲處理。所以，設定的暫存資料記錄的時間視窗越長，處理效能越差，甚至嚴重影響Stream處理的實時性。
有關基於時間的Streaming處理，可以參考官方文件，在Flink中借鑑了Google使用的WaterMark實現方式，可以查閱相關資料。
基本架構 Flink系統的架構與Spark類似，是一個基於Master-Slave風格的架構，如下圖所示：

Flink叢集啟動時，會啟動一個JobManager程序、至少一個TaskManager程序。在Local模式下，會在同一個JVM內部啟動一個JobManager程序和TaskManager程序。當Flink程式提交後，會建立一個Client來進行預處理，並轉換為一個並行資料流，這是對應著一個Flink Job，從而可以被JobManager和TaskManager執行。在實現上，Flink基於Actor實現了JobManager和TaskManager，所以JobManager與TaskManager之間的資訊交換，都是通過事件的方式來進行處理。
如上圖所示，Flink系統主要包含如下3個主要的程序：
JobManager JobManager是Flink系統的協調者，它負責接收Flink Job，排程組成Job的多個Task的執行。同時，JobManager還負責收集Job的狀態資訊，並管理Flink叢集中從節點TaskManager。JobManager所負責的各項管理功能，它接收到並處理的事件主要包括：

• RegisterTaskManager
  

在Flink叢集啟動的時候，TaskManager會向JobManager註冊，如果註冊成功，則JobManager會向TaskManager回覆訊息AcknowledgeRegistration。

• SubmitJob
  

Flink程式內部通過Client向JobManager提交Flink Job，其中在訊息SubmitJob中以JobGraph形式描述了Job的基本資訊。

• CancelJob
  

請求取消一個Flink Job的執行，CancelJob訊息中包含了Job的ID，如果成功則返回訊息CancellationSuccess，失敗則返回訊息CancellationFailure。

• UpdateTaskExecutionState
  

TaskManager會向JobManager請求更新ExecutionGraph中的ExecutionVertex的狀態資訊，更新成功則返回true。

• RequestNextInputSplit
  

執行在TaskManager上面的Task，請求獲取下一個要處理的輸入Split，成功則返回NextInputSplit。

• JobStatusChanged
  

ExecutionGraph向JobManager傳送該訊息，用來表示Flink Job的狀態發生的變化，例如：RUNNING、CANCELING、FINISHED等。

TaskManager TaskManager也是一個Actor，它是實際負責執行計算的Worker，在其上執行Flink Job的一組Task。每個TaskManager負責管理其所在節點上的資源資訊，如記憶體、磁碟、網路，在啟動的時候將資源的狀態向JobManager彙報。TaskManager端可以分成兩個階段：

• 註冊階段
  

TaskManager會向JobManager註冊，傳送RegisterTaskManager訊息，等待JobManager返回AcknowledgeRegistration，然後TaskManager就可以進行初始化過程。

• 可操作階段
  

該階段TaskManager可以接收並處理與Task有關的訊息，如SubmitTask、CancelTask、FailTask。如果TaskManager無法連線到JobManager，這是TaskManager就失去了與JobManager的聯絡，會自動進入“註冊階段”，只有完成註冊才能繼續處理Task相關的訊息。

Client 當用戶提交一個Flink程式時，會首先建立一個Client，該Client首先會對使用者提交的Flink程式進行預處理，並提交到Flink叢集中處理，所以Client需要從使用者提交的Flink程式配置中獲取JobManager的地址，並建立到JobManager的連線，將Flink Job提交給JobManager。Client會將使用者提交的Flink程式組裝一個JobGraph， 並且是以JobGraph的形式提交的。一個JobGraph是一個Flink Dataflow，它由多個JobVertex組成的DAG。其中，一個JobGraph包含了一個Flink程式的如下資訊：JobID、Job名稱、配置資訊、一組JobVertex等。
元件棧
Flink是一個分層架構的系統，每一層所包含的元件都提供了特定的抽象，用來服務於上層元件。Flink分層的元件棧如下圖所示：

下面，我們自下而上，分別針對每一層進行解釋說明：

• Deployment層
  

該層主要涉及了Flink的部署模式，Flink支援多種部署模式：本地、叢集（Standalone/YARN）、雲（GCE/EC2）。Standalone部署模式與Spark類似，這裡，我們看一下Flink on YARN的部署模式，如下圖所示：

瞭解YARN的話，對上圖的原理非常熟悉，實際Flink也實現了滿足在YARN叢集上執行的各個元件：Flink YARN Client負責與YARN RM通訊協商資源請求，Flink JobManager和Flink TaskManager分別申請到Container去執行各自的程序。通過上圖可以看到，YARN AM與Flink JobManager在同一個Container中，這樣AM可以知道Flink JobManager的地址，從而AM可以申請Container去啟動Flink TaskManager。待Flink成功執行在YARN叢集上，Flink YARN Client就可以提交Flink Job到Flink JobManager，並進行後續的對映、排程和計算處理。

• Runtime層
  

Runtime層提供了支援Flink計算的全部核心實現，比如：支援分散式Stream處理、JobGraph到ExecutionGraph的對映、排程等等，為上層API層提供基礎服務。

• API層
  

API層主要實現了面向無界Stream的流處理和麵向Batch的批處理API，其中面向流處理對應DataStream API，面向批處理對應DataSet API。

• Libraries層
  

該層也可以稱為Flink應用框架層，根據API層的劃分，在API層之上構建的滿足特定應用的實現計算框架，也分別對應於面向流處理和麵向批處理兩類。面向流處理支援：CEP（複雜事件處理）、基於SQL-like的操作（基於Table的關係操作）；面向批處理支援：FlinkML（機器學習庫）、Gelly（圖處理）。
內部原理 容錯機制 Flink基於Checkpoint機制實現容錯，它的原理是不斷地生成分散式Streaming資料流Snapshot。在流處理失敗時，通過這些Snapshot可以恢復資料流處理。理解Flink的容錯機制，首先需要了解一下Barrier這個概念：
Stream Barrier是Flink分散式Snapshotting中的核心元素，它會作為資料流的記錄被同等看待，被插入到資料流中，將資料流中記錄的進行分組，並沿著資料流的方向向前推進。每個Barrier會攜帶一個Snapshot ID，屬於該Snapshot的記錄會被推向該Barrier的前方。因為Barrier非常輕量，所以並不會中斷資料流。帶有Barrier的資料流，如下圖所示：
 
基於上圖，我們通過如下要點來說明：

• 出現一個Barrier，在該Barrier之前出現的記錄都屬於該Barrier對應的Snapshot，在該Barrier之後出現的記錄屬於下一個Snapshot
• 來自不同Snapshot多個Barrier可能同時出現在資料流中，也就是說同一個時刻可能併發生成多個Snapshot
• 當一箇中間（Intermediate）Operator接收到一個Barrier後，它會發送Barrier到屬於該Barrier的Snapshot的資料流中，等到Sink Operator接收到該Barrier後會向Checkpoint Coordinator確認該Snapshot，直到所有的Sink Operator都確認了該Snapshot，才被認為完成了該Snapshot
  

這裡還需要強調的是，Snapshot並不僅僅是對資料流做了一個狀態的Checkpoint，它也包含了一個Operator內部所持有的狀態，這樣才能夠在保證在流處理系統失敗時能夠正確地恢復資料流處理。也就是說，如果一個Operator包含任何形式的狀態，這種狀態必須是Snapshot的一部分。
Operator的狀態包含兩種：一種是系統狀態，一個Operator進行計算處理的時候需要對資料進行緩衝，所以資料緩衝區的狀態是與Operator相關聯的，以視窗操作的緩衝區為例，Flink系統會收集或聚合記錄資料並放到緩衝區中，直到該緩衝區中的資料被處理完成；另一種是使用者自定義狀態（狀態可以通過轉換函式進行建立和修改），它可以是函式中的Java物件這樣的簡單變數，也可以是與函式相關的Key/Value狀態。
對於具有輕微狀態的Streaming應用，會生成非常輕量的Snapshot而且非常頻繁，但並不會影響資料流處理效能。Streaming應用的狀態會被儲存到一個可配置的儲存系統中，例如HDFS。在一個Checkpoint執行過程中，儲存的狀態資訊及其互動過程，如下圖所示：
 
在Checkpoint過程中，還有一個比較重要的操作——Stream Aligning。當Operator接收到多個輸入的資料流時，需要在Snapshot Barrier中對資料流進行排列對齊，如下圖所示：
 
具體排列過程如下：

• Operator從一個incoming Stream接收到Snapshot Barrier n，然後暫停處理，直到其它的incoming Stream的Barrier n（否則屬於2個Snapshot的記錄就混在一起了）到達該Operator
• 接收到Barrier n的Stream被臨時擱置，來自這些Stream的記錄不會被處理，而是被放在一個Buffer中
• 一旦最後一個Stream接收到Barrier n，Operator會emit所有暫存在Buffer中的記錄，然後向Checkpoint Coordinator傳送Snapshot n
• 繼續處理來自多個Stream的記錄
  

基於Stream Aligning操作能夠實現Exactly Once語義，但是也會給流處理應用帶來延遲，因為為了排列對齊Barrier，會暫時快取一部分Stream的記錄到Buffer中，尤其是在資料流並行度很高的場景下可能更加明顯，通常以最遲對齊Barrier的一個Stream為處理Buffer中快取記錄的時刻點。在Flink中，提供了一個開關，選擇是否使用Stream Aligning，如果關掉則Exactly Once會變成At least once。
排程機制 在JobManager端，會接收到Client提交的JobGraph形式的Flink Job，JobManager會將一個JobGraph轉換對映為一個ExecutionGraph，如下圖所示：


通過上圖可以看出：
JobGraph是一個Job的使用者邏輯視圖表示，將一個使用者要對資料流進行的處理表示為單個DAG圖（對應於JobGraph），DAG圖由頂點（JobVertex）和中間結果集（IntermediateDataSet）組成，其中JobVertex表示了對資料流進行的轉換操作，比如map、flatMap、filter、keyBy等操作，而IntermediateDataSet是由上游的JobVertex所生成，同時作為下游的JobVertex的輸入。
而ExecutionGraph是JobGraph的並行表示，也就是實際JobManager排程一個Job在TaskManager上執行的邏輯檢視，它也是一個DAG圖，是由ExecutionJobVertex、IntermediateResult（或IntermediateResultPartition）組成，ExecutionJobVertex實際對應於JobGraph圖中的JobVertex，只不過在ExecutionJobVertex內部是一種並行表示，由多個並行的ExecutionVertex所組成。另外，這裡還有一個重要的概念，就是Execution，它是一個ExecutionVertex的一次執行Attempt，也就是說，一個ExecutionVertex可能對應多個執行狀態的Execution，比如，一個ExecutionVertex執行產生了一個失敗的Execution，然後還會建立一個新的Execution來執行，這時就對應這個2次執行Attempt。每個Execution通過ExecutionAttemptID來唯一標識，在TaskManager和JobManager之間進行Task狀態的交換都是通過ExecutionAttemptID來實現的。
下面看一下，在物理上進行排程，基於資源的分配與使用的一個例子，來自官網，如下圖所示：


說明如下：

• 左上子圖：有2個TaskManager，每個TaskManager有3個Task Slot
• 左下子圖：一個Flink Job，邏輯上包含了1個data source、1個MapFunction、1個ReduceFunction，對應一個JobGraph
• 左下子圖：使用者提交的Flink Job對各個Operator進行的配置——data source的並行度設定為4，MapFunction的並行度也為4，ReduceFunction的並行度為3，在JobManager端對應於ExecutionGraph
• 右上子圖：TaskManager 1上，有2個並行的ExecutionVertex組成的DAG圖，它們各佔用一個Task Slot
• 右下子圖：TaskManager 2上，也有2個並行的ExecutionVertex組成的DAG圖，它們也各佔用一個Task Slot
• 在2個TaskManager上執行的4個Execution是並行執行的
  

迭代機制 機器學習和圖計算應用，都會使用到迭代計算，Flink通過在迭代Operator中定義Step函式來實現迭代演算法，這種迭代演算法包括Iterate和Delta Iterate兩種型別，在實現上它們反覆地在當前迭代狀態上呼叫Step函式，直到滿足給定的條件才會停止迭代。下面，對Iterate和Delta Iterate兩種型別的迭代演算法原理進行說明：

• Iterate
  

Iterate Operator是一種簡單的迭代形式：每一輪迭代，Step函式的輸入或者是輸入的整個資料集，或者是上一輪迭代的結果，通過該輪迭代計算出下一輪計算所需要的輸入（也稱為Next Partial Solution），滿足迭代的終止條件後，會輸出最終迭代結果，具體執行流程如下圖所示：
 
Step函式在每一輪迭代中都會被執行，它可以是由map、reduce、join等Operator組成的資料流。下面通過官網給出的一個例子來說明Iterate Operator，非常簡單直觀，如下圖所示：
 
上面迭代過程中，輸入資料為1到5的數字，Step函式就是一個簡單的map函式，會對每個輸入的數字進行加1處理，而Next Partial Solution對應於經過map函式處理後的結果，比如第一輪迭代，對輸入的數字1加1後結果為2，對輸入的數字2加1後結果為3，直到對輸入數字5加1後結果為變為6，這些新生成結果數字2~6會作為第二輪迭代的輸入。迭代終止條件為進行10輪迭代，則最終的結果為11~15。

• Delta Iterate
  

Delta Iterate Operator實現了增量迭代，它的實現原理如下圖所示：
 
基於Delta Iterate Operator實現增量迭代，它有2個輸入，其中一個是初始Workset，表示輸入待處理的增量Stream資料，另一個是初始Solution Set，它是經過Stream方向上Operator處理過的結果。第一輪迭代會將Step函式作用在初始Workset上，得到的計算結果Workset作為下一輪迭代的輸入，同時還要增量更新初始Solution Set。如果反覆迭代知道滿足迭代終止條件，最後會根據Solution Set的結果，輸出最終迭代結果。
比如，我們現在已知一個Solution集合中儲存的是，已有的商品分類大類中購買量最多的商品，而Workset輸入的是來自線上實時交易中最新達成購買的商品的人數，經過計算會生成新的商品分類大類中商品購買量最多的結果，如果某些大類中商品購買量突然增長，它需要更新Solution Set中的結果（原來購買量最多的商品，經過增量迭代計算，可能已經不是最多），最後會輸出最終商品分類大類中購買量最多的商品結果集合。更詳細的例子，可以參考官網給出的“Propagate Minimum in Graph”，這裡不再累述。

Backpressure監控 Backpressure在流式計算系統中會比較受到關注，因為在一個Stream上進行處理的多個Operator之間，它們處理速度和方式可能非常不同，所以就存在上游Operator如果處理速度過快，下游Operator處可能機會堆積Stream記錄，嚴重會造成處理延遲或下游Operator負載過重而崩潰（有些系統可能會丟失資料）。因此，對下游Operator處理速度跟不上的情況，如果下游Operator能夠將自己處理狀態傳播給上游Operator，使得上游Operator處理速度慢下來就會緩解上述問題，比如通過告警的方式通知現有流處理系統存在的問題。
Flink Web介面上提供了對執行Job的Backpressure行為的監控，它通過使用Sampling執行緒對正在執行的Task進行堆疊跟蹤取樣來實現，具體實現方式如下圖所示：


JobManager會反覆呼叫一個Job的Task執行所線上程的Thread.getStackTrace()，預設情況下，JobManager會每間隔50ms觸發對一個Job的每個Task依次進行100次堆疊跟蹤呼叫，根據呼叫呼叫結果來確定Backpressure，Flink是通過計算得到一個比值（Radio）來確定當前執行Job的Backpressure狀態。在Web介面上可以看到這個Radio值，它表示在一個內部方法呼叫中阻塞（Stuck）的堆疊跟蹤次數，例如，radio=0.01，表示100次中僅有1次方法呼叫阻塞。Flink目前定義瞭如下Backpressure狀態：

• OK: 0 <= Ratio <= 0.10
• LOW: 0.10 < Ratio <= 0.5
• HIGH: 0.5 < Ratio <= 1
  

另外，Flink還提供了3個引數來配置Backpressure監控行為：

引數名稱	預設值	說明
jobmanager.web.backpressure.refresh-interval	60000	預設1分鐘，表示取樣統計結果重新整理時間間隔
jobmanager.web.backpressure.num-samples	100	評估Backpressure狀態，所使用的堆疊跟蹤呼叫次數
jobmanager.web.backpressure.delay-between-samples	50	預設50毫秒，表示對一個Job的每個Task依次呼叫的時間間隔

通過上面個定義的Backpressure狀態，以及調整相應的引數，可以確定當前執行的Job的狀態是否正常，並且保證不影響JobManager提供服務。