在原始碼解讀前我們有必要先了解一下Flink的一些基本的但卻很關鍵的概念。這有助於幫助我們理解整個架構。在翻譯文件的同時，對於有爭議的或者不是非常適合用中文表達的地方，我儘量保留原始英文單詞。

程式和資料流

Flink程式的基本構建塊是streams和transformations（注意，DataSet在內部也是一個stream）。一個stream可以看成一箇中間結果，而一個transformations是以一個或多個stream作為輸入的某種operation，該operation利用這些stream進行計算從而產生一個或多個result stream。

在執行時，Flink上執行的程式會被對映成streaming dataflows

，它包含了streams和transformations operators。每一個dataflow以一個或多個sources開始以一個或多個sinks結束。dataflow類似於任意的有向無環圖（DAG），當然特定形式的環可以通過iteration構建。在大部分情況下，程式中的transformations跟dataflow中的operator是一一對應的關係。但有時候，一個transformation可能對應多個operator。

並行資料流

程式在Flink內部的執行具有並行、分散式的特性。stream被分割成stream partition，operator被分割成operator subtask

，這些operator subtasks在不同的執行緒、不同的物理機或不同的容器中彼此互不依賴得執行。

一個特定operator的subtask的個數被稱之為其parallelism(並行度)。一個stream的並行度總是等同於其producing operator的並行度。一個程式中，不同的operator可能具有不同的並行度。

Stream在operator之間傳輸資料的形式可以是one-to-one(forwarding)的模式也可以是redistributing的模式。

• One-to-one : strem(比如在source和map operator之間)維護著分割槽以及元素的順序。那意味著map operator的subtask看到的元素的個數以及順序跟source operator的subtask生產的元素的個數、順序相同。
• Redistributing : stream(map()跟keyBy/window之間或者keyBy/window跟sink之間)的分割槽會發生改變。每一個operator subtask依據所選擇的transformation傳送資料到不同的目標subtask。例如，keyBy() （基於hash碼重分割槽），broadcast()或者rebalance()（隨機redistribution）。在一個redistribution的交換中，只有每一個傳送、接收task對的順序才會被維持（比如map()的subtask和keyBy/window的subtask）。

tasks & operator chains

出於分散式執行的目的，Flink將operator的subtask連結在一起形成task。每個task在一個執行緒中執行。將operators連結成task是非常有效的優化：它能減少執行緒之間的切換和基於快取區的資料交換，在減少時延的同時提升吞吐量。連結的行為可以在程式設計API中進行指定。

下面這幅圖，展示了5個subtask以5個並行的執行緒來執行。

分散式執行

Master,Worker,Client

Flink執行時包含了兩種型別的處理器：

• master處理器：也稱之為JobManagers用於協調分散式執行。它們用來排程task，協調檢查點，協調失敗時恢復等。

Flink執行時至少存在一個master處理器。一個高可用的執行模式會存在多個master處理器，它們其中有一個是leader，而其他的都是standby。

• worker處理器：也稱之為TaskManagers用於執行一個dataflow的task(或者特殊的subtask)、資料緩衝和data stream的交換。

Flink執行時至少會存在一個worker處理器。

master和worker處理器可以以如下方式中的任意一種啟動：直接在物理機上啟動，通過容器，或者通過像YARN這樣的資源排程框架。worker連線到master，告知自身的可用性進而獲得任務分配。

客戶端不是執行時和程式執行的一部分。但它用於準備併發送dataflow給master。然後，客戶端斷開連線或者維持連線以等待接收計算結果。客戶端可以以兩種方式執行：要麼作為Java/Scala程式的一部分被程式觸發執行，要麼以命令列./bin/flink run的方式執行。

Workers,Slots,Resources

每一個worker(TaskManager)是一個JVM程序，它可能會在獨立的執行緒上執行一個或多個subtask。為了控制一個worker能接收多少個task。worker通過task slot來進行控制（一個worker至少有一個task slot）。

每個task slot表示TaskManager擁有資源的一個固定大小的子集。假如一個TaskManager有三個slot，那麼它會將其管理的記憶體分成三份給各個slot。資源slot化意味著一個subtask將不需要跟來自其他job的subtask競爭被管理的記憶體，取而代之的是它將擁有一定數量的記憶體儲備。需要注意的是，這裡不會涉及到CPU的隔離，slot目前僅僅用來隔離task的受管理的記憶體。

通過調整task slot的數量，允許使用者定義subtask之間如何互相隔離。如果一個TaskManager一個slot，那將意味著每個task group執行在獨立的JVM中（該JVM可能是通過一個特定的容器啟動的）。而一個TaskManager多個slot意味著更多的subtask可以共享同一個JVM。而在同一個JVM程序中的task將共享TCP連線（基於多路複用）和心跳訊息。它們也可能共享資料集和資料結構，因此這減少了每個task的負載。

