基于云原生的Flink計算平臺實踐
?Labs 導讀
隨著云原生的普及，越來越多的后端應用進行了容器化遷移，并通過k8s進行編排管理。而最近這幾年，大數據領域比如Flink，Spark等計算引擎也紛紛表示對k8s的支持，使得大數據應用從傳統的yarn時代轉變為云原生時代。本文以Flink和k8s為主要技術手段，介紹如何搭建一個云原生計算平臺。
Part 01 K8s概述
Kubernetes，又稱為k8s或者kube，是一種可自動實施Linux容器操作的開源平臺。它可以幫助用戶省去應用容器化過程的許多手動部署和擴展操作。也就是說，您可以將運行Linux容器的多組主機聚合在一起。由k8s幫助您輕松高效地管理這些集群。而且，這些集群可跨公共云、私有云或混合云部署主機。其架構圖如下：
注：
- kubectl: 使用Kubernetees API與Kubernetes集群的控制面通信的命令行工具。
- ETCD: 一種分布式健值存儲，用于分布式系統或計算機集群的??配置、服務發現和調度協調。
- Node(節點): 負責執行請求和所分配任務的計算機。
- Pod(容器集): 被部署在單個節點上的，且包含一個或多個容器的容器組。
- Kubelet: 運行在節點上的服務，可讀取容器清單，確保指定的容器啟動并運行。
1.在Master上，是由Controller，API Server，Scheduler 以及包括做存儲的Etcd等構成。Etcd可以算成Master，也可以作為獨立于Master之外的存儲來對待。Master的Controller、API Server、Scheduler都是單獨的進程模式。這和Yarn有一些不同，Yarn的整個Master是一個單進程的模式。K8s的Master還可以在多個Master之間完成自發的選舉，然后由active狀態的Master對外提供服務。
2.在Slave上，它主要是包括Kube proxy、Kubelet，以及Docker等相關的組件，每個Node上部署的相關組件都是類似的，通過它來管理上面運行的多個Pod。
3.根據不同用戶的習慣，可以通過UI或者CLI的方式向K8s提交任務。用戶可以通過K8s提供的Dashboard Web UI的方式將任務進行提交，也可以通過Kubectl命令行的方式進行提交。
Part 02 Flink概述
Apache Flink 是一個框架和分布式處理引擎，用于在無邊界和有邊界數據流上進行有狀態的計算。Flink 能在所有常見集群環境中運行，并能以內存速度和任意規模進行計算。其架構圖如下：
Flink整個系統包含三個部分：
Client：Flink Client主要給用戶提供向Flink系統提交用戶任務（流式作業）的能力。
TaskManager：Flink系統的業務執行節點，執行具體的用戶任務。TaskManager可以有多個，各個TaskManager都平等。
JobManager：Flink系統的管理節點，管理所有的TaskManager，并決策用戶任務在哪些Taskmanager執行。JobManager在HA模式下可以有多個，但只有一個主JobManager。
Part 03 Flink on K8s部署模式
Flink的部署方式有很多，支持Local，Standalone，Yarn，Docker，Kubernetes模式等。而根據Flink job的提交模式，又可以分為三種模式：
模式1：Application Mode
Flink提交的程序，被當做集群內部Application，不再需要Client端做繁重的準備工作。（例如執行main函數，生成JobGraph，下載依賴并分發到各個節點等）。main函數被提交給JobManager執行。一個Application一個Cluster實例。
模式2：Per-Job Mode
可以理解為 Client 模式的Application Mode，利用資源管理框架，例如Yarn，Mesos等，資源隔離性更強。一個Job一個Cluster實例。逐漸被廢棄。
模式3：Session Mode
也是由Client提交，做一些預備工作。但是Cluster的實例已經被創建，是所有Job共享的。一個Job導致的JobManager失敗可能會導致所有的Job失敗。
總的來說，在k8s上可以運行以下四種模式：standalone session，standalone application， native session和native application。其特點如下：
根據以上特點，一般生產采用native application模式部署。其啟動流程圖如下：
1.首先創建出了 Service、Master 和 ConfigMap 這幾個資源以后，Flink Master Deployment 里面已經帶了一個用戶 Jar，這個時候 Cluster Entrypoint 就會從用戶 Jar 里面去提取出或者運行用戶的 main，然后產生 JobGraph。之后再提交到 Dispatcher，由 Dispatcher 去產生 Master，然后再向 ResourceManager 申請資源，后面的邏輯的就和 Session 的方式是一樣的。
2.它和 Session 最大的差異就在于它是一步提交的。因為沒有了兩步提交的需求，如果不需要在任務起來以后訪問外部 UI，就可以不用外部的 Service?？芍苯油ㄟ^一步提交使任務運行。通過本地的 port-forward 或者是用 K8s ApiServer 的一些 proxy 可以訪問 Flink 的 Web UI。此時，External Service 就不需要了，意味著不需要再占用一個 LoadBalancer 或者占用 NodePort。
Part 04 Flink on native k8s部署實戰
部署Flink之前需要一個正在運行的k8s集群，且滿足以下幾點需求：
- Kubernetes >= 1.9。
- KubeConfig，它可以列出、創建、刪除 Pod 和服務，可通過 ~/.kube/config 進行配置。您可以通過運行 kubectl auth can-i <list|create|edit|delete> pods 來驗證權限。
- 啟用 Kubernetes DNS。
- 具有創建、刪除 Pod 的 RBAC 權限的默認服務帳戶。
在有了以上的k8s集群后，接下去我們需要執行以下步驟來啟動一個k8s native application任務：
1.創建一個名字為flink-cluster的namespace
kubectl create namespace flink-cluster
2.創建一個賬戶
Kubectl create serviceaccount flink -n flink-cluster
3.Service account和角色綁定
kubectl create clusterrolebinding flink-role-binding-flink \
--clusterrole=edit \
--serviceaccount=flink-cluster:flink
4.編寫Dockerfile文件
# base image
FROM apache/flink:1.14.5-scala_2.11
RUN mkdir -p $FLINK_HOME/usrlib
RUN mkdir -p $FLINK_HOME/hadoopconf
COPY flink-1.14.5/examples/streaming/TopSpeedWindowing.jar $FLINK_HOME/usrlib/TopSpeedWindowing.jar
COPY core-site.xml $FLINK_HOME/hadoopconf
COPY hdfs-site.xml $FLINK_HOME/hadoopconf
# 添加hdfs的相關的jar，為了讀取hdfs
COPY flink-shaded-hadoop2-uber-2.8.3-1.8.3.jar $FLINK_HOME/lib
# 配置HADOOP_CONF_DIR為了獲取hadoop的core-site.xml and hdfs-site.xml ，因為checkpoint是存在hdfs的，需要讀寫hdfs
ENV HADOOP_CONF_DIR=$FLINK_HOME/hadoopconf:$HADOOP_CONF_DIR
5.打鏡像
docker build -t apache/flink:v0.1 .
打鏡像的目錄下存在的文件
core-site.xml
Dockerfile
flink-1.14.5
flink-shaded-hadoop2-uber-2.8.3-1.8.3.jar
hdfs-site.xml
6.啟動application mode的任務
./flink-1.14.5/bin/flink run-application \
--target kubernetes-application \
-Dkubernetes.namespace=flink-cluster \
-Dkubernetes.jobmanager.service-account=flink \
-Dkubernetes.cluster-id=my-first-application-cluster \
-Dkubernetes.high-availability=org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory \
-Dkubernetes.cluster-id=my-first-application-cluster \
-Dhigh-availability.storageDir=hdfs://${your hdfs cluster name}/flink/recovery \
-Dkubernetes.container.image=apache/flink:v0.1 \
-Dkubernetes.rest-service.exposed.type=NodePort \
-Dstate.backend=rocksdb \
-Dstate.checkpoints.dir=hdfs://${your hdfs cluster name}/flink/flink-checkpoints \
-Dstate.backend.incremental=true \
local:///opt/flink/usrlib/TopSpeedWindowing.jar
啟動之后會在最后出現如下的日志，最后的地址就是該任務的web ui地址
2022-07-27 16:45:00,320 INFO org.apache.flink.kubernetes.KubernetesClusterDescriptor [] - Create flink application cluster my-first-application-cluster successfully, JobManager Web Interface: http://ip:port
啟動之后我們會看到k8s啟動了以下的flink的組件?
[root@master1 ~]# kubectl get svc -n flink-cluster
NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE
my-first-application-cluster ClusterIP None <none> 6123/TCP,6124/TCP 1h
my-first-application-cluster-rest ClusterIP ip xxxx <none> 8082/TCP 46h
[root@master1 ~]# kubectl get deployments -n flink-cluster
NAME READY UP-TO-DATE AVAILABLE AGE
my-first-application-cluster 1/1 1 1 1h
[root@master1 ~]# kubectl get pods -n flink-cluster
NAME READY STATUS RESTARTS AGE
my-first-application-cluster-7c4d9d7994-6vwjr 1/1 Running 0 1h
my-first-application-cluster-taskmanager-1-1 1/1 Running 0 461h
這樣，就表示Flink啟動成功了。我們看到在k8s上有2個pod，分別是jobmanger和taskmanger，一個service用于端口映射。
Part 05 Flink平臺k8s改造實踐
在k8s改造之前，我們已經有了一個Flink計算平臺。其主要包含3個部分，web ui用于用戶提交SQL任務或者JAR任務，web服務端用于解析SQL或者JAR，然后生成Flink Client提交命令，最后執行命令提交任務到集群上去。
如上圖所示，我們的平臺已經有Yarn和Standalone模式提交。那么支持云原生，只需要在原先基礎上增加一個native k8s提交方式即可。在代碼層面就是在提交任務抽象類上增加一個native k8s的實現。
下圖為具體的一個native k8s SQL任務提交頁面。只需要設置運行模式、運行參數和依賴的第三方jar包，再加上具體的SQL腳本即可完成一個Flink任務的提交。
文獻來源
[1]Flink官網,《Flink Documentation》
[2]Kubernetes官網,《kubernetes Documentation》、
[3]騰訊云開發者社區，《原生的在K8s上運行Flink》
[4]騰訊云開發者社區，《Flink 1.13 在Native k8s的部署實踐》
[5]51CTO,《網易游戲 Flink SQL 平臺化實踐》?
