Apache Flink on K8s：四種運行模式，我該選擇哪一種？

1. 前言

Apache Flink 是一個分佈式流處理引擎，它提供了豐富且易用的API來處理有狀態的流處理應用，而且在支持容錯的前提下，高效、大規模的運行此類應用。經過支持事件時間(event-time)、計算狀態(state)以及剛好一次(exactly-once)的容錯保證，Flink迅速被不少公司採納，成爲了新一代的流計算處理引擎。

2020 年 2 月 11 日，社區發佈了 Flink 1.10.0 版本, 該版本對性能和穩定性作了很大的提高，同時引入了 native Kubernetes 的特性。對於 Flink 的下一個穩定版本，社區在 2020 年 4 月底凍結新特性的合入，預計在 2020 年 5 月中旬會推出 Flink 1.11，在新版本中將重點引入新特性，以擴容 Flink 的使用場景。

1.1 Flink 爲何選擇 Kubernetes

Kubernetes 項目源自 Google 內部 Borg 項目，基於 Borg 多年來的優秀實踐和其超前的設計理念，並憑藉衆多豪門、大廠的背書，時至今日，Kubernetes 已經成長爲容器管理領域的事實標準。在大數據及相關領域，包括 Spark，Hive，Airflow，Kafka 等衆多知名產品正在遷往 Kubernetes，Apache Flink 也是其中一員。

Flink 選擇 Kubernetes 做爲其底層資源管理平臺，緣由包括兩個方面：

1）Flink 特性：流式服務通常是常駐進程，常常用於電信網質量監控、商業數據即席分析、實時風控和實時推薦等對穩定性要求比較高的場景；

2）Kubernetes 優點：爲在線業務提供了更好的發佈、管理機制，並保證其穩定運行，同時 Kubernetes 具備很好的生態優點，能很方便的和各類運維工具集成，如 prometheus 監控，主流的日誌採集工具等；同時 K8S 在資源彈性方面提供了很好的擴縮容機制，很大程度上提升了資源利用率。

1.2 Flink on Kubernetes 的發展歷史

在 Flink 的早期發行版 1.2 中，已經引入了 Flink Session 集羣模式，用戶得以將 Flink 集羣部署在 Kubernetes 集羣之上。

隨着 Flink 的逐漸普及，愈來愈多的 Flink 任務被提交在用戶的集羣中，用戶發如今 session 模式下，任務之間會互相影響，隔離性比較差，所以在 Flink 1.6 版本中，推出了 Per Job 模式，單個任務獨佔一個 Flink 集羣，很大的程度上提升了任務的穩定性。

在知足了穩定性以後，用戶以爲這兩種模式，沒有作到資源按需建立，每每須要憑用戶經驗來事先指定 Flink 集羣的規格，在這樣的背景之下，native session 模式應用而生，在 Flink 1.10 版本進入 Beta 階段，咱們增長了 native per job 模式，在資源按需申請的基礎上，提升了應用之間的隔離性。

本文根據 Flink 在 Kubernetes 集羣上的運行模式的趨勢，依次分析了這些模式的特色，並在最後介紹了 Flink operator 方案及其優點。

2. Flink運行模式

本文首先分析了 Apache Flink 1.10 在 Kubernetes 集羣上已經GA（生產可用）的兩種部署模式，而後分析了處於 Beta 版本的 native session 部署模式和即將在 Flink 1.11 發佈的 native per-job 部署模式，最後根據這些部署模式的利弊，介紹了當前比較 native kubernetes 的部署方式，flink-operator。

咱們正在使用的 Flink 版本已經很好的支持了 native session 和 native per-job 兩種模式，在 flink-operator 中，咱們也對這兩種模式也作了支持。

接下來將按照如下順序分析了 Flink 的運行模式，讀者能夠結合自身的業務場景，考量適合的 Flink 運行模式。

• Flink session 模式
• Flink per-job 模式
• Flink native session 模式
• Flink native per-job 模式

這四種部署模式的優缺點對比，能夠用以下表格來歸納，更多的內容，請參考接下來的詳細描述。

2.1 Session Cluster 模式

2.1.1 原理簡介

Session 模式下，Flink 集羣處於長期運行狀態，當集羣的Master組件接收到客戶端提交的任務後，對任務進行分析並處理。用戶將Flink集羣的資源描述文件提交到 Kubernetes 以後，Flink 集羣的 FlinkMaster 和 TaskManager 會被建立出來，以下圖所示，TaskManager 啓動後會向 ResourceManager 模塊註冊，這時 Flink Session 集羣已經準備就緒。當用戶經過 Flink Clint 端提交了 Job 任務時，Dispatcher 收到該任務請求，將請求轉發給 JobMaster，由 JobMaster 將任務分配給具體的 TaskManager。

