全球今亮點！Application模式和Session模式有什么區別？

總結下來就幾點：

1、Native模式比Standalone模式好

Standalone模式需要提前確認好每個任務需要使用的資源，并在配置文件里面配置，每一個任務都是固定資源大小，申請多了浪費，少了怕出問題。

(資料圖)

Native模式不需要預先確定需要使用的資源數量，系統會實時根據任務需要自動去k8s集群申請能申請到的資源。

2、Application和Session模式各有優劣，不同情況使用不同模式

Application模式資源隔離性強，每個人物都是單獨的集群，不會出現并發問題。每個任務都需要啟動一個集群，會先啟動JobManager，然后啟動TaskManager，效率會比較低。適合流處理任務

對比yarn環境下的perjob提交任務速度快很多，大約是十幾秒就能提交執行；yarn環境下提交任務需要一分多鐘。

Session模式需要提前創建好集群，所有任務共享集群資源，并發下可能會有問題。共用集群，只需要啟動TaskManager，效率高。適合批處理任務

operator模式的利弊

還有一種方式叫operator模式，這種方式的優點是有一個開源服務，這個服務來幫你管理yml配置文件，你不需要自己去管理各種資源的配置。但是需要單獨啟動這個服務，然后調用這個服務的api去管理yml文件的配置功能。

優點：

管理 Flink 集群更加便捷

flink-operator 更便于我們管理 Flink 集群，我們不需要針對不同的 Flink 集群維護 Kubenretes 底層各種資源的部署腳本，唯一需要的，就是 FlinkCluster 的一個自定義資源的描述文件。用戶只需要在該文件中聲明期望的 Flink 集群配置，flink-operator 會自動完成 Flink 集群的創建和維護工作。如果創建 Per Job 集群，也只需要在該 yaml 中聲明 Job 的屬性，如 Job 名稱，Jar 包路徑即可。通過 flink-operator，上文提到的四種 Flink 運行模式，分別對應一個 yaml 文件即可，非常方便。

聲明式

通過執行腳本命令式的創建 Flink 集群各個底層資源，需要用戶保證資源是否依次創建成功，往往伴隨著輔助的檢查腳本。借助 flink operator 的控制器模式，用戶只需聲明所期望的 Flink 集群的狀態，剩下的工作全部由 Flink operator 來保證。在 Flink 集群運行的過程中，如果出現資源異常，如 JobMaster 意外停止甚至被刪除，Flink operator 都會重建這些資源，自動的修復 Flink 集群。

自定義保存點

用戶可以指定 autoSavePointSeconds 和保存路徑，Flink operator 會自動為用戶定期保存快照。

自動恢復

流式任務往往是長期運行的，甚至 2-3 年不停止都是常見的。在任務執行的過程中，可能會有各種各樣的原因導致任務失敗。用戶可以指定任務重啟策略，當指定為 FromSavePointOnFailure，Flink operator 自動從最近的保存點重新執行任務。

Ingress 集成

用戶可以定義 Ingress 資源，flink operator 將會自動創建 Ingress 資源。云廠商托管的 Kubernetes 集群一般都有 Ingress 控制器，否則需要用戶自行實現 Ingress controller。

Prometheus 集成

通過在 Flink 集群的 yaml 文件里指定 metric exporter 和 metric port，可以與 Kubernetes 集群中的 Prometheus 進行集成。

缺點：

需要單獨啟動一個服務

它的很多優點基于api的方式也能實現

3、啟動方式

Standalone模式：定義好配置文件，然后通過kubectl命令去創建集群，目前沒找到api方式創建

Native模式：

通過flink客戶端去創建集群

也可以使用api的方式去創建

Flink On Kubernetes 的部署演進

Flink 在 K8s 上最簡單的方式是以 Standalone 方式進行部署。這種方式部署的好處在于不需要對 Flink 做任何改動，同時 Flink 對 K8s 集群是無感知的，通過外部手段即可讓 Flink 運行起來。

Standalone Session On K8s

Standalone方式在k8s運行步驟：

如圖所示：

步驟1， 使用 Kubectl 或者 K8s 的 Dashboard 提交請求到 K8s Master。

步驟2， K8s Master 將創建 Flink Master Deployment、TaskManager Deployment、ConfigMap、SVC 的請求分發給 Slave 去創建這四個角色，創建完成后，這時 Flink Master、TaskManager 啟動了。步驟3， TaskManager 注冊到 JobManager。在非 HA 的情況下，是通過內部 Service 注冊到 JobManager。至此，Flink 的 Sesion Cluster 已經創建起來。此時就可以提交任務了。步驟4，在 Flink Cluster 上提交 Flink run 的命令，通過指定 Flink Master 的地址，將相應任務提交上來，用戶的 Jar 和 JobGrapth 會在 Flink Client 生成，通過 SVC 傳給 Dispatcher。步驟5，Dispatcher 會發現有一個新的 Job 提交上來，這時會起一個新的 JobMaster，去運行這個 Job。步驟6，JobMaster 會向 ResourceManager 申請資源，因為 Standalone 方式并不具備主動申請資源的能力，所以這個時候會直接返回，而且我們已經提前把 TaskManager 起好，并且已經注冊回來了。步驟7-8，這時 JobMaster 會把 Task 部署到相應的 TaskManager 上，整個任務運行的過程就完成了。

