全球今亮點(diǎn)！Application模式和Session模式有什么區(qū)別？

總結(jié)下來就幾點(diǎn)：

1、Native模式比Standalone模式好

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

(資料圖)

Native模式不需要預(yù)先確定需要使用的資源數(shù)量，系統(tǒng)會(huì)實(shí)時(shí)根據(jù)任務(wù)需要自動(dòng)去k8s集群申請能申請到的資源。

2、Application和Session模式各有優(yōu)劣，不同情況使用不同模式

Application模式資源隔離性強(qiáng)，每個(gè)人物都是單獨(dú)的集群，不會(huì)出現(xiàn)并發(fā)問題。每個(gè)任務(wù)都需要啟動(dòng)一個(gè)集群，會(huì)先啟動(dòng)JobManager，然后啟動(dòng)TaskManager，效率會(huì)比較低。適合流處理任務(wù)

對比yarn環(huán)境下的perjob提交任務(wù)速度快很多，大約是十幾秒就能提交執(zhí)行；yarn環(huán)境下提交任務(wù)需要一分多鐘。

Session模式需要提前創(chuàng)建好集群，所有任務(wù)共享集群資源，并發(fā)下可能會(huì)有問題。共用集群，只需要啟動(dòng)TaskManager，效率高。適合批處理任務(wù)

operator模式的利弊

還有一種方式叫operator模式，這種方式的優(yōu)點(diǎn)是有一個(gè)開源服務(wù)，這個(gè)服務(wù)來幫你管理yml配置文件，你不需要自己去管理各種資源的配置。但是需要單獨(dú)啟動(dòng)這個(gè)服務(wù)，然后調(diào)用這個(gè)服務(wù)的api去管理yml文件的配置功能。

優(yōu)點(diǎn)：

管理 Flink 集群更加便捷

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

聲明式

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

自定義保存點(diǎn)

用戶可以指定 autoSavePointSeconds 和保存路徑，F(xiàn)link operator 會(huì)自動(dòng)為用戶定期保存快照。

自動(dòng)恢復(fù)

流式任務(wù)往往是長期運(yùn)行的，甚至 2-3 年不停止都是常見的。在任務(wù)執(zhí)行的過程中，可能會(huì)有各種各樣的原因?qū)е氯蝿?wù)失敗。用戶可以指定任務(wù)重啟策略，當(dāng)指定為 FromSavePointOnFailure，F(xiàn)link operator 自動(dòng)從最近的保存點(diǎn)重新執(zhí)行任務(wù)。

Ingress 集成

用戶可以定義 Ingress 資源，flink operator 將會(huì)自動(dòng)創(chuàng)建 Ingress 資源。云廠商托管的 Kubernetes 集群一般都有 Ingress 控制器，否則需要用戶自行實(shí)現(xiàn) Ingress controller。

Prometheus 集成

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

缺點(diǎn)：

需要單獨(dú)啟動(dòng)一個(gè)服務(wù)

它的很多優(yōu)點(diǎn)基于api的方式也能實(shí)現(xiàn)

3、啟動(dòng)方式

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

Native模式：

通過flink客戶端去創(chuàng)建集群

也可以使用api的方式去創(chuàng)建

Flink On Kubernetes 的部署演進(jìn)

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

Standalone Session On K8s

Standalone方式在k8s運(yùn)行步驟：

如圖所示：

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

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

//創(chuàng)建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//提交任務(wù)到集群./bin/flink run -m localhost:8081 ./examples/streaming/WordCount.jar

Standalone perjob on K8s

現(xiàn)在我們看一下 Perjob 的部署，因?yàn)?Session Cluster 和 Perjob 分別都有不同的適用場景，一個(gè) Session 里面可以跑多個(gè)任務(wù)，但是每個(gè)任務(wù)之間沒有辦法達(dá)到更好的隔離性。而 Perjob 的方式，每個(gè)job都會(huì)有一個(gè)自己獨(dú)立的 Flink Cluster 去運(yùn)行，它們之間相互獨(dú)立。

■ Perjob 的特點(diǎn):

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

■ 執(zhí)行步驟：

由 Standalone JobCluster EntryPoint 執(zhí)行，從 classpath 找到用戶 Jar，執(zhí)行它的 main 方法得到 JobGrapth 。再提交到 Dispathcher，這時(shí)候走 Recover Job 的邏輯，提交到 JobMaster。JobMaster 向 ResourceManager 申請資源，請求 slot，執(zhí)行 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 的技術(shù)細(xì)節(jié)

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

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

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

Native Kubernetes Session 方式

首先 Session 的方式。

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

Session方式代碼

// 啟動(dòng)session集群，可以指定clusterId，image地址，還有一些CPU，內(nèi)存的設(shè)定./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// 提交任務(wù)到session集群，需要指定clusterId，而且session集群的service必須暴露為8081端口，應(yīng)該是flink客戶端默認(rèn)值就是提交到8081端口./bin/flink run \    --target kubernetes-session \    -Dkubernetes.cluster-id=flink-session-first-cluster-v1 \    ./examples/streaming/WordCount.jar

Native Kubernetes Perjob 方式

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

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

Application模式提交任務(wù)

// 不需要提前啟動(dòng)集群，直接提交任務(wù)創(chuàng)建集群執(zhí)行任務(wù)./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云原生的方式部署方案詳情

背景：目前大多數(shù)服務(wù)都基于k8s去一鍵部署，可以解決環(huán)境帶來的問題并大大提高部署效率，更優(yōu)的方案是基于云原生的方式去部署，解決動(dòng)態(tài)擴(kuò)縮容問題，提高資源利用率。所以大數(shù)據(jù)服務(wù)也需要能基于k8s云原生的方式去部署。

調(diào)研：目前比較常見的解決方案都是基于k8s上面部署yarn，然后在yarn里面啟動(dòng)flink集群。這個(gè)方案解決了k8s部署問題，但是沒辦法解決資源利用率問題，任務(wù)啟動(dòng)的時(shí)候必須指定資源數(shù)量，資源少了不夠用，資源多了浪費(fèi)，沒法實(shí)現(xiàn)動(dòng)態(tài)擴(kuò)縮容。

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

方案一：網(wǎng)上找了一下方案，都是說任務(wù)啟動(dòng)的時(shí)候動(dòng)態(tài)去下載需要的jar包，這樣也需要提前把一個(gè)任務(wù)打成jar包，放到可以下載的服務(wù)上，還是不夠靈活。

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

最終采取了方案二實(shí)現(xiàn)，方案一不符合整體FlinkSP架構(gòu)的易用性這一點(diǎn)，方案二更符合我們整體架構(gòu)的思路，通過任務(wù)管理平臺(tái)去創(chuàng)建任務(wù)，任務(wù)數(shù)據(jù)保存到MySQL數(shù)據(jù)庫，然后Flink任務(wù)解析服務(wù)通過任務(wù)名稱去獲取任務(wù)詳情，并提交任務(wù)到Flink環(huán)境執(zhí)行任務(wù)。