2.1.2 特色分析

這種類型的 Flink 集羣，FlinkMaster 和 TaskManager 是以Kubernetes deployment的形式長期運行在 Kubernetes 集羣中。在提交做業以前，必須先建立好 Flink session 集羣。多個任務能夠同時運行在同一個集羣內，任務之間共享 K8sResourceManager 和 Dispatcher，可是 JobMaster 是單獨的。這種方式比較適合運行短時做業、即席查詢、任務提交頻繁、或者對任務啓動時長比較敏感的場景。

優勢：做業提交的時候，FlinkMaster 和 TaskManager已經準備好了，當資源充足時，做業可以當即被分配到 TaskManager 執行，無需等待 FlinkMaster，TaskManager，Service 等資源的建立；

缺點：1）須要在提交 Job 任務以前先建立 Flink 集羣，須要提早指定 TaskManager 的數量，可是在提交任務前，是難以精準把握具體資源需求的，指定的多了，會有大量 TaskManager 處於閒置狀態，資源利用率就比較低，指定的少了，則會有任務分配不到資源，只能等集羣中其餘做業執行完成後，釋放了資源，下一個做業纔會被正常執行。

2) 隔離性比較差，多個 Job 任務之間存在資源競爭，互相影響；若是一個 Job 異常致使 TaskManager crash 了，那麼全部運行在這個 TaskManager 上的 Job 任務都會被重啓；進而，更壞的狀況是，多個 Jobs 任務的重啓，大量併發的訪問文件系統，會致使其餘服務的不可用；最後一點是，在 Rest interface 上是能夠看到同一個 session 集羣裏其餘人的 Job 任務。

2.2 Per Job Cluster 模式

顧名思義，這種方式會專門爲每一個 Job 任務建立一個單獨的 Flink 集羣，當資源描述文件被提交到 Kubernetes 集羣, Kubernetes 會依次建立 FlinkMaster Deployment、TaskManagerDeployment 並運行任務，任務完成後，這些 Deployment 會被自動清理。

2.2.1 特色分析

優勢：隔離性比較好，任務之間資源不衝突，一個任務單獨使用一個 Flink 集羣；相對於 Flink session 集羣並且，資源隨用隨建，任務執行完成後馬上銷燬資源，資源利用率會高一些；

缺點：須要提早指定 TaskManager 的數量，若是 TaskManager 指定的少了會致使做業運行失敗，指定的多了仍會下降資源利用率；資源是實時建立的，用戶的做業在被運行前，須要先等待如下過程：

· Kubernetes scheduler爲FlinkMaster和 TaskManager 申請資源並調度到宿主機上進行建立；

· Kubernetes kubelet拉取FlinkMaster、TaskManager 鏡像，並建立出FlinkMaster、TaskManager容器；

· TaskManager啓動後，向Flink ResourceManager 註冊。

這種模式比較適合對啓動時間不敏感、且長時間運行的做業。不適合對任務啓動時間比較敏感的場景。

2.3 Native Session Cluster 模式

2.3.1 原理分析

1. Flink提供了Kubernetes模式的入口腳本 kubernetes-session.sh，當用戶執行了該腳本以後，Flink 客戶端會生成 Kubernets 資源描述文件，包括 FlinkMaster Service，FlinkMasterDeloyment，Configmap，Service並設置了owner reference，在 Flink 1.10 版本中，是將 FlinkMaster Service 做爲其餘資源的 Owner，也就意味着在刪除 Flink 集羣的時候，只須要刪除 FlinkMaster service，其餘資源則會被以及聯的方式自動刪除；
2. Kubernetes 收到來自 Flink 的資源描述請求後，開始建立FlinkMaster Service，FlinkMaster Deloyment，以及 Configmap 資源，從圖中能夠看到，伴隨着 FlinkMaster 的建立，Dispatch 和K8sResMngr 組件也同時被建立了，這裏的 K8sResMngr 就是 Native 方式的核心組件，正是這個組件去和 Kubernetes API server 進行通訊，申請 TaskManager 資源；當前，用戶已經能夠向Flink 集羣提交任務請求了；
3. 用戶經過 Flink client 向 Flink 集羣提交任務，flink client 會生成 Job graph，而後和 jar 包一塊兒上傳；當任務提交成功後，JobSubmitHandler 收到了請求並提交給 Dispatcher並生成 JobMaster, JobMaster 用於向 KubernetesResourceManager 申請 task 資源；
4. Kubernetes－Resource－Manager 會爲 taskmanager 生成一個新的配置文件，包含了 service 的地址，這樣當 Flink Master 異常重建後，能保證 taskmanager 經過 Service 仍然能鏈接到新的 Flink Master；
5. TaskManager 建立成功後註冊到 slotManager，這時 slotManager向TaskManager 申請 slots，TaskManager 提供本身的空閒 slots，任務被部署並運行；

