?Labs 導(dǎo)讀

隨著云原生的普及，越來(lái)越多的后端應(yīng)用進(jìn)行了容器化遷移，并通過(guò)k8s進(jìn)行編排管理。而最近這幾年，大數(shù)據(jù)領(lǐng)域比如Flink，Spark等計(jì)算引擎也紛紛表示對(duì)k8s的支持，使得大數(shù)據(jù)應(yīng)用從傳統(tǒng)的yarn時(shí)代轉(zhuǎn)變?yōu)樵圃鷷r(shí)代。本文以Flink和k8s為主要技術(shù)手段，介紹如何搭建一個(gè)云原生計(jì)算平臺(tái)。

Part 01  K8s概述 

Kubernetes，又稱為k8s或者kube，是一種可自動(dòng)實(shí)施Linux容器操作的開源平臺(tái)。它可以幫助用戶省去應(yīng)用容器化過(guò)程的許多手動(dòng)部署和擴(kuò)展操作。也就是說(shuō)，您可以將運(yùn)行Linux容器的多組主機(jī)聚合在一起。由k8s幫助您輕松高效地管理這些集群。而且，這些集群可跨公共云、私有云或混合云部署主機(jī)。其架構(gòu)圖如下：

注：

1. kubectl: 使用Kubernetees API與Kubernetes集群的控制面通信的命令行工具。
2. ETCD: 一種分布式健值存儲(chǔ)，用于分布式系統(tǒng)或計(jì)算機(jī)集群的??配置、服務(wù)發(fā)現(xiàn)和調(diào)度協(xié)調(diào)。
3. Node(節(jié)點(diǎn)): 負(fù)責(zé)執(zhí)行請(qǐng)求和所分配任務(wù)的計(jì)算機(jī)。
4. Pod(容器集): 被部署在單個(gè)節(jié)點(diǎn)上的，且包含一個(gè)或多個(gè)容器的容器組。
5. Kubelet: 運(yùn)行在節(jié)點(diǎn)上的服務(wù)，可讀取容器清單，確保指定的容器啟動(dòng)并運(yùn)行。

1.在Master上，是由Controller，API Server，Scheduler 以及包括做存儲(chǔ)的Etcd等構(gòu)成。Etcd可以算成Master，也可以作為獨(dú)立于Master之外的存儲(chǔ)來(lái)對(duì)待。Master的Controller、API Server、Scheduler都是單獨(dú)的進(jìn)程模式。這和Yarn有一些不同，Yarn的整個(gè)Master是一個(gè)單進(jìn)程的模式。K8s的Master還可以在多個(gè)Master之間完成自發(fā)的選舉，然后由active狀態(tài)的Master對(duì)外提供服務(wù)。

2.在Slave上，它主要是包括Kube proxy、Kubelet，以及Docker等相關(guān)的組件，每個(gè)Node上部署的相關(guān)組件都是類似的，通過(guò)它來(lái)管理上面運(yùn)行的多個(gè)Pod。

3.根據(jù)不同用戶的習(xí)慣，可以通過(guò)UI或者CLI的方式向K8s提交任務(wù)。用戶可以通過(guò)K8s提供的Dashboard Web UI的方式將任務(wù)進(jìn)行提交，也可以通過(guò)Kubectl命令行的方式進(jìn)行提交。

Part 02  Flink概述

Apache Flink 是一個(gè)框架和分布式處理引擎，用于在無(wú)邊界和有邊界數(shù)據(jù)流上進(jìn)行有狀態(tài)的計(jì)算。Flink 能在所有常見集群環(huán)境中運(yùn)行，并能以內(nèi)存速度和任意規(guī)模進(jìn)行計(jì)算。其架構(gòu)圖如下：

Flink整個(gè)系統(tǒng)包含三個(gè)部分：

Client：Flink Client主要給用戶提供向Flink系統(tǒng)提交用戶任務(wù)（流式作業(yè)）的能力。

TaskManager：Flink系統(tǒng)的業(yè)務(wù)執(zhí)行節(jié)點(diǎn)，執(zhí)行具體的用戶任務(wù)。TaskManager可以有多個(gè)，各個(gè)TaskManager都平等。

JobManager：Flink系統(tǒng)的管理節(jié)點(diǎn)，管理所有的TaskManager，并決策用戶任務(wù)在哪些Taskmanager執(zhí)行。JobManager在HA模式下可以有多個(gè)，但只有一個(gè)主JobManager。

Part 03  Flink on K8s部署模式 

Flink的部署方式有很多，支持Local，Standalone，Yarn，Docker，Kubernetes模式等。而根據(jù)Flink job的提交模式，又可以分為三種模式：

模式1：Application Mode

Flink提交的程序，被當(dāng)做集群內(nèi)部Application，不再需要Client端做繁重的準(zhǔn)備工作。（例如執(zhí)行main函數(shù)，生成JobGraph，下載依賴并分發(fā)到各個(gè)節(jié)點(diǎn)等）。main函數(shù)被提交給JobManager執(zhí)行。一個(gè)Application一個(gè)Cluster實(shí)例。

模式2：Per-Job Mode

