flink on native kubernetes

上一章講了flink 的 application mode。今天咱們主要經過該模式提交WordCount做業，而且選擇的是native kubernetes。

Native Kubernetes 原理

下圖描述了flink如何與kubernetes集成：

建立Flink Kubernetes Session集羣時，Flink客戶端將首先鏈接到Kubernetes ApiServer提交集羣描述，包括ConfigMap規範，Job Manager服務規範，Job Manager Deployment規範和Owner Reference。而後，Kubernetes將建立JobManager Deployment，在此期間，Kubelet將拉取鏡像，準備並安裝卷，而後執行啓動命令。啓動JobManager Pod 後，Dispatcher和KubernetesResourceManager 就緒可用，而且集羣已準備好接受一個或多個做業。

當用戶經過Flink客戶端提交做業時，客戶端將生成Job Graph，並將其與用戶jar一塊兒上傳到Dispatcher。

JobManager向KubernetesResourceManager請求插槽的資源。若是沒有可用的插槽，資源管理器將建立TaskManager Pod並在集羣中註冊它們。

示例

Application mode容許用戶建立一個包含其Job和Flink運行時的鏡像，這將根據須要自動建立和銷燬集羣組件。 Flink社區提供了針對任何用例定製的基本docker鏡像。

下載

首先去官網下載 flink1.11。包中包含如下內容：

bin  conf  Dockerfile  examples  lib  LICENSE  licenses  log  NOTICE  opt  plugins  README.txt

其中：

• bin 下包含了flink相關的一些可執行文件以及一些shell腳本，主要用來提交任務或是建立中止集羣
• conf 包含了flink和日誌等相關的配置文件
• examples， 顧名思義，就是一些demo示例，好比咱們今天的WordCount 就位於此路徑下
• lib 包含 flink 依賴的包
• opt 包含了一些擴展的包，好比flink 對接s3的專用包
• plugins 包含了監控相關，好比對接prometheus，graphite等。

構建鏡像

而後咱們建立一個Dockerfile，用於定製鏡像。Dockerfile內容以下：

FROM flink
RUN mkdir -p $FLINK_HOME/usrlib
COPY ./examples/streaming/WordCount.jar $FLINK_HOME/usrlib/my-flink-job.jar

構建鏡像：

docker build -t iyacontrol/flink-world-count:v0.0.1 .
Sending build context to Docker daemon  362.7MB
Step 1/3 : FROM flink
latest: Pulling from library/flink
e9afc4f90ab0: Already exists 
989e6b19a265: Already exists 
af14b6c2f878: Already exists 
68a79816c3e1: Pull complete 
037cc5cb1b83: Pull complete 
d3efdb331614: Pull complete 
bf82d2b871ad: Pull complete 
4ff2e8c5d83f: Pull complete 
f15a0d59303a: Pull complete 
81130e2e9fdd: Pull complete 
40bdeebc27c6: Pull complete 
8fe3a85e5402: Pull complete 
Digest: sha256:665db47d0a2bcc297e9eb4df7640d3e4c1d398d25849252a726c8ada112722cf
Status: Downloaded newer image for flink:latest
 ---> 43f070a908e6
Step 2/3 : RUN mkdir -p $FLINK_HOME/usrlib
 ---> Running in c44a726b85a9
Removing intermediate container c44a726b85a9
 ---> 67ab6686e049
Step 3/3 : COPY ./examples/streaming/WordCount.jar $FLINK_HOME/usrlib/my-flink-job.jar
 ---> ab3686ebc7e5
Successfully built ab3686ebc7e5

推送鏡像到鏡像倉庫：

docker push  iyacontrol/flink-world-count:v0.0.1
The push refers to repository [docker.io/iyacontrol/flink-world-count]
b3b3d0402b8d: Pushed 
b1757ffb6e42: Pushed 
3af0e2838f53: Mounted from library/flink 
cf0f92755ad7: Mounted from library/flink 
1f8a2f4bd423: Mounted from library/flink 
eedc301c6f3f: Mounted from library/flink 
d23c0e026b3e: Mounted from library/flink 
37f26e989a45: Mounted from library/flink 
e658c78cae16: Mounted from library/flink 
d8859f270d7a: Mounted from library/flink 
7ab97ad88178: Mounted from library/flink 
527ade4639e0: Mounted from library/flink 
c2c789d2d3c5: Mounted from library/flink 
8803ef42039d: Mounted from library/flink 
v0.0.1: digest: sha256:fcd99fedbba2734796226a725789bf7db109131b04f2a13c1cd1bc773ff3b8c0 size: 3253

注意：此處須要換成本身的鏡像倉庫。或是能夠繞過構建步驟，直接使用我打好的鏡像。

配置kubernetes RBAC權限