//創建session集群kubectl create -f flink-configuration-configmap.yamlkubectl create -f jobmanager-service.yamlkubectl create -f jobmanager-rest-service.yamlkubectl create -f jobmanager-deployment.yamlkubectl create -f taskmanager-deployment.yaml//提交任務到集群./bin/flink run -m localhost:8081 ./examples/streaming/WordCount.jar

Standalone perjob on K8s

現在我們看一下 Perjob 的部署，因為 Session Cluster 和 Perjob 分別都有不同的適用場景，一個 Session 里面可以跑多個任務，但是每個任務之間沒有辦法達到更好的隔離性。而 Perjob 的方式，每個job都會有一個自己獨立的 Flink Cluster 去運行，它們之間相互獨立。

■ Perjob 的特點:

用戶的 Jar 和依賴都是在鏡像里提前編譯好，或者通過 Init Container 方式，在真正 Container 啟動之前進行初始化。每個 Job 都會啟動一個新的 Cluster。一步提交，不需要像 Session Cluster 一樣先啟動集群再提交任務。用戶的 main 方法是在 Cluster 里運行。在特殊網絡環境情況下，main 方法需要在 Cluster 里運行的話，Session 方式是無法做到的，而 Perjob 方式是可以執行的。

■ 執行步驟：

由 Standalone JobCluster EntryPoint 執行，從 classpath 找到用戶 Jar，執行它的 main 方法得到 JobGrapth 。再提交到 Dispathcher，這時候走 Recover Job 的邏輯，提交到 JobMaster。JobMaster 向 ResourceManager 申請資源，請求 slot，執行 Job。

kubectl create -f flink-configuration-configmap.yamlkubectl create -f jobmanager-service.yamlkubectl create -f jobmanager-rest-service.yamlkubectl create -f jobmanager-job.yamlkubectl create -f taskmanager-job-deployment.yaml

Navtive Integration 的技術細節

為什么叫 Native 方式？包括如下幾個含義。

資源申請方式：Flink 的 Client 內置了一個 K8s Client，可以借助 K8s Client 去創建 JobManager，當 Job 提交之后，如果對資源有需求，JobManager 會向 Flink 自己的 ResourceManager 去申請資源。這個時候 Flink 的 ResourceManager 會直接跟 K8s 的 API Server 通信，將這些請求資源直接下發給 K8s Cluster，告訴它需要多少個 TaskManger，每個 TaskManager 多大。當任務運行完之后，它也會告訴 K8s Cluster釋放沒有使用的資源。相當于 Flink 用很原生的方式了解到 K8s Cluster 的存在，并知曉何時申請資源，何時釋放資源。Native 是相對于 Flink 而言的，借助 Flink 的命令就可以達到自治的一個狀態，不需要引入外部工具就可以通過 Flink 完成任務在 K8s 上的運行。

具體如何工作？主要分 Session 和 Perjob 兩個方面來給大家介紹。

Native Kubernetes Session 方式

首先 Session 的方式。

第一個階段：啟動 Session Cluster。Flink Client 內置了 K8s Client，告訴 K8s Master 創建 Flink Master Deployment，ConfigMap，SVC。創建完成后，Master 就拉起來了。這時，Session 就部署完成了，并沒有維護任何 TaskManager。第二個階段：當用戶提交 Job 時，可以通過 Flink Client 或者 Dashboard 的方式，然后通過 Service 到 Dispatcher，Dispatcher 會產生一個 JobMaster。JobMaster 會向 K8sResourceManager 申請資源。ResourceManager 會發現現在沒有任何可用的資源，它就會繼續向 K8s 的 Master 去請求資源，請求資源之后將其發送回去，起新的 Taskmanager。Taskmanager 起來之后，再注冊回來，此時的 ResourceManager 再向它去申請 slot 提供給 JobMaster，最后由 JobMaster 將相應的 Task 部署到 TaskManager 上。這樣整個從 Session 的拉起到用戶提交都完成了。需注意的是，圖中 SVC 是一個 External Service。必須要保證 Client 通過 Service 可以訪問到 Master。在很多 K8s 集群里，K8s 和 Flink Client 是不在同一個網絡環境的，這時候可以通過 LoadBalancer 的方式或者 NodePort 的方式，使 Flink Client 可以訪問到 Jobmanager Dispatcher，否則 Jar 包是無法提交的。

Session方式代碼