可以理解為 Client 模式的Application Mode，利用資源管理框架，例如Yarn，Mesos等，資源隔離性更強(qiáng)。一個(gè)Job一個(gè)Cluster實(shí)例。逐漸被廢棄。

模式3：Session Mode

也是由Client提交，做一些預(yù)備工作。但是Cluster的實(shí)例已經(jīng)被創(chuàng)建，是所有Job共享的。一個(gè)Job導(dǎo)致的JobManager失敗可能會(huì)導(dǎo)致所有的Job失敗。

總的來(lái)說(shuō)，在k8s上可以運(yùn)行以下四種模式：standalone session，standalone application， native session和native application。其特點(diǎn)如下：

根據(jù)以上特點(diǎn)，一般生產(chǎn)采用native application模式部署。其啟動(dòng)流程圖如下：

1.首先創(chuàng)建出了 Service、Master 和 ConfigMap 這幾個(gè)資源以后，Flink Master Deployment 里面已經(jīng)帶了一個(gè)用戶 Jar，這個(gè)時(shí)候 Cluster Entrypoint 就會(huì)從用戶 Jar 里面去提取出或者運(yùn)行用戶的 main，然后產(chǎn)生 JobGraph。之后再提交到 Dispatcher，由 Dispatcher 去產(chǎn)生 Master，然后再向 ResourceManager 申請(qǐng)資源，后面的邏輯的就和 Session 的方式是一樣的。

2.它和 Session 最大的差異就在于它是一步提交的。因?yàn)闆]有了兩步提交的需求，如果不需要在任務(wù)起來(lái)以后訪問外部 UI，就可以不用外部的 Service?？芍苯油ㄟ^(guò)一步提交使任務(wù)運(yùn)行。通過(guò)本地的 port-forward 或者是用 K8s ApiServer 的一些 proxy 可以訪問 Flink 的 Web UI。此時(shí)，External Service 就不需要了，意味著不需要再占用一個(gè) LoadBalancer 或者占用 NodePort。

Part 04  Flink on native k8s部署實(shí)戰(zhàn)

部署Flink之前需要一個(gè)正在運(yùn)行的k8s集群，且滿足以下幾點(diǎn)需求：

• Kubernetes >= 1.9。
• KubeConfig，它可以列出、創(chuàng)建、刪除 Pod 和服務(wù)，可通過(guò) ~/.kube/config 進(jìn)行配置。您可以通過(guò)運(yùn)行 kubectl auth can-i <list|create|edit|delete> pods 來(lái)驗(yàn)證權(quán)限。
• 啟用 Kubernetes DNS。
• 具有創(chuàng)建、刪除 Pod 的 RBAC 權(quán)限的默認(rèn)服務(wù)帳戶。

在有了以上的k8s集群后，接下去我們需要執(zhí)行以下步驟來(lái)啟動(dòng)一個(gè)k8s native application任務(wù)：

1.創(chuàng)建一個(gè)名字為flink-cluster的namespace

kubectl create namespace flink-cluster

2.創(chuàng)建一個(gè)賬戶

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的相關(guān)的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 ，因?yàn)閏heckpoint是存在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.啟動(dòng)application mode的任務(wù)

./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

啟動(dòng)之后會(huì)在最后出現(xiàn)如下的日志，最后的地址就是該任務(wù)的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

啟動(dòng)之后我們會(huì)看到k8s啟動(dòng)了以下的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啟動(dòng)成功了。我們看到在k8s上有2個(gè)pod，分別是jobmanger和taskmanger，一個(gè)service用于端口映射。

Part 05  Flink平臺(tái)k8s改造實(shí)踐

在k8s改造之前，我們已經(jīng)有了一個(gè)Flink計(jì)算平臺(tái)。其主要包含3個(gè)部分，web ui用于用戶提交SQL任務(wù)或者JAR任務(wù)，web服務(wù)端用于解析SQL或者JAR，然后生成Flink Client提交命令，最后執(zhí)行命令提交任務(wù)到集群上去。

如上圖所示，我們的平臺(tái)已經(jīng)有Yarn和Standalone模式提交。那么支持云原生，只需要在原先基礎(chǔ)上增加一個(gè)native k8s提交方式即可。在代碼層面就是在提交任務(wù)抽象類上增加一個(gè)native k8s的實(shí)現(xiàn)。

下圖為具體的一個(gè)native k8s SQL任務(wù)提交頁(yè)面。只需要設(shè)置運(yùn)行模式、運(yùn)行參數(shù)和依賴的第三方j(luò)ar包，再加上具體的SQL腳本即可完成一個(gè)Flink任務(wù)的提交。

文獻(xiàn)來(lái)源

[1]Flink官網(wǎng),《Flink Documentation》

[2]Kubernetes官網(wǎng),《kubernetes Documentation》、

[3]騰訊云開發(fā)者社區(qū)，《原生的在K8s上運(yùn)行Flink》

[4]騰訊云開發(fā)者社區(qū)，《Flink 1.13 在Native k8s的部署實(shí)踐》

[5]51CTO,《網(wǎng)易游戲 Flink SQL 平臺(tái)化實(shí)踐》?