須要給flink授予RBAC某些權限，而且在提交任務的時候經過參數(-Dkubernetes.jobmanager.service-account=flink)指定， JobManager 方可建立做業Pod。

kubectl create serviceaccount flink -n stream

kubectl create clusterrolebinding flink-role-binding-flink -n stream  --clusterrole=edit --serviceaccount=stream:flink

並無選擇default 命名空間，這裏建立了一個stream的命名空間。

提交做業

執行如下命令提交WorldCount 做業:

./bin/flink run-application -p 8 -t kubernetes-application \
  -Dkubernetes.cluster-id=my-first-cluster \
  -Dtaskmanager.memory.process.size=4096m \
  -Dkubernetes.taskmanager.cpu=2 \
  -Dtaskmanager.numberOfTaskSlots=4 \
  -Dkubernetes.container.image=iyacontrol/flink-world-count:v0.0.1 \
  -Dkubernetes.namespace=stream \
  -Dkubernetes.jobmanager.service-account=flink \
  -Dkubernetes.rest-service.exposed.type=LoadBalancer \
  -Dkubernetes.rest-service.annotations=service.beta.kubernetes.io/aws-load-balancer-type:nlb \
  local:///opt/flink/usrlib/my-flink-job.jar

只有 application mode 的架構支持local。假定jar位於鏡像中，而不位於Flink客戶端中。

以下相似輸出：

2020-07-17 17:57:55,455 INFO  org.apache.flink.kubernetes.utils.KubernetesUtils            [] - Kubernetes deployment requires a fixed port. Configuration blob.server.port will be set to 6124
2020-07-17 17:57:55,455 INFO  org.apache.flink.kubernetes.utils.KubernetesUtils            [] - Kubernetes deployment requires a fixed port. Configuration taskmanager.rpc.port will be set to 6122
2020-07-17 17:57:55,511 WARN  org.apache.flink.kubernetes.kubeclient.decorators.HadoopConfMountDecorator [] - Found 0 files in directory null/etc/hadoop, skip to mount the Hadoop Configuration ConfigMap.
2020-07-17 17:57:55,511 WARN  org.apache.flink.kubernetes.kubeclient.decorators.HadoopConfMountDecorator [] - Found 0 files in directory null/etc/hadoop, skip to create the Hadoop Configuration ConfigMap.
2020-07-17 17:57:56,348 INFO  org.apache.flink.kubernetes.KubernetesClusterDescriptor      [] - Create flink application cluster my-first-cluster successfully, JobManager Web Interface: http://F1DD312BB1102AC0AE558F66FA.gr7.ap-southeast-1.eks.amazonaws.com:8081

下面咱們講下提交參數：

• kubernetes.cluster-id 能夠指定，更加語義化，若是不指定，系統會使用uuid。
• kubernetes.rest-service.exposed.type 指定暴露Jobmanager 服務的方式。
• kubernetes.rest-service.annotations 若是你的k8s集羣使用的是公有云的託管k8s，並且 kubernetes.rest-service.exposed.type 爲LoadBalancer，該參數通常都須要設置。
• kubernetes.container.image 指定咱們做業的鏡像。

而後咱們查看一下再k8s當中建立了那些資源：

kubectl get all -n stream
NAME                                    READY   STATUS    RESTARTS   AGE
pod/my-first-cluster-64ff98cd96-sprk7   1/1     Running   0          24s
pod/my-first-cluster-taskmanager-1-1    1/1     Running   0          16s
pod/my-first-cluster-taskmanager-1-2    0/1     Pending   0          12s

NAME                            TYPE           CLUSTER-IP       EXTERNAL-IP                                                                          PORT(S)             AGE
service/my-first-cluster        ClusterIP      None             <none>                                                                               6123/TCP,6124/TCP   24s
service/my-first-cluster-rest   LoadBalancer   10.100.xx.179   a4fd46cfa3985f99582310dbfd-0ce036fe28648b82.elb.ap-southeast-1.amazonaws.com   8081:32756/TCP      24s

NAME                               READY   UP-TO-DATE   AVAILABLE   AGE
deployment.apps/my-first-cluster   1/1     1            1           24s

NAME                                          DESIRED   CURRENT   READY   AGE
replicaset.apps/my-first-cluster-64ff98cd96   1         1         1       24s

• 名稱爲my-first-cluster 的Deployment ，即JobManager
• JobManager根據提交參數建立的兩個TaskManager Pod
• 名稱爲my-first-cluster-rest的 服務，用於集羣外訪問JobManager的管理UI。
• 名稱爲my-first-cluster的服務，該服務爲Headless 服務，用於TaskManager 的Pod 訪問JobManager。

訪問UI

使用上面的url訪問JobManager的UI，以下：

總結

flink native kubernets 在1.10的時候推出，目前還處於開發當中，某些參數可能會在以後版本中變化。不過1.11 版本已經比較穩定了。