[[416662]]

本文轉載自微信公眾號「數倉寶貝庫」，作者張利兵。轉載本文請聯系數倉寶貝庫公眾號。

一架構介紹

Flink系統架構主要分為APIs & Libraries、Core和Deploy三層，如圖1所示，其中APIs層主要實現了面向流處理對應的DataStream API，面向批處理對應的DataSet API。Libraries層也被稱作Flink應用組件層，是根據API層的劃分，在API層之上構建滿足了特定應用領域的計算框架，分別對應了面向流處理和面向批處理兩類，其中面向流處理支持CEP(復雜事件處理)、基于類似SQL的操作(基于Table的關系操作);面向批處理支持Flink ML(機器學習庫)、Gelly(圖處理)。運行時層提供了Flink計算的全部核心實現，例如支持分布式Stream作業執行、JobGraph到ExecutionGraph的映射和調度等，為API層提供了基礎服務。Deploy層支持多種部署模式，包括本地、集群(Standalone、YARN、Kubernetes)及云部署(GCE/EC2)。

圖1　Flink整體架構

1、編程接口

Flink提供了多種抽象的編程接口，適用于不同層級的用戶。數據分析人員和偏向業務的數據開發人員可以使用Flink SQL定義流式作業。如圖2所示，Flink編程接口分為4層。

圖2　Flink編程接口抽象

Flink SQL

一項大數據技術如果想被用戶接受和使用，除了應具有先進的架構理念之外，另一點非常重要的就是要具有非常好的易用性。我們知道雖然Pig中的操作更加靈活和高效，但是在都滿足數據處理需求的前提下，數據開發者更愿意選擇Hive作為大數據處理的開發工具。其中最重要的原因是，Hive能夠基于SQL標準進行拓展，提出了HQL語言，這就讓很多只會SQL的用戶也能夠快速掌握大數據處理技術。因此Hive技術很快得到普及。

對于Flink同樣如此，如果想贏得更多的用戶，就必須不斷增強易用性。FlinkSQL基于關系型概念構建流式和離線處理應用，使用戶能夠更加簡單地通過SQL構建Flink作業。

Table API

Flink SQL解析生成邏輯執行計劃和物理執行計劃，然后轉換為Table之間的操作，最終轉換為JobGraph并運行在集群上。Table API和Spark中的DataSet/DataFrame接口類似，都提供了面向領域語言的編程接口。相比Flink SQL，Table API更加靈活，既可以在Java & Scala SDK中與DataStream和DataSet API相互轉換，也能結合Flink SQL進行數據處理。

DataStream & DataSet API

在早期的Flink版本中，DataSet API和DataStream API分別用于流處理和批處理場景。DataSet用于處理離線數據集，DataStream用于處理流數據集。DataFlow模型希望使用同一套流處理框架統一處理有界和無界數據，那么為什么Flink還要抽象出兩套編程接口來處理有界數據集和無界數據集呢?這也是近年來Flink社區不斷探討的話題。目前Table和SQL API層面雖然已經能夠做到批流一體，但這僅是在邏輯層面上的，最終還是會轉換成DataSet API和DataStream API對應的作業。后期Flink社區將逐漸通過DataStream處理有界數據集和無界數據集，社區已經在1.11版本中對DataStream API中的SourceFunction接口進行了重構，使DataStream可以接入和處理有界數據集。在后期的版本中，Flink將逐步實現真正意義上的批流一體化。

Stateful Processing Function接口

Stateful Processing Function接口提供了強大且靈活的編程能力，在其中可以直接操作狀態數據、TimeService等服務，同時可以注冊事件時間和處理時間回調定時器，使程序能夠實現更加復雜的計算。使用Stateful Processing Function接口需要借助DataStream API。雖然Stateful Processing Function接口靈活度很高，但是接口使用復雜度也相對較高，且在DataStream API中已經基于Stateful Process Function接口封裝了非常豐富的算子，這些算子可以直接使用，因此，除非用戶需要自定義比較復雜的算子(如直接操作狀態數據等)，否則無須使用Stateful Processing Function接口開發Flink作業。

2、運行時執行引擎

用戶使用組件棧和接口編寫的Flink作業最終都會在客戶端轉換成JobGraph對象，然后提交到集群中運行。除了任務的提交和運行之外，運行時還包含資源管理器Resource-Manager以及負責接收和執行Task的TaskManager，這些服務各司其職，相互合作。運行時提供了不同類型(有界和無界)作業的執行和調度功能，最終將任務拆解成Task執行和調度。同時，運行時兼容了不同類型的集群資源管理器，可以提供不同的部署方式，并統一管理Slot計算資源。

3、物理部署層

物理部署層的主要功能是兼容不同的資源管理器，如支持集群部署模式的Hadoop YARN、Kubernetes及Standalone等。這些資源管理器能夠為在Flink運行時上運行的作業提供Slot計算資源。第4章會重點介紹Flink物理部署層的實現，幫助大家了解如何將運行時運行在不同的資源管理器上并對資源管理器提供的計算資源進行有效管理。

二Flink集群架構

如圖3所示，Flink集群主要包含3部分：JobManager、TaskManager和客戶端，三者均為獨立的JVM進程。Flink集群啟動后，會至少啟動一個JobManager和多個Task-Manager。客戶端將任務提交到JobManager，JobManager再將任務拆分成Task并調度到各個TaskManager中執行，最后TaskManager將Task執行的情況匯報給JobManager。

圖3　Flink集群架構圖