// 啟動session集群，可以指定clusterId，image地址，還有一些CPU，內存的設定./bin/kubernetes-session.sh \-Dkubernetes.cluster-id=k8s-session-1 \-Dkubernetes.container.image=flink-on-kubernetes-job:1.0.2 \-Dkubernetes.container.image.pull-policy=Always \-Djobmanager.heap.size=4096m \-Dtaskmanager.memory.process.size=4096m \-Dtaskmanager.numberOfTaskSlots=4 \-Dkubernetes.jobmanager.cpu=1 -Dkubernetes.taskmanager.cpu=2// 提交任務到session集群，需要指定clusterId，而且session集群的service必須暴露為8081端口，應該是flink客戶端默認值就是提交到8081端口./bin/flink run \    --target kubernetes-session \    -Dkubernetes.cluster-id=flink-session-first-cluster-v1 \    ./examples/streaming/WordCount.jar

Native Kubernetes Perjob 方式

我們再來看一下 Perjob 的方式，如圖所示，Perjob 方式其實和之前是有一些類似，差別在于不需要先去起一個 Session Cluster，再提交任務，而是一步的。

首先創建出了 Service、Master 和 ConfigMap 這幾個資源以后，Flink Master Deployment 里面已經帶了一個用戶 Jar，這個時候 entrypoint 就會從用戶 Jar 里面去提取出或者運行用戶的 main，然后產生 JobGraph。之后再提交到 Dispatcher，由 Dispatcher 去產生 Master，然后再向 ResourceManager 申請資源，后面的邏輯的就和 Session 的方式是一樣的。它和 Session 最大的差異就在于它是一步提交的。因為沒有了兩步提交的需求，如果不需要在任務起來以后訪問外部 UI，就可以不用外部的 Service?？芍苯油ㄟ^一步提交使任務運行。通過本地的 port-forward 或者是用 K8s ApiServer 的一些 proxy 可以訪問 Flink 的 Web UI。此時，External Service 就不需要了，意味著不需要再占用一個 LoadBalancer 或者占用 NodePort。這就是 perjob 方式。

Application模式提交任務

// 不需要提前啟動集群，直接提交任務創建集群執行任務./bin/flink run-application -p 10 -t kubernetes-application \-Dkubernetes.cluster-id=k8s-app1 \-Dkubernetes.container.image=flink-on-kubernetes-job:1.0.2 \-Dkubernetes.container.image.pull-policy=Always \-Djobmanager.heap.size=4096m -Dtaskmanager.memory.process.size=4096m \-Dkubernetes.jobmanager.cpu=1 -Dkubernetes.taskmanager.cpu=2 \-Dtaskmanager.numberOfTaskSlots=4 \local:///opt/flink/examples/streaming/WindowJoin.jar

Session 與 Perjob 方式的不同

我們來看一下 Session 和 Perjob 方式有哪些不同？

flink基于K8s云原生的方式部署方案詳情

背景：目前大多數服務都基于k8s去一鍵部署，可以解決環境帶來的問題并大大提高部署效率，更優的方案是基于云原生的方式去部署，解決動態擴縮容問題，提高資源利用率。所以大數據服務也需要能基于k8s云原生的方式去部署。

調研：目前比較常見的解決方案都是基于k8s上面部署yarn，然后在yarn里面啟動flink集群。這個方案解決了k8s部署問題，但是沒辦法解決資源利用率問題，任務啟動的時候必須指定資源數量，資源少了不夠用，資源多了浪費，沒法實現動態擴縮容。

實現方案：直接基于k8s的云原生方案去實現，去除yarn層，而且可以基于API的方式啟動任務，還可以動態配置容器資源，目前可以設置CPU和內存參數。但是還有一個比較棘手的問題需要解決：APP方式提交任務，需要提前把任務代碼的jar包打到鏡像里面，啟動任務的時候指定jar包路徑和名稱，而且需要一個任務一個jar包，N個任務N個jar包。這種方式比較麻煩，而且沒法動態實現任務的啟動。

方案一：網上找了一下方案，都是說任務啟動的時候動態去下載需要的jar包，這樣也需要提前把一個任務打成jar包，放到可以下載的服務上，還是不夠靈活。

方案二：翻看源碼，發現flink1.11到1.12版本支持一個特殊參數：kubernetes.container-start-command-template，defaultValue："%java% %classpath% %jvmmem% %jvmopts% %logging% %class% %args% %redirects%"，參數說明："Template for the kubernetes jobmanager and taskmanager container start invocation."，通過參數說明可以發現，這個參數可以配置k8s啟動容器時執行jar服務的命令。其中包括classpath設置、jvm相關的參數設置、日志配置，啟動類class設置、main函數的args參數設置等等?；谶@個發現，大膽做了一個設想方案，開發一個jar服務，獲取java服務啟動jvmopts里面或者args里面的參數，兩種方式都可以，然后根據參數去數據庫讀取任務信息，根據獲取到的信息執行任務。

最終采取了方案二實現，方案一不符合整體FlinkSP架構的易用性這一點，方案二更符合我們整體架構的思路，通過任務管理平臺去創建任務，任務數據保存到MySQL數據庫，然后Flink任務解析服務通過任務名稱去獲取任務詳情，并提交任務到Flink環境執行任務。