Flink-3-ApacheFlink架構

時間 2020-11-26

標籤 web 算法數據庫 apache 後端緩存網絡數據結構多線程架構欄目系統架構简体版

原文原文鏈接

第3章 Apache Flink架構

第3章 Apache Flink架構

參考書籍web

Stream Processing with Apache Flinkhttps://www.oreilly.com/library/view/stream-processing-with/9781491974285/算法

《基於Apache Flink的流處理》https://book.douban.com/subject/34912177/數據庫

注：本文主要是針對《基於Apache Flink的流處理》的筆記apache

1-8章筆記下載地址後端

在這一章中，咱們對Flink的架構進行了一個高層次的介紹，並描述了Flink如何解決咱們以前討論過的流處理相關問題。特別地，咱們重點解釋Flink的分佈式架構，展現它在流處理應用中是如何處理時間和狀態的，並討論了它的容錯機制。緩存

3.1 系統架構

Flink是一個用於狀態化並行數據流處理的分佈式系統。Flink設置由多個進程組成，這些進程一般分佈在多臺機器上運行。網絡

分佈式系統須要解決的常見挑戰是數據結構

集羣中計算資源的分配和管理
進程協調
持久和高可用性數據存儲
故障恢復

Flink自己並無實現全部這些功能。它只關注於其核心功能——分佈式數據流處理，可是利用了不少現有的開源中間件和框架來實現其餘非核心部分。多線程

Flink與集羣資源管理器(如Apache Mesos、YARN和Kubernetes)集成得很好，但也能夠配置爲做爲獨立集羣運行。
Flink不提供持久的分佈式存儲。相反，它利用了像HDFS這樣的分佈式文件系統或S3這樣的對象存儲。
對於高可用設置中的領導選舉，Flink依賴於Apache ZooKeeper。

3.1.1 搭建Flink所需的組件

Flink的搭建由四個不一樣的組件組成，它們一塊兒工做來執行流應用程序。這些組件是JobManager、ResourceManager、TaskManager和Dispatcher。因爲Flink是用Java和Scala實現的，因此全部組件都運行在Java虛擬機(jvm)上。各組成部分的職責將在下面四個子小節分別介紹。架構

3.1.1.1 JobManager

應用管理

JobManager是控制 單個應用程序執行的主進程，每一個應用程序由一個的JobManager控制。（一對一關係）

JobManager負責接收要執行的應用程序。該應用程序由一個所謂的JobGraph（一個邏輯數據流圖）和一個JAR文件組成（JAR文件捆綁了該應用程序全部必需的類、庫和其餘資源）。
JobManager將JobGraph轉換爲名爲ExecutionGraph的物理數據流圖，該數據流圖由可並行執行的任務組成。
JobManager從ResourceManager請求必要的資源(TaskManager槽)來執行任務。一旦它接收到足夠數量的TaskManager槽，它就會將ExecutionGraph的任務分配給執行它們的TaskManager。
在執行期間，JobManager負責全部須要集中協調的操做，如檢查點的協調。

3.1.1.2 RecourceManager

資源管理

Flink爲 不一樣的環境和資源提供者(如YARN、Mesos、Kubernetes和獨立部署)提供了多個資源管理器。

ResourceManager負責管理Flink的處理資源單元---TaskManager槽。
當JobManager請求TaskManager槽時，ResourceManager會命令某個帶有空閒槽的TaskManager將它的空閒槽提供給JobManager。
若是ResourceManager沒有足夠的槽來知足JobManager的請求，則ResourceManager能夠與資源提供者對話，讓資源提供者嘗試啓動更多的TaskManager。
ResourceManager還負責終止空閒的TaskManager以釋放計算資源。

3.1.1.3 TaskManager

工做進程，執行任務的

TaskManager是Flink的工做進程(worker process，工人)

一般，在一個Flink集羣中有多個TaskManager在運行。
每一個TaskManager提供必定數量的槽。槽的數量限制了TaskManager能夠執行的任務數量。
- 在TaskManager啓動以後，TaskManager將它的槽註冊到ResourceManager。
- 當JobManager請求槽的時候，根據ResourceManager的指示，TaskManager向JobManager提供一個或多個槽。
- 而後JobManager能夠將任務分配到槽中，讓TaskManager執行這些任務。
- 在執行期間，TaskManager與運行相同應用可是不一樣任務的其餘TaskManager交換數據。

