Flink是目前最热门的分布式流/批处理框架，而Kubernetes是目前最热门的资源管理和调度平台。Flink反对在Kubernetes上采纳Session模式或Application模式部署作业。基于实践经验，本文次要探讨在Kubernetes上部署Flink作业须要留神的中央。

环境：

  • k8s: 1.15
  • flink-client：flink-1.11.2

测试虽基于flink-1.11.2，参考1.12的文档也不妨：

native_kubernetes

k8s上运行flink工作有两种模式，session模式和application模式（晚期还有一种pre-job模式，但已废除）

session模式下，在k8s上会部署一个运行jobmanager的pod（以及包含deployment/rs/service/configmap等资源）。后续通过提交命令提交的工作，都由这个jobmanager的pod负责向k8s申请taskmanager的pod。这种模式与standalone有些相似，惟一的不同是，standalone模式须要实现部署好master和worker。

application模式下，每个作业会在k8s上部署一套jm和tm，这跟yarn模式是相似的。

筹备

RBCA

基于上述原理，不论是哪种模式都须要pod有权限创立其余资源，因而须要思考RBAC。

正如文档所说，须要当时为default这个servicename设置对应的role，实际中咱们部署如下配置，参考kubernetes-log-user-systemserviceaccountdefaultdefault-cannot-get-services：

apiVersion: rbac.authorization.k8s.io/v1beta1kind: ClusterRoleBindingmetadata:  name: fabric8-rbacsubjects:  - kind: ServiceAccount    # Reference to upper's `metadata.name`    name: default    # Reference to upper's `metadata.namespace`    namespace: defaultroleRef:  kind: ClusterRole  name: cluster-admin  apiGroup: rbac.authorization.k8s.io

最开始遗留上述步骤节约了大量的调试工夫。

在k8s节点上提交

另外，提交工作最好在k8s的节点上进行，因为如下起因

KubeConfig, which has access to list, create, delete pods and services, configurable via ~/.kube/config. You can verify permissions by running kubectl auth can-i <list|create|edit|delete> pods.

CoreDNS

k8s上须要实现装置好CoreDNS

解决日志配置

flink1.11的客户端对log配置解决并不好，这造成调试和排错艰难，所以倡议上来先解决一下客户端的配置：

在flink-conf.yaml减少如下配置：

kubernetes.container-start-command-template: %java% %classpath% %jvmmem% %jvmopts% %logging% %class% %args%kubernetes.container.image.pull-policy: Always
  • kubernetes.container-start-command-template的作用是生成jobmanager pod时的启动命令。这里去掉提交命令中最初的的%redirect%。默认%redirect%会将规范输入和规范谬误重定向到文件，如果pod出错挂掉的话，无奈通过kubectl logs命令查看日志
  • 镜像始终拉取。在内网环境下，流量不是问题，始终拉取镜像，不便前面批改根底镜像后，能及时拉取

将logback-console.xml和log4j-console.properties重命名为logback.xml和log4j.properties。这将使得日志打印到stdout和stderr，否则日志将打印到文件

Session mode

首先通过如下命令提交jobmanager：

$ ./bin/kubernetes-session.sh -Dkubernetes.cluster-id=my-first-flink-cluster

如果不胜利的的话，倡议通过kubectl logs命令看下问题。

接下来提交作业：

$ ./bin/flink run --target kubernetes-session -Dkubernetes.cluster-id=my-first-flink-cluster ./examples/streaming/TopSpeedWindowing.jar

可能会报错：

