该页面介绍了如何在Kubernetes上本地部署Flink 。

入门

本入门部分将指导您在Kubernetes上设置功能齐全的Flink集群。

介绍

Kubernetes是一种流行的容器编排系统，用于自动执行计算机应用程序的部署，扩展和管理。Flink的本地Kubernetes集成使您可以在正在运行的Kubernetes集群上直接部署Flink。此外，Flink能够根据所需资源动态分配和取消分配TaskManager，因为它可以直接与Kubernetes通信。

准备

在入门部分假定运行Kubernetes集群满足以下要求：

Kubernetes> = 1.9。

KubeConfig，可以访问列表，创建，删除容器和服务，可以通过进行配置~/.kube/config。您可以通过运行来验证权限kubectl auth can-i <list|create|edit|delete> pods。

启用Kubernetes DNS。

default具有RBAC权限的服务帐户可以创建，删除Pod。

如果在设置Kubernetes集群时遇到问题，请看一下如何设置Kubernetes集群。

在Kubernetes上启动Flink Session

在运行Kubernetes集群并kubectl配置为指向它之后，您可以通过以下方式在会话模式下启动Flink集群：

# (1) Start Kubernetes session

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

# (2) Submit example job

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

# (3) Stop Kubernetes session by deleting cluster deployment

$ kubectl delete deployment/my-first-flink-cluster

注意使用Minikube时，需要调用minikube tunnel以便在Minikube上公开Flink的LoadBalancer服务。

恭喜你！通过在Kubernetes上部署Flink，您已经成功运行了Flink应用程序。

部署方式

对于生产用途，我们建议在应用程序模式下部署Flink应用程序，因为这些模式为应用程序提供了更好的隔离。

应用模式

所述应用程序模式需要用户代码是捆绑在一起的 Flink镜像一起，因为它在集群上运行用户代码的main()方法。应用程序模式可确保在终止应用程序后正确清理所有Flink组件。

Flink社区提供了一个基本的Docker映像，可用于捆绑用户代码：

FROM flink

RUN mkdir -p $FLINK_HOME/usrlib

COPY /path/of/my-flink-job.jar $FLINK_HOME/usrlib/my-flink-job.jar

在下创建并发布Docker映像后custom-image-name，您可以使用以下命令启动应用程序集群：

$ ./bin/flink run-application \

--targetkubernetes-application \

-Dkubernetes.cluster-id=my-first-application-cluster \

-Dkubernetes.container.image=custom-image-name \

local:///opt/flink/usrlib/my-flink-job.jar

注意 local是“应用程序模式”中唯一受支持的方案。

该kubernetes.cluster-id选项指定集群名称，并且必须是唯一的。如果您未指定此选项，则Flink将生成一个随机名称。

该kubernetes.container.image选项指定用于启动Pod的image。

部署应用程序集群后，您可以与其进行交互：

# List running job on the cluster

$ ./bin/flink list --target kubernetes-application -Dkubernetes.cluster-id=my-first-application -cluster

# Cancel running job

$ ./bin/flink cancel --target kubernetes-application -Dkubernetes.cluster-id=my-first-application-cluster  <jobId>

您可以conf/flink-conf.yaml通过将键值对传递-Dkey=value给来覆盖设置中的配置bin/flink。

Per-Job Cluster Mode

Kubernetes上的Flink不支持 Per-Job Cluster Mode。

会话模式

在此页面顶部的《入门指南》中，您已经看到了会话集群的部署。

会话模式可以以两种模式执行：

分离模式（默认）：kubernetes-session.sh在Kubernetes上部署Flink集群，然后终止。

附加模式（-Dexecution.attached=true）：kubernetes-session.sh保持活动状态，并允许输入命令来控制正在运行的Flink群集。例如，stop停止正在运行的会话群集。键入help以列出所有受支持的命令。

为了my-first-flink-cluster使用集群ID重新附加到正在运行的Session集群，请使用以下命令：

$ ./bin/kubernetes-session.sh\

-Dkubernetes.cluster-id=my-first-flink-cluster\

-Dexecution.attached=true

您可以conf/flink-conf.yaml通过将键值对传递-Dkey=value给来覆盖设置中的配置bin/kubernetes-session.sh。

停止正在运行的会话群集

为了停止正在运行的具有群集ID的会话群集，my-first-flink-cluster您可以删除Flink部署，也可以使用：

$ echo'stop' | ./bin/kubernetes-session.sh \

-Dkubernetes.cluster-id=my-first-flink-cluster \

-Dexecution.attached=true

在Kubernetes上的Flink参考

Configuring Flink on Kubernetes

配置页面上列出了特定于Kubernetes的配置选项。

访问Flink的Web UI

Flink的Web UI和REST端点可以通过kubernetes.rest-service.exposed.type配置选项以多种方式公开。

ClusterIP：在群集内部IP上公开服务。该服务仅在群集内可用。如果要访问JobManager UI或将作业提交到现有会话，则需要启动本地代理。然后，您可以localhost:8081用于将Flink作业提交到会话或查看仪表板。

$ kubectl port-forward service/ 8081

NodePort：在静态端口（NodePort）上的每个节点的IP上公开服务。 <NodeIP>:<NodePort>可用于联系JobManager服务。 NodeIP也可以用Kubernetes ApiServer地址代替。您可以在kube配置文件中找到其地址。