2.3.2. 特色分析

以前咱們提到的兩種部署模式，在 Kubernetes 上運行 Flink 任務是須要事先指定好 TaskManager 的數量，可是大部分狀況下，用戶在任務啓動前是沒法準確的預知該任務所需的 TaskManager 數量和規格。

指定的多了會資源浪費，指定的少了會致使任務的執行失敗。最根本的緣由，就是沒有 Native 的使用 Kubernetes 資源，這裏的 Native，能夠理解爲 Flink 直接與 Kuberneter 通訊來申請資源。

這種類型的集羣，也是在提交任務以前就建立好了，不過只包含了 FlinkMaster 及其 Entrypoint（Service），當任務提交的時候，Flink client 會根據任務計算出並行度，進而肯定出所需 TaskManager 的數量，而後 Flink 內核會直接向 Kubernetes API server 申請 taskmanager，達到資源動態建立的目的。

• 優勢：相對於前兩種集羣而言，taskManager 的資源是實時的、按需進行的建立，對資源的利用率更高，所需資源更精準。
• 缺點：taskManager 是實時建立的，用戶的做業真正運行前, 與 Per Job集羣同樣, 仍須要先等待 taskManager 的建立, 所以對任務啓動時間比較敏感的用戶，須要進行必定的權衡。

2.4 Native Per Job 模式

在當前的 Apache Flink 1.10 版本里，Flink native per-job 特性還沒有發佈，預計在後續的 Flink 1.11 版本中提供，咱們能夠提早一覽 native per job 的特性。

2.4.1 原理分析

當任務被提交後，一樣由 Flink 來向 kubernetes 申請資源，其過程與以前提到的 native session 模式類似，不一樣之處在於：

1. Flink Master是隨着任務的提交而動態建立的；
2. 用戶能夠將 Flink、做業 Jar 包和 classpath 依賴打包到本身的鏡像裏；
3. 做業運行圖由 Flink Master 生成，因此無需經過 RestClient 上傳 Jar 包（圖 2 步驟 3）。

2.4.2. 特色分析

native per-job cluster 也是任務提交的時候才建立 Flink 集羣，不一樣的是，無需用戶指定 TaskManager 資源的數量，由於一樣藉助了 Native 的特性，Flink 直接與 Kubernetes 進行通訊並按需申請資源。

• 優勢：資源按需申請，適合一次性任務，任務執行後當即釋放資源，保證了資源的利用率；
• 缺點：資源是在任務提交後開始建立，一樣意味着對於提交任務後對延時比較敏感的場景，須要必定的權衡;

3. Flink-operator

3.1 簡介

分析以上四種部署模式，咱們發現，對於 Flink 集羣的使用，每每須要用戶自行維護部署腳本，向 Kubernetes 提交各類所需的底層資源描述文件（Flink Master，TaskManager，配置文件，Service）。

在 session cluster 下，若是集羣再也不使用，還須要用戶自行刪除這些的資源，由於這類集羣的資源使用了 Kubernetes 的垃圾回收機制 owner reference，在刪除 Flink 集羣的時候，須要經過刪除資源的 Owner 來進行及聯刪除，這對於不熟悉 Kubernetes 的 Flink 用戶來講，就顯得不是很友好了。

而經過 Flink-operator，咱們能夠把 Flink 集羣描述成 yaml 文件，這樣，藉助 Kubernetes 的聲明式特性和協調控制器，咱們能夠直接管理 Flink 集羣及其做業，而無需關注底層資源如 Deployment，Service，ConfigMap 的建立及維護。

當前 Flink 官方還未給出 flink-operator 方案，不過 GoogleCloudPlatform 提供了一種基於 kubebuilder 構建的 flink-operator方案。接下來，將介紹 flink-operator 的安裝方式和對 Flink 集羣的管理示例。

3.2 Flink-operator 原理及優點

當 Fink operator 部署至 Kubernetes 集羣后， FlinkCluster 資源和 Flink Controller 被建立。其中 FlinkCluster 用於描述 Flink 集羣，如 JobMaster 規格、TaskManager 和 TaskSlot 數量等；Flink Controller 實時處理針對 FlinkCluster 資源的 CRUD 操做，用戶能夠像管理內置 Kubernetes 資源同樣管理 Flink 集羣。

