上一章讲了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 flinkRUN mkdir -p $FLINK_HOME/usrlibCOPY ./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.7MBStep 1/3 : FROM flinklatest: Pulling from library/flinke9afc4f90ab0: 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:665db47d0a2bcc297e9eb4df7640d3e4c1d398d25849252a726c8ada112722cfStatus: Downloaded newer image for flink:latest ---> 43f070a908e6Step 2/3 : RUN mkdir -p $FLINK_HOME/usrlib ---> Running in c44a726b85a9Removing intermediate container c44a726b85a9 ---> 67ab6686e049Step 3/3 : COPY ./examples/streaming/WordCount.jar $FLINK_HOME/usrlib/my-flink-job.jar ---> ab3686ebc7e5Successfully built ab3686ebc7e5

推送镜像到镜像仓库：

docker push  iyacontrol/flink-world-count:v0.0.1The 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 streamkubectl 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 61242020-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 61222020-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 streamNAME                                    READY   STATUS    RESTARTS   AGEpod/my-first-cluster-64ff98cd96-sprk7   1/1     Running   0          24spod/my-first-cluster-taskmanager-1-1    1/1     Running   0          16spod/my-first-cluster-taskmanager-1-2    0/1     Pending   0          12sNAME                            TYPE           CLUSTER-IP       EXTERNAL-IP                                                                          PORT(S)             AGEservice/my-first-cluster        ClusterIP      None             <none>                                                                               6123/TCP,6124/TCP   24sservice/my-first-cluster-rest   LoadBalancer   10.100.xx.179   a4fd46cfa3985f99582310dbfd-0ce036fe28648b82.elb.ap-southeast-1.amazonaws.com   8081:32756/TCP      24sNAME                               READY   UP-TO-DATE   AVAILABLE   AGEdeployment.apps/my-first-cluster   1/1     1            1           24sNAME                                          DESIRED   CURRENT   READY   AGEreplicaset.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 版本曾经比较稳定了。