Caused by: java.util.concurrent.CompletionException: [org.apache.flink.shaded.netty4.io](http://org.apache.flink.shaded.netty4.io/).netty.channel.AbstractChannel$AnnotatedConnectException: 回绝连贯: /192.168.21.73:8081at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292)at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:308)at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:957)at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:940)... 19 more

显然是客户端尝试将jobgraph提交给容器中的jobmanager时无奈连贯到jobmanager。这里想到首先要检查一下jobmanager是否失常，进入jm容器，日志没有什么报错，应该没问题。容器启动的是8081的端口，在里面是无法访问的。这是k8s环境的“通病”：简单网络环境。

有几种计划：

  1. 通过kubectl port-forward进行本地端口转发，例如：

    kubectl port-forward my-first-flink-cluster-6446bbb6f6-4nnm5 8081:8081 --address 0.0.0.0

  2. jobmanager会启动一个rest service资源，默认采纳LoadBalancer类型，咱们能够在集群中装置一个LoadBalancer，上面介绍了如何装置metallb

    metallb自身也是pod运行的，依照官网装置没什么问题：

    kubectl apply -f https://raw.githubusercontent.com/metallb/metallb/v0.9.5/manifests/namespace.yamlkubectl apply -f https://raw.githubusercontent.com/metallb/metallb/v0.9.5/manifests/metallb.yaml

    首次装置需运行

    kubectl create secret generic -n metallb-system memberlist --from-literal=secretkey="$(openssl rand -base64 128)"

    采纳layer2的形式配置，而且地址池的地址段与node是同一个地址段，这样不须要配置其余货色：

    config.yamlapiVersion: v1kind: ConfigMapmetadata:  namespace: metallb-system  name: configdata:  config: |    address-pools:    - name: default      protocol: layer2      addresses:      - 192.168.21.210-192.168.21.215

配置loadbalancer当前：

能够看到xxx-rest的svc，其中的EXTERNAL-IP原先始终是<pending>，当初从地址段中调配了一个地址了。这个地址无奈ping，然而能够拜访：

解决了服务内部拜访的问题，就能失常运行测试作业了。

session模式总结：

  1. session模式下，在k8s上提交一个jobmanager的pod，要保障服务账号有治理pod的权限，否则无奈运行
  2. 客户端提交的时候要留神买通网络，能够用port-forward来做
  3. 客户端提交终端会阻塞，但ctrl-C也没关系
  4. jobmanager会为作业向k8s申请生成taskmanager容器运行作业

Application mode

Session mode尽管看似简略，然而对于扫清环境阻碍起到至关重要的作用。下面提到Application mode与yarn其实是比拟相似的，是一种更靠近生产的部署模式。

首先须要将打包好的应用程序jar包打入镜像：

FROM flink:1.11.2-scala_2.11 RUN mkdir -p $FLINK_HOME/usrlibCOPY jax-flink-entry-2.0-SNAPSHOT.jar $FLINK_HOME/usrlib/jax-flink-entry-2.0-SNAPSHOT.jarCOPY kafka-clients-2.2.0.jar $FLINK_HOME/lib/kafka-clients-2.2.0.jar

以下面的Dockerfile为例，把咱们的利用程序包放到$FLINK_HOME/usrlib（这是个非凡的目录，默认Flink在运行的时候会从这个目录加载用户的jar包）。同时，咱们把依赖包放到$FLINK_HOME/lib下。

构建镜像并推送到外部的镜像仓库：

docker build -t xxxxx:5000/jax-flink:lastest .docker push xxxx:5000/jax-flink:lastest

以Application mode提交作业

./bin/flink run-application \--target kubernetes-application \-Dkubernetes.cluster-id=jax-application-cluster \-Dkubernetes.container.image=xxxx:5000/jax-flink:lastest \-c com.eoi.jax.flink_entry.FlinkMainEntry  \local:///opt/flink/usrlib/jax-flink-entry-2.0-SNAPSHOT.jar ...

作业会启动独立的jobmanager和taskmanager。Applicatoin mode的特点是作业的构建（生成jobgraph的过程）不在客户端实现，而是在jobmanager上实现，这一点与spark的driver是相似的。

一些提交命令参数的作用：

  • 利用本身的参数：会在flink-conf.yaml中生成：$internal.application.program-args。这将最终最为用户main函数的参数[]String
  • -class：会在flink-conf.yaml中生成$internal.application.main
  • -C: 会在flink-conf.yarml中生成pipeline.classpaths（必须是非法的URL）。然而，pipeline.classpaths中的URL不会被加到运行用户main函数的类加载器中，这意味着-C指定的依赖包无奈被用户代码应用。笔者曾经向Flink提交了相干的issue和PR，曾经被确定为BUG。FLINK-21289
  • containerized.taskmanager.env和containerized.master.env测试下来是失效的，能够生成容器的env