客戶端是Flink專門用于提交任務的客戶端實現，可以運行在任何設備上，并且兼容Windows、macOS、Linux等操作系統，只需要運行環境與JobManager之間保持網絡暢通即可。用戶可以通過./bin/f?link run命令或Scala Shell交互式命令行提交作業。客戶端會在內部運行提交的作業，然后基于作業的代碼邏輯構建JobGraph結構，最終將JobGraph提交到運行時中運行。JobGraph是客戶端和集群運行時之間約定的統一抽象數據結構，也就是說，不管是什么類型的作業，都會通過客戶端將提交的應用程序構建成JobGraph結構，最后提交到集群上運行。

JobManager是整個集群的管理節點，負責接收和執行來自客戶端提交的JobGraph。JobManager也會負責整個任務的Checkpoint協調工作，內部負責協調和調度提交的任務，并將JobGraph轉換為ExecutionGraph結構，然后通過調度器調度并執行ExecutionGraph的節點。ExecutionGraph中的ExecutionVertex節點會以Task的形式在TaskManager中執行。

除了對Job的調度和管理之外，JobManager會對整個集群的計算資源進行統一管理，所有TaskManager的計算資源都會注冊到JobManager節點中，然后分配給不同的任務使用。當然，JobManager還具備非常多的功能，例如Checkpoint的觸發和協調等。

TaskManager作為整個集群的工作節點，主要作用是向集群提供計算資源，每個TaskManager都包含一定數量的內存、CPU等計算資源。這些計算資源會被封裝成Slot資源卡槽，然后通過主節點中的ResourceManager組件進行統一協調和管理，而任務中并行的Task會被分配到Slot計算資源中。

根據底層集群資源管理器的不同，TaskManager的啟動方式及資源管理形式也會有所不同。例如，在基于Standalone模式的集群中，所有的TaskManager都是按照固定數量啟動的;而YARN、Kubernetes等資源管理器上創建的Flink集群則支持按需動態啟動TaskManager節點。

三核心概念

1、有狀態計算

在Flink架構體系中，有狀態計算是非常重要的特性之一。如圖4所示，有狀態計算是指在程序計算過程中，程序內部存儲計算產生的中間結果，并將其提供給后續的算子進行計算。狀態數據可以存儲在本地內存中，也可以存儲在第三方存儲介質中，例如Flink已經實現的RocksDB。

圖4　有狀態處理和無狀態處理

和有狀態計算不同，無狀態計算不會存儲計算過程中產生的結果，也不會將結果用于下一步計算。程序只會在當前的計算流程中執行，計算完成就輸出結果，然后接入下一條數據，繼續處理。

無狀態計算實現的復雜度相對較低，實現起來也比較容易，但是無法應對比較復雜的業務場景，例如處理實時CEP問題，按分鐘、小時、天進行聚合計算，求取最大值、均值等聚合指標等。如果不借助Flink內部提供的狀態存儲，一般都需要通過外部數據存儲介質，常見的有Redis等鍵值存儲系統，才能完成復雜指標的計算。

和Storm等流處理框架不同，Flink支持有狀態計算，可以應對更加復雜的數據計算場景。

2、時間概念與水位線機制

在DataFlow模型中，時間會被分為事件時間和處理時間兩種類型。如圖5所示，Flink中的時間概念基本和DataFlow模型一致，且Flink在以上兩種時間概念的基礎上增加了進入時間(ingestion time)的概念，也就是數據接入到Flink系統時由源節點產生的時間。

圖5　Flink時間概念

事件時間指的是每個事件在其生產設備上發生的時間。通常在進入Flink之前，事件時間就已經嵌入數據記錄，后續計算從每條記錄中提取該時間。基于事件時間，我們可以通過水位線對亂序事件進行處理。事件時間能夠準確地反映事件發生的先后關系，這對流處理系統而言是非常重要的。在涉及較多的網絡傳輸時，在傳輸過程中不可避免地會發生數據發送順序改變，最終導致流系統統計結果出現偏差，從而很難通過實時計算的方式得到正確的統計結果。

處理時間是指執行相應算子操作的機器系統時間。當應用基于處理時間運行時，所有基于時間的算子操作(如時間窗口)將使用運行相應算子機器的系統時鐘。例如，應用程序在上午9:15運行，則第一個每小時處理時間窗口包括在上午9:15到上午10:00之間處理的事件，下一個窗口包括在上午10:00到11:00之間處理的事件。

處理時間是最簡單的時間概念，不需要在流和機器之間進行協調，它提供了最佳的性能和最低的延遲。但在分布式和異步環境中，處理時間不能提供確定性，因為它容易受到記錄到達系統的速度(例如從消息隊列到達系統)以及系統內算子之間流動速度的影響。

接入時間是指數據接入Flink系統的時間，它由SourceOperator自動根據當前時鐘生成。后面所有與時間相關的Operator算子都能夠基于接入時間完成窗口統計等操作。接入時間的使用頻率并不高，當接入的事件不具有事件時間時，可以借助接入時間來處理數據。

相比于處理時間，接入時間的實現成本較高，但是它的數據只產生一次，且不同窗口操作可以基于統一的時間戳，這可以在一定程度上避免處理時間過度依賴處理算子的時鐘的問題。

不同于事件時間，接入時間不能完全刻畫出事件產生的先后關系。在Flink內部，接入時間只是像事件時間一樣對待和處理，會自動分配時間戳和生成水位線。因此，基于接入時間并不能完全處理亂序時間和遲到事件。

本文摘編于《Flink設計與實現：核心原理與源碼解析》，經出版方授權發布。