3.1.1.4 Dispatcher

與用戶直接對話

Dispatcher提供一個REST接口讓用戶提交要執行的應用。

應用提交執行後，它將啓動JobManager，並將應用交給它來執行。
Dispatcher還運行一個web儀表盤來提供關於做業執行的信息。

3.1.1.5 總體架構圖

3.1.2 應用部署

Flink應用程序能夠以兩種不一樣的模式來部署。

3.1.2.1 框架模式

在這種模式下，Flink應用程序被打包到一個JAR文件中，並由客戶端提交給一個正在運行的服務。該服務能夠是Flink Dispatcher、Flink JobManager或YARN的ResourceManager。

若是應用程序被提交到JobManager，它將當即開始執行應用程序。
若是應用程序被提交給Dispatcher或YARN ResourceManager，它將啓動JobManager並移交應用程序，而後JobManager將開始執行應用程序。

3.1.2.2 庫模式

在這種模式下，Flink應用程序被綁定在一個應用程序特定的容器鏡像中，好比Docker鏡像。

該鏡像還包括運行JobManager和ResourceManager的代碼。
當容器從鏡像啓動時，它會自動啓動ResourceManager和JobManager，並執行綁定的應用程序。
第二個獨立於應用程序的鏡像用於部署TaskManager容器。
- 從這個鏡像啓動的容器會自動啓動TaskManager，它鏈接到ResourceManager並註冊它的槽。
- 一般，外部資源管理器(如Kubernetes)負責啓動鏡像，並負責在發生故障時從新啓動容器。

第一種模式比較傳統，第二種模式經常使用於微服務中。

3.1.3 任務執行

TaskManager能夠同時執行多個任務。

這些任務能夠

屬於同一算子(數據並行)、
不一樣算子(任務並行)的子任務
甚至是來自不一樣應用程序的子任務(應用並行)。

TaskManager提供固定數量的處理槽來控制它可以併發執行的任務的數量。一個處理槽能夠執行應用程序的某個算子的一個並行任務。下圖是一個TaskManager、處理槽、任務以及算子關係的例子。

左側是一個JobGraph（應用程序的非並行表示，邏輯圖）。

它由5個算子組成。
算子A和C是數據源，算子E是數據匯。

右側是一個ExecutionGraph（物理圖）

算子C和E的並行度爲2。其餘算子的並行度爲4。
因爲最大算子並行度是4個，應用程序至少須要4個可用的處理槽來執行。
給定兩個各有兩個處理槽的Taskmanager，就知足了這個需求。
JobManager將JobGraph擴展爲一個ExecutionGraph，並將任務分配給四個可用插槽。
並行度爲4的算子各自有4個並行任務，這些任務被分配給每一個槽。
運算符C和E的各自有兩個並行任務，分別被分配到槽1.1和2.1以及槽1.2和2.2。
將多個不一樣算子的任務 分配到同一個插槽的優勢是這些任務能夠在同一個進程中高效地交換數據，而不須要訪問網絡。

每一個TaskManager是一個JVM，而每一個Slot是JVM中的一個線程。TaskManager在同一個JVM進程中以多線程方式執行它的任務。線程比單獨的進程更輕量，通訊成本更低，但不會嚴格地將任務彼此隔離。所以，一個行爲不正常的任務能夠殺死整個TaskManager進程和運行在它上面的全部任務。

3.1.4 高可用性設置

流式應用程序一般設計爲24x7運行。所以，即便內部進程失敗，也不能中止運行。

而要想從失敗中恢復

系統首先須要從新啓動失敗的進程
其次，從新啓動應用程序並恢復其狀態。

本小節主要學習如何從新啓動失敗的進程。

3.1.4.1 TaskManager故障

下面舉例說明TaskManager故障應該如何處理

假設咱們的應用程序要以最大並行度爲8來執行，那麼四個TaskManager(每一個TaskManager提供兩個插槽)能夠知足咱們對並行度的需求。
若是其中一個TaskManager發生故障，可用插槽的數量將減小到6個。
在這種狀況下，JobManager將請求ResourceManager提供更多處理槽。
若是請求失敗，JobManager會按照必定的時間間隔連續地重啓應用。直到重啓成功（有足夠多的空閒插槽就能重啓成功）。