LoadBalancer：使用云提供商的负载平衡器在外部公开服务。由于云提供商和Kubernetes需要一些时间来准备负载均衡器，因此您可能会NodePort在客户端日志中获得JobManager Web界面。您可以kubectl get services/<cluster-id>-rest用来获取EXTERNAL-IP并手动构建负载均衡器JobManager Web界面 http://<EXTERNAL-IP>:8081。

请参阅Kubernetes中有关发布服务的官方文档以获取更多信息。

Logging

Kubernetes集成向Pod公开conf/log4j-console.properties并conf/logback-console.xml作为ConfigMap。这些文件的更改将对新启动的群集可见。

访问日志

默认情况下，JobManager和TaskManager会将日志/opt/flink/log同时输出到控制台和每个窗格中。该STDOUT和STDERR输出只会被重定向到控制台。您可以通过访问它们

$ kubectl logs

如果Pod正在运行，则还可以使用它kubectl exec -it <pod-name> bash来建立隧道并查看日志或调试过程。

访问TaskManagers的日志

Flink将自动取消分配空闲的TaskManager，以免浪费资源。此行为可能使访问相应pod的日志变得更加困难。您可以通过配置resourcemanager.taskmanager-timeout来增加释放空闲的TaskManager之前的时间，以便您有更多的时间检查日志文件。

动态更改日志级别

如果已将记录器配置为自动检测配置更改，则可以通过更改相应的ConfigMap（假设集群ID为my-first-flink-cluster）来动态调整日志级别：

$ kubectl edit cm flink-config-my-first-flink-cluster

使用插件

为了使用插件，您必须将它们复制到Flink JobManager / TaskManager pod的正确位置。您可以使用内置插件，而无需装载卷或构建自定义Docker image。例如，使用以下命令为您的Flink会话群集启用S3插件。

$ ./bin/kubernetes-session.sh

-Dcontainerized.master.env.ENABLE_BUILT_IN_PLUGINS=flink-s3-fs-hadoop-1.12.0.jar \

-Dcontainerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS=flink-s3-fs-hadoop-1.12.0.jar

自定义Docker image

如果要使用自定义Docker映像，则可以通过configuration选项指定它kubernetes.container.image。Flink社区提供了丰富的Flink Docker image，可以作为一个很好的起点。查看如何自定义Flink的Docker映像，以了解如何启用插件，添加依赖项和其他选项。

使用秘密

Kubernetes Secrets是一个包含少量敏感数据（例如密码，令牌或密钥）的对象。否则，此类信息可能会放在pod specification或image中。Kubernetes上的Flink可以通过两种方式使用Secrets：

将Secrets用作Pod中的文件；

使用Secrets作为环境变量；

将Secrets 用作Pod中的文件

以下命令会将密钥安装在启动的容器中mysecret的路径下/path/to/secret：

$ ./bin/kubernetes-session.sh -Dkubernetes.secrets=mysecret:/path/to/secret

mysecret然后可以找到秘密文件的用户名和密码，并将它们存储在/path/to/secret/username和中/path/to/secret/password。有关更多详细信息，请参见Kubernetes官方文档。

使用Secrets 作为环境变量

以下命令将mysecret在启动的pod中将密钥作为环境变量公开：

$ ./bin/kubernetes-session.sh-Dkubernetes.env.secretKeyRef= \

env:SECRET_USERNAME,secret:mysecret,key:username; \

env:SECRET_PASSWORD,secret:mysecret,key:password

env变量SECRET_USERNAME包含用户名，而env变量SECRET_PASSWORD包含secret的密码mysecret。有关更多详细信息，请参见Kubernetes官方文档。

Kubernetes的高可用性

为了在Kubernetes上实现高可用性，您可以使用现有的高可用性服务。

手动资源清理

Flink使用Kubernetes OwnerReference来清理所有集群组件。所有flink创建的资源，其中包括ConfigMap，Service，和Pod，有OwnerReference设定为deployment/<cluster-id>。删除部署后，所有相关资源将自动删除。

$ kubectl delete deployment/<cluster-id>

支持的Kubernetes版本

当前，>= 1.9支持所有Kubernetes版本。

命名空间

Kubernetes中的命名空间通过资源配额在多个用户之间划分集群资源。Kubernetes上的Flink可以使用名称空间来启动Flink集群。可以通过kubernetes.namespace配置名称空间。

RBAC

基于角色的访问控制（RBAC）是一种基于企业内各个用户的角色来调节对计算资源或网络资源的访问的方法。用户可以配置JobManager用来访问Kubernetes集群中的Kubernetes API服务器的RBAC角色和服务帐户。

每个名称空间都有一个默认服务帐户。但是，default服务帐户可能无权在Kubernetes集群中创建或删除Pod。用户可能需要更新default服务帐户的权限或指定绑定了正确角色的另一个服务帐户。

$ kubectl create clusterrolebinding flink-role-binding-default--clusterrole=edit--serviceaccount=default:default

如果您不想使用default服务帐户，请使用以下命令创建新的flink-service-account服务帐户并设置角色绑定。然后使用config选项-Dkubernetes.service-account=flink-service-account使JobManager窗格使用flink-service-account服务帐户创建/删除TaskManager窗格和领导者ConfigMap。同样，这将允许TaskManager监视领导者ConfigMap，以检索JobManager和ResourceManager的地址。

$ kubectl create serviceaccount flink-service-account

$ kubectl create clusterrolebinding flink-role-binding-flink--clusterrole=edit--serviceaccount=default:flink-service-account

请参阅有关RBAC授权的Kubernetes官方文档以获取更多信息。