例如，用戶經過 yaml 文件描述指望的 Flink 集羣並向 Kubernetes 提交，Flink controller 分析用戶的 yaml，獲得 FlinkCluster CR，而後調用 API server 建立底層資源,如JobMaster Service, JobMaster Deployment，TaskManager Deployment。

經過使用 Flink Operator，有以下優點：

1. 管理 Flink 集羣更加便捷

flink-operator 更便於咱們管理 Flink 集羣，咱們不須要針對不一樣的 Flink 集羣維護 Kubenretes 底層各類資源的部署腳本，惟一須要的，就是 FlinkCluster 的一個自定義資源的描述文件。建立一個 Flink session 集羣，只須要一條 kubectl apply 命令便可，下圖是 Flink Session集羣的 yaml 文件，用戶只須要在該文件中聲明指望的 Flink 集羣配置，flink-operator 會自動完成 Flink 集羣的建立和維護工做。若是建立 Per Job 集羣，也只須要在該 yaml 中聲明 Job 的屬性，如 Job 名稱，Jar 包路徑便可。經過 flink-operator，上文提到的四種 Flink 運行模式，分別對應一個 yaml 文件便可，很是方便。

apiVersion: flinkoperator.k8s.io/v1beta1kind: FlinkClustermetadata:  name: flinksessioncluster-samplespec:  image:    name: flink:1.10.0    pullPolicy: IfNotPresent  jobManager:    accessScope: Cluster    ports:      ui: 8081    resources:      limits:        memory: "1024Mi"        cpu: "200m"  taskManager:    replicas: 1    resources:      limits:        memory: "2024Mi"        cpu: "200m"    volumes:      - name: cache-volume        emptyDir: {}    volumeMounts:      - mountPath: /cache        name: cache-volume  envVars:    - name: FOO      value: bar  flinkProperties:    taskmanager.numberOfTaskSlots: "1"

2. 聲明式

經過執行腳本命令式的建立 Flink 集羣各個底層資源，須要用戶保證資源是否依次建立成功，每每伴隨着輔助的檢查腳本。藉助 flink operator 的控制器模式，用戶只需聲明所指望的 Flink 集羣的狀態，剩下的工做所有由 Flink operator 來保證。在 Flink 集羣運行的過程當中，若是出現資源異常，如 JobMaster 意外中止甚至被刪除，Flink operator 都會重建這些資源，自動的修復 Flink 集羣。

3. 自定義保存點

用戶能夠指定 autoSavePointSeconds 和保存路徑，Flink operator 會自動爲用戶按期保存快照。

4. 自動恢復

流式任務每每是長期運行的，甚至 2-3 年不中止都是常見的。在任務執行的過程當中，可能會有各類各樣的緣由致使任務失敗。用戶能夠指定任務重啓策略，當指定爲 FromSavePointOnFailure，Flink operator 自動從最近的保存點從新執行任務。

5. sidecar containers

sidecar 容器也是 Kubernetes 提供的一種設計模式，用戶能夠在 TaskManager Pod 裏運行 sidecar 容器，爲 Job 提供輔助的自定義服務或者代理服務。

6. Ingress 集成

用戶能夠定義 Ingress 資源，flink operator 將會自動建立 Ingress 資源。雲廠商託管的 Kubernetes 集羣通常都有 Ingress 控制器，不然須要用戶自行實現 Ingress controller。

7. Prometheus 集成

經過在 Flink 集羣的 yaml 文件裏指定 metric exporter 和 metric port，能夠與 Kubernetes 集羣中的 Prometheus 進行集成。

最後

經過本文，咱們瞭解了 Flink 在 Kubernetes 上運行的不一樣模式，其中 Native 模式在資源按需申請方面比較突出，藉助 kubernetes operator，咱們能夠將 Flink 集羣當成Kubernetes原生的資源同樣進行 CRUD 操做。限於篇幅，本文主要分析了 Flink 在 Kubernetes 上的運行模式的區別，後續將會有更多的文章來對 Flink 在 Kubernetes 上的最佳實踐進行描述，敬請期待。

參考文檔

Kubernetes native integration

https://docs.google.com/docum...

Flink operator 使用文檔

https://github.com/tkestack/f...

【騰訊雲原生】雲說新品、雲研新術、雲遊新活、雲賞資訊，掃碼關注同名公衆號，及時獲取更多幹貨！！