3.1.4.2 JobManager故障

比TaskManager失敗更具挑戰性的問題是JobManager失敗。

JobManager控制流應用程序的執行，並保存有關其執行的元數據，例如指向已完成檢查點的指針。
若是負責的JobManager進程失敗，流應用程序將沒法繼續處理。
這使得JobManager成爲Flink中的應用程序的一個單點失效組件（也就是若是這個組件失效，那麼整個系統失效）。

爲了克服這個問題，Flink支持一種高可用模式，該模式能夠在原始JobManager失效時將應用的管理權和應用的元數據 遷移到另外一個JobManager。

Flink的高可用模式 基於 ZooKeeper

它是一個分佈式系統，來提供分佈式協調和共識服務。
Flink使用ZooKeeper進行領袖選舉，並將其做爲一個高可用性和持久的數據存儲。
在高可用性模式下操做時，JobManager將JobGraph和全部必需的元數據(如應用程序的JAR文件)寫入遠程持久存儲系統。
此外，JobManager將一個指向存儲位置的指針 寫入ZooKeeper的數據存儲中。
在應用程序執行期間，JobManager接收各個任務檢查點的狀態句柄(存儲位置)。當檢查點完成後，JobManager將狀態寫入遠程存儲，並將指向此遠程存儲位置的指針寫入ZooKeeper。
所以，從JobManager故障中恢復所需的全部數據都存儲在遠程存儲中，而ZooKeeper持有指向存儲位置的指針。
圖3-3說明了這種設計。

當JobManager失敗時，接管它工做的新JobManager執行如下步驟:

它從ZooKeeper請求存儲位置，而後從遠程存儲中獲取JobGraph、JAR文件和應用程序最後一個檢查點的存儲位置。
它向ResourceManager請求處理槽以繼續執行應用程序。
它將從新啓動應用程序，並將其全部任務的狀態重置爲檢查點中的狀態值。

最後還有一個問題，當TaskManager或者JobManager失效時，誰會觸發它們的重啓？

在容器環境(如Kubernetes)中做爲庫部署運行應用程序時，失敗的JobManager或TaskManager容器一般由容器編排服務自動從新啓動。
在YARN或Mesos上運行時，Flink的其他進程將觸發JobManager或TaskManager進程的從新啓動。

3.2 Flink中的數據傳輸

在運行過程當中，應用的任務不斷地交換數據。TaskManager 負責將數據從發送任務發送到接收任務。TaskManager的網絡組件在發送記錄以前在緩衝區中收集記錄，就是說，記錄不是一個一個發送的，而是先緩存到緩衝區中而後一批一批發送。這種技術是有效使用網絡資源和實現高吞吐量的基礎。

每一個TaskManager都有一個 網絡緩衝池(默認大小爲32 KB)用於發送和接收數據。

若是發送方任務和接收方任務在不一樣的TaskManager進程中運行，則它們經過網絡通訊。
每對TaskManager維護一個永久的TCP鏈接來交換數據。
使用shuffle鏈接模式時，每一個發送方任務都須要可以向每一個接收方任務發送數據。TaskManager須要爲每一個接收任務提供一個專用的網絡緩衝區，此任務對應的發送方會向該緩衝區發送數據。

圖3-4顯示了這個架構。

在shuffle鏈接模式下，因爲接收端的並行度爲4，因此每一個發送端都須要4個網絡緩衝區來向接收端任務發送數據
因爲發送端的並行度也是4，因此每一個接收端也都須要4個網絡緩衝區來接受發送端發送的數據
同一個TaskManager中的緩存區會共用同一條網絡鏈接
在shuffle模式或者broadcast模式下，須要的緩衝區的大小將是並行度的平方級
Flink的網絡緩衝區的默認配置對於中小型的設置是足夠的。

當發送方任務和接收方任務在同一個TaskManager進程中運行時

發送方任務將傳出的記錄序列化到緩衝區中，並在緩衝區填滿後將其放入隊列中。
接收任務從隊列中獲取緩衝區，並對傳入的記錄進行反序列化。
所以，在同一TaskManager上運行的任務之間的數據傳輸不會致使網絡通訊。

3.2.1 基於信用值的流量控制