預設，如果subtask是來自相同job，但不是相同的task，Flink允許subtask共享slot。結果是，一個slot可能hold住該job的整個pipeline。允許slot共享有兩個好處：

• Flink叢集確實需要許多task slots來讓Job達到最高的並行度。不需要計算一個程式總共包含多少個task。
• 更容易獲得更好的資源利用。如果沒有slot共享，非密集型的source/map()的subtask將阻塞跟密集型的window的subtask一樣多的佔用資源。而如果有slot共享，基本的併發度通過完整地利用共享的slot資源將獲得2到6倍的提升，同時仍然保證每一個TaskManager會在任務繁重的subtask之間進行合理的slot共享。

slot共享行為可以通過API來控制，以防止不合理的共享。這個機制稱之為resource groups，它定義了subtask可能共享的slot是什麼資源。

作為一個約定俗成的規則，task slot推薦的預設值是CPU的核數。基於超執行緒技術，每個slot佔用兩個或者更多的實際執行緒上下文。

時間和視窗

聚合事件（比如count,sum）工作起來比起批處理略微有些不同。例如，它不能一次完成對流中所有元素的數量統計，然後返回結果。因為流通常都是無限的（無邊界）。取而代之的是，在流上的聚合（count，sum等）被隔離到window域中，比如，“統計最近5分鐘的數量”或“對最近100個元素求和”。

視窗可以是時間驅動的（比如，每30秒）也可以是資料驅動的（比如，每100個元素）。通常我們將視窗劃分為：tumbing windows(不重疊)，sliding windows（有重疊）和session windows(有空隙的活動)。

時間

當在流式程式設計中涉及到時間的（比如定義一個視窗），可能會牽扯到時間的不同定義：

• Event Time：指一個事件的建立時間。通常在event中用時間戳來描述，比如，可能是由生產事件的感測器或生產服務來附加。Flink訪問事件時間戳通過時間戳分配器。
• Ingestion time：指一個事件從source operator進入Flink dataflow的時間。
• Processing time：每一個執行一個基於時間操作的operator的本地時間。

狀態和失敗容忍

在dataflow中的許多操作一次只關注一個獨立的事件（比如一個事件解析器），還有一些操作能記住多個獨立事件的資訊（比如，window operator），而這些操作被稱為stateful（有狀態的）。

有狀態的操作，其狀態被維護的地方，可以將其看作是一個內嵌的key/value儲存器。狀態和流一起被嚴格得分割槽和分佈以供有狀態的operator讀取。因此，訪問key/value的狀態僅能在keyed streams中（在執行keyBy()函式之後產生keyed stream），並且只能根據當前事件的鍵來訪問其值。對齊stream的鍵和狀態可以確保所有的狀態更新都是本地操作，在不需要事務開銷的情況下保證一致性。這個對齊機制也允許Flink重新分佈狀態並顯式調整stream的分割槽。

用於失敗容忍的檢查點

Flink實現失敗容忍使用了流重放和檢查點的混合機制。一個檢查點會在流和狀態中定義一個一致點，在該一致點streaming dataflow可以恢復並維持一致性（exactly-once的處理語義）。在最新的檢查點之後的事件或狀態更新將在input stream中被重放。

檢查點的設定間隔意味著在執行時對失敗容忍產生的額外開銷以及恢復時間（也決定了需要被重放的事件數）。

狀態的最終儲存

給key/value構建索引的資料結構最終被儲存的地方取決於狀態最終儲存的選擇。其中一個選擇是在記憶體中基於hash map，另一個是RocksDB。另外用來定義Hold住這些狀態的資料結構，狀態的最終儲存也實現了基於時間點的快照機制，給key/value做快照，並將快照作為檢查點的一部分來儲存。

基於流的批處理

Flink執行批處理程式是將其作為流處理程式的一個特例來看待。它將其看作有界的流（有限數量的元素）。DataSet在內部被當作一個流資料，因此上面的這些適用於流處理的這些概念在批處理中同樣適用，只有很少的幾個例外：

• DataSet的程式設計API不適用檢查點。恢復機制是通過重放完整的流資料來進行。那是合理的，因為輸入時有界的。它將開銷更多地引入到恢復操作上，但另一方面也使得執行時的常規流程代價更低，因為它規避了檢查點機制。
• DataSet的有狀態的operation API簡單地使用in-memory/out-of-core的資料結構，而不是基於key/value的索引機制
• DataSet的API引進了獨特的同步迭代機制（基於superstep），它僅在有界的流中存在。

微信掃描關注公眾號：Apache_Flink