經過網絡鏈接發送單條記錄很低效，而且形成很大的開銷。緩衝是充分利用網絡鏈接的帶寬的關鍵。在流處理上下文中，緩衝的一個缺點是增長了延遲，由於記錄是在緩衝區中收集的，而不是當即發送的。

Flink實現了一個基於信用值的流控制機制，其工做原理以下。

接收任務向發送任務 授予必定的信用值，也就是告訴發送端爲了接收其數據，我爲你保留的緩衝區的大小
一旦發送方收到信用值通知，就會在信用值容許範圍內儘量多的傳輸緩衝數據，並會附帶上積壓量大小（已經填滿準備傳輸的網絡緩衝數目）
接收方使用預留的緩衝來處理髮送的數據，同時依據各發送端的積壓量信息來計算全部發送方在下一輪的信用值分別是多少。

基於信用值的好處

基於信用的流控制減小了延遲，由於一旦接收方有足夠的資源接受數據，發送方就能夠發送數據。
此外，在數據分佈不均的狀況下，它是一種有效的分配網絡資源的機制，由於信用是根據發送方的積壓的大小授予的。
所以，基於信用的流控制是Flink實現高吞吐和低延遲的重要一環。

3.2.2 任務連接

Flink提供了一種被稱爲任務連接的優化技術，它能夠減小特定條件下本地通訊的開銷。

爲了知足任務連接的要求，被連接的全部算子必須配置相同的並行性，並經過本地轉發通道進行鏈接。
圖3-5所示的操做管道知足這些要求。它由三個算子組成，它們都被配置爲任務並行度爲2，並與本地轉發鏈接鏈接。

圖3-6描述瞭如何在任務連接模式下執行管道。

多個算子函數被融合到單個任務中，由單個線程執行。
經過一個簡單的方法調用，一個函數產生的記錄被單獨地移交給下一個函數。
所以，在函數之間傳遞記錄基本上沒有序列化開銷和沒有通訊開銷。

Flink在默認狀況下會開啓任務連接，可是也能夠經過配置關閉這個功能

3.3 事件時間處理

正如上一節所述，事件時間語義會生成可重複且一致性的結果，這是許多流應用的剛性需求。下面，咱們將描述Flink如何在內部實現和處理事件時間戳和水位線，以支持具備事件時間語義的流應用。

3.3.1 時間戳

Flink事件時間流應用處理的全部記錄都必須帶時間戳。時間戳將記錄與特定的時間點關聯起來，一般是記錄所表示的事件發生的時間點。此外，在現實環境中，時間戳亂序幾乎不可避免。

當Flink以事件時間模式處理數據流時，它會根據記錄的事件時間戳來觸發基於時間的算子操做。

例如，時間窗口操做符根據相關的時間戳將記錄分配給窗口。
Flink將時間戳編碼爲8字節長的Long值，並將它們做爲元數據附加到記錄中。
而後內置算子或者用戶自定義的算子解析這個Long值就能夠得到事件時間。

3.3.2 水位線

水位線用於標註事件時間應用程序中每一個任務當前的事件時間。

基於時間的操做符使用這段時間來觸發相關的計算計算並推進這個流進行。
例如，基於時間窗口的任務會在水位線超過窗口邊界的時候觸發計算而且發出結果

在Flink中，水位線被實現爲一種帶時間戳的特殊記錄。如圖3-8所示，水位線像常規記錄同樣在數據流中移動。

水位線有兩個基本特性:

水位線必須是單調遞增的，以確保任務的事件時間時鐘是前進的，而不是向後的。
水位線與記錄的時間戳存在關係。一個時間戳爲T的水位線表示：全部後續記錄的時間戳都應該大於T。

第二個屬性用於處理數據流中時間戳亂序的記錄，例如圖3-8中具備時間戳2和5的記錄。

基於時間的算子任務可能會處理帶有無序時間戳的記錄，每一個任務都會維護一個本身的事件時鐘，並經過時間戳來更新這個時鐘。
任務有可能接收到違反水位線屬性且時間戳 小於先前接收的水位線的記錄，該記錄所屬的計算可能已經完成。這樣的記錄稱爲遲到記錄。

水位線的一個意義是，它們容許應用控制結果的完整性和延遲。

3.3.3 水位線傳播和事件時間

在本節中，咱們將討論算子如何處理水位線。

Flink將水位線實現爲算子任務接收和發出的特殊記錄。
任務內部的時間服務會維護一些計時器(Timer)，任務能夠在計時器服務上註冊計時器，以便未來在特定的時間點執行計算，這些計時器依靠收到的水位線來激活。
例如，窗口操做符爲每一個活動窗口註冊一個計時器，當事件時間超過窗口的結束時間時，計時器將清除窗口的狀態。

當一個任務收到水位線時，會發生如下操做:

任務根據水位線的時間戳 更新其內部事件時間時鐘。
任務的時間服務根據更新後的時鐘來執行那些超時計時器的回調。對於每一個過時的計時器，任務將調用一個回調函數，該函數能夠執行計算併發出記錄。
任務根據更新後的時鐘向下遊任務發送水位線。

考慮到任務並行，咱們將詳細介紹一個任務如何將水位線發送到多個下游任務，以及它從多個上游任務獲取水位線以後如何推進事件時間時鐘前進。具體的方式以下

任務爲每一個輸入分區 維護分區水位線。
當它從一個分區接收到水位線時，它將相應的分區水位線 更新爲接收值和當前值的最大值。
隨後，任務將其內部事件時間時鐘 更新爲全部分區水印的最小值。
若是事件時間時鐘前進，任務處理全部觸發的計時器，最後經過向全部鏈接的輸出分區 發出更新後的水位線，向全部下游任務廣播它的新事件時間。

下圖舉了一個有4個輸入分區和3個輸出分區的任務在接受到水位線以後是如何更新它的分區水位線和事件時間時鐘的。

Flink的水位線傳播算法確保算子任務所發出帶時間戳的記錄和水位線必定會對齊。

然而，它依賴於這樣一個事實，即全部的分區都不斷地提供自增的水位線。
一旦一個分區不推動它的水位線，或者變成徹底空閒而再也不發送任何記錄和水位線，任務的事件時間時鐘將不會推動，進而致使計時器不會觸發。
所以，若是一個任務沒有按期從全部輸入任務接收到新的水位線，那麼任務的處理延遲和狀態大小會顯著增長。

對於具備兩個輸入流且水位線差距很大的算子，也會出現相似的效果。具備兩個輸入流的任務的事件時間時鐘將受制於較慢的流，一般較快的流的記錄或中間結果將處於緩衝狀態，直到事件時間時鐘容許處理它們。

3.3.4 時間戳分配和水位線生成

下面介紹時間戳和水位線是如何產生的。

時間戳和水位線一般是在流應用接收數據流時 分配和生成的。Flink DataStream應用能夠經過三種方式完成該工做

在數據源完成：當一個流被讀入到一個應用中時。數據源算子將產生帶有時間戳的記錄流。水位線能夠做爲特殊記錄在任什麼時候間點發出。若是數據源暫時再也不發出水位線了，能夠將本身聲明爲空閒，Flink會在後續算子計算水位線時將那些來自空閒數據源的流分區排除在外。
週期性分配器(Periodic Assigner)：這個Assigner能夠從每一個記錄中提取一個時間戳，並按期查詢當前的水位線。提取到的時間戳被分配給相應的記錄，所查詢的水印被加入到流中。
定點分配器(Punctuated Assigner)：它能夠用於根據特殊輸入記錄來生成水位線

3.4 狀態管理

大多數流應用是有狀態的。許多算子不斷讀取和更新某種狀態。無論是內置狀態仍是用戶自定義狀態，Flink的處理方式都是同樣的。

在本節中，咱們將討論

Flink支持的不一樣類型的狀態。
狀態後端如何存儲和維護狀態
有狀態應用程序如何經過進行狀態再分配來實現擴縮容。

一般，須要任務去維護並用於計算結果的數據都屬於任務的狀態。圖3-10顯示了任務與其狀態之間的典型交互。

任務接收一些輸入數據。
在處理數據時，任務能夠讀取和更新其狀態，
並根據其輸入數據和狀態計算其結果。

然而，高效可靠的狀態管理更具挑戰性。這包括處理很是大的狀態(可能超過內存)，並確保在發生故障時不會丟失任何狀態。全部與狀態一致性、故障處理、高效存儲和訪問相關的問題都由Flink處理，以便開發人員可以將重點放在應用程序的邏輯上。

在Flink中，狀態老是與一個特定的算子相關聯。爲了讓Flink的運行時知道算子有哪些狀態，算子須要對其狀態進行註冊。根據做用域的不一樣，有兩種類型的狀態：算子狀態和鍵值分區狀態

3.4.1 算子狀態

算子狀態的做用域爲算子的單個任務。這意味着由同一並行任務以內的記錄均可以訪問同一狀態。算子狀態不能被其餘任務訪問。以下圖

Flink爲算子狀態提供了三類原語

列表狀態：將狀態表示爲一個條目列表
聯合列表狀態：一樣將狀態表示爲一個條目列表。可是，在出現故障或從保存點啓動應用程序時，它的恢復方式與常規列表狀態不一樣。
廣播狀態：專門爲哪些須要保證算子的每一個任務狀態都相同的場景而設計

3.4.2 鍵值分區狀態

鍵值分區狀態是根據算子輸入記錄中定義的鍵來維護和訪問的。Flink爲每一個鍵維護一個狀態實例，該狀態實例老是位於那個處理對應鍵值記錄的任務上。當任務處理一個記錄時，它自動將狀態訪問範圍限制到當前記錄的鍵。所以，具備相同鍵值分區的全部記錄都訪問相同的狀態。圖3-12顯示了任務如何與鍵值分區狀態交互。

鍵值分區狀態是一個在算子的全部並行任務上進行分區的分佈式鍵值映射。鍵值分區狀態原語以下

單值狀態：爲每一個鍵存儲一個任意類型的值。該值能夠是一個任意複雜的數據結構。
列表狀態：爲每一個鍵儲存一個列表。列表條目能夠是任意類型。
映射狀態：爲每一個鍵存儲鍵值映射。映射的鍵和值能夠是任意類型。

3.4.3 狀態後端

爲了確保快速的狀態訪問，每一個並行任務都在本地維護其狀態。至於狀態的具體存儲、訪問和維護，則由一個稱爲狀態後端的可拔插組件來完成。

狀態後端負責兩件事:

本地狀態管理
將狀態以檢查點的形式寫入遠程存儲

對於本地狀態管理，Flink提供兩種實現

第一種狀態後端，將狀態做爲存儲在JVM堆內存數據結構中的對象進行管理。
第二種狀態後端，序列化狀態對象並將它們放入RocksDB中，這種方式是基於硬盤的。
雖然第一種實現提供很是快的訪問速度，但它受到內存空間大小的限制。訪問RocksDB會比較慢，可是空間大。

狀態檢查點很重要，由於Flink是一個分佈式系統，狀態只能在本地維護。TaskManager進程可能在任什麼時候間點失敗。所以，它的存儲必須被認爲是易失的。狀態後端負責將任務的狀態檢查點指向遠程和持久存儲。用於檢查點的遠程存儲能夠是分佈式文件系統或數據庫系統。狀態後端在狀態檢查點的方式上有所不一樣。例如，RocksDB狀態後端支持增量檢查點，這能夠顯著減小很是大的狀態的檢查點開銷。

3.4.4 有狀態的算子的擴縮容

流應用的一個基本需求是根據輸入速率的增長或減小而調整算子的並行性。有狀態算子，調整並行度比較難。由於咱們須要把狀態從新分組，分配到與以前數量不等的並行任務上。

3.4.4.1 帶有鍵值分區狀態的算子擴縮容

帶有鍵值分區狀態的算子能夠經過將鍵從新劃分來進行任務的擴縮容。可是，爲了提升效率，Flink不會以鍵爲單位來進行劃分。相反，Flink以鍵組做爲單位來從新分配，每一個鍵組裏面包含了多個鍵。

3.4.4.2 帶有算子列表狀態的算子擴縮容

帶有算子列表狀態的算子在擴縮容時會對列表中的條目進行從新分配。理論上來講，全部並行任務的列表項會被統一收集起來，並再均勻從新分配。若是列表項的數量少於算子的新並行度，一些任務將以空狀態開始。圖3-14顯示了操做符列表狀態的從新分配。

3.4.4.3 帶有算子聯合狀態的算子擴縮容

帶有算子聯合狀態的算子會在擴縮容時把狀態列表中的所有條目 廣播到所有任務中。而後，任務本身來選擇使用哪些項和丟棄哪些項。如圖3-15顯示。

3.4.4.4 帶有算子廣播狀態的算子擴縮容

帶有算子廣播狀態的算子在擴縮容時會把狀態拷貝到所有新任務上。這樣作是由於廣播狀態要確保全部任務具備相同的狀態。在縮容的狀況下，直接簡單地停掉多餘的任務便可。如圖3-16顯示。

3.5 檢查點、保存點、狀態恢復

Flink是一個分佈式的數據處理系統，且任務在本地維護它們的狀態，Flink必須確保這種狀態不會丟失，而且在發生故障時保持一致。

在本節中，咱們將介紹Flink的檢查點和故障恢復機制，看一下它們是如何提供精確一次的狀態一致性保障。此外，咱們還討論了Flink獨特的保存點(savepoint)功能，它就像一把瑞士軍刀，解決了運行流式應用過程當中的諸多難題。

3.5.1 一致性檢查點

有狀態流應用程序的一致檢查點是在全部任務都處理完等量的原始輸出後對所有任務狀態進行的一個拷貝。咱們能夠經過一個樸素算法來對應用創建一致性檢查點的過程進行解釋。樸素算法的步驟爲:

暫停接收全部輸入流。
等待全部流入系統的數據被徹底處理，即全部任務已經處理完全部的輸入數據。
將全部任務的狀態複製到遠程持久存儲，生成檢查點。當全部任務拷貝完成後，檢查點就完成了
恢復接收全部輸入流。

下圖展現了一個一致性檢查點的例子，這個算法讀取數據，而後對奇數和偶數分別求和

3.5.2 從一致性檢查點中恢復

在流應用執行期間，Flink週期性爲應用程序生成檢查點。一旦發生故障，Flink會使用最新的檢查點將應用狀態恢復到某個一致性的點並重啓應用。圖3-18顯示了恢復過程。

應用程序恢復分爲三個步驟:

重啓整個應用程序。
將全部狀態重置爲最新的檢查點。
恢復全部任務的處理。

假設全部算子都將它們的狀態寫入檢查點並從中恢復，而且全部輸入流的消費位置都能重置到檢查點生成那一刻，那麼這種檢查點和恢復機制能夠爲整個應用提供精確一次的一致性保障。輸入流是否能夠重置，取決於它的具體實現以及所消費外部系統是否提供相關接口。例如，像Apache Kafka這樣的事件日誌能夠從以前的某個偏移讀取記錄。相反，若是是從socket消費而來則沒法重置，由於socket一旦消耗了數據就會丟棄數據。

咱們必須指出，Flink的檢查點和恢復機制只能重置流應用內部的狀態。根據應用所採用的數據彙算子，在恢復期間，某些結果記錄可能被屢次發送到下游系統，例如事件日誌、文件系統或數據庫。對於某些存儲系統，Flink提供的數據匯能夠保證了精確一次輸出。

3.5.3 Flink檢查點算法

Flink基於Chandy-Lamport的分佈式快照算法來實現檢查點。該算法並不會暫停整個應用程序，在部分任務持久化狀態的過程當中，其餘任務能夠繼續執行。

Flink的檢查點算法使用一種稱爲檢查點分隔符的特殊類型的記錄，它與水位線相似。檢查點分隔符攜帶一個檢查點ID來標識它所屬的檢查點，分隔符從邏輯上將流分割爲兩個部分。由檢查點以前的記錄 引發的全部狀態修改都包含在分隔符對應的檢查點中，而由屏障以後的記錄引發的全部修改都不包含在分隔符對應的檢查點中。

下面咱們經過一個簡單的例子來解釋這個算法

咱們使用一個簡單的流應用程序示例逐步解釋該算法。應用程序由兩個數據源任務組成，每一個數據源任務消耗一個不斷增加的數字流。數據源任務的分別輸出奇數分區和偶數分區。每一個分區都由一個任務處理，該任務計算全部接收到的數字的總和，並將更新後的總和發送給下游數據匯。該應用程序如圖3-19所示。

JobManager經過向每一個數據源任務 發送一個新的帶有檢查點編號的消息來啓動檢查點生成流程，如圖3-20所示。

當數據源任務接收到檢查點消息時，

它暫停處理數據流，並利用狀態後端 生成本地狀態的檢查點，併發送到遠程存儲
把該檢查點分隔符廣播至全部下游任務。
狀態後端會在檢查點保存好以後通知TaskManager，TaskManager會給JobManager發送確認消息。
在發出了分隔符以後，數據源將恢復正常的工做狀態。
以下圖所示

數據源發出的檢查點分隔符被廣播給下游任務。當下游任務接收到新的檢查點分隔符時，將繼續等待來自全部其餘上游任務的分隔符到達檢查點。在等待期間，它繼續處理那些還沒有提供分隔符的上游任務的記錄，而那些提供了分隔符的上游任務的記錄會被緩存，等待稍後處理。等待全部檢查點到達的過程稱爲檢查點對齊，如圖3-22所示。

一旦一個任務從它的全部上游任務收到分隔符，它就會讓狀態後端生成一個檢查點，並將檢查點分隔符廣播給它的全部下游任務，如圖3-23所示。

在發出檢查點分隔符後，任務就開始處理緩衝的記錄。在處理完全部緩衝記錄以後，任務會繼續處理其輸入流。圖3-24顯示了此時的應用程序。

最後，檢查點分隔符到達數據匯。當數據匯接收到分割符時，會先進行對齊操做，而後將自身狀態寫入檢查點，並向JobManager確認接收到該分隔符。一旦應用的全部任務都發送了檢查點確認，JobManager就會將應用程序的檢查點記錄爲已完成。圖3-25顯示了檢查點算法的最後一步。如前所述，已完成的檢查點可用於從故障中恢復應用。

3.5.4 檢查點對性能的影響

Flink的檢查點算法從流應用中產生一致的分佈式檢查點，而不會中止整個應用。可是，它會增長應用的處理延遲。Flink實現了一些調整，能夠在某些條件下減輕性能影響。

任務在將其狀態寫入檢查點的過程當中，將被阻塞。一種好的方法是先將檢查點寫入本地，而後任務繼續執行它的常規處理，另外一個進程負責將檢查點傳到遠端存儲。

此外，還能夠在分隔符對齊的過程當中不緩存那些已經收到分隔符所對應分區的記錄，而是直接處理。但這會讓一致性保證從精確一次下降到至少一次

3.5.5 保存點

Flink最有價值和最獨特的功能之一是保存點。原則上，保存點的生成算法與檢查點生成算法同樣，所以能夠把保存點看做是帶有一些額外元數據的檢查點。Flink不會自動生成保存點，而是須要用戶顯式的調用來生成保存點。

3.5.5.1 保存點的使用

給定一個應用和一個兼容的保存點，咱們能夠從該保存點啓動應用。這將把應用的狀態初始化爲保存點的狀態，並從獲取保存點的位置運行應用。

保存點能夠用在不少狀況

能夠從保存點啓動一個不一樣但兼容的應用程序。這意味着能夠修復一些小bug以後從保存點重啓
可使用不一樣的並行度啓動原應用
能夠在不一樣的集羣上啓動原應用
可使用保存點暫停應用程序並在稍後恢復它。這樣就能夠爲其餘高優先級的應用騰出集羣資源
能夠用保存點來完成歸檔操做

3.5.5.2 從保存點啓動應用

在本節中，咱們將描述Flink在從保存點啓動時如何去初始化應用狀態。

一個典型的應用程序包含多個狀態，它們分佈在不一樣算子的不一樣任務上。

下圖顯示了一個具備三個算子的應用程序，每一個算子各運行兩個任務。其中一個算子(OP-1)有一個算子狀態(OS-1)，另外一個算子(OP-2)有兩個鍵值分區狀態(KS-1和KS-2)。當生成保存點時，全部任務的狀態都會被複制到一個持久化存儲位置上。

保存點中的狀態副本會按照算子標識符和狀態名稱進行組織。該算子標識符和狀態名須要可以將保存點的狀態數據映射到應用啓動後的狀態上。當從保存點啓動應用程序時，Flink將保存點數據從新分發給相應算子的任務。

若是應用發生了修改，只有那些算子標識符和狀態名稱沒變的狀態副本才能被成功還原。默認狀況下，Flink會分配惟一的算子標識符。可是，算子的標識符是基於其前面算子的標識符生成的。這樣，假如上游的算子標識符發生了變化，那麼下游的算子也會變化。所以，咱們強烈建議爲操做符手動分配惟一標識符，而不依賴於Flink的默認賦值。