大数据计算

Demo 示例：如何原生的在 K8s 上运行 Flink？

2020-07-24  本文已影响0人  Flink中文社区

Kubernetes 简介

什么是 Kubernetes?

Kubernetes 相信大家都比较熟悉，近两年大家都在讨论云原生的话题，讨论 Kubernetes。那么什么是 Kubernetes 呢？

如果大家对 Yarn、 Mesos 熟悉，假设给定一批裸的物理机，将资源管理系统部署上去之后，可以在此之上基于它的 API 或者 SDK 开发一些分布式软件或者应用程序。例如可以在 Yarn 上开发传统的 MapReduce，在 K8s 上可以开发一些分布式的 Web Server，或者是大数据计算任务等等。

不同于传统的 Yarn，K8s 在所有的进程运行过程中，是全部基于容器化的，但这里的容器并不只是单纯的 Docker 容器，它也包括 Rocket 等其他相关的隔离措施。如果在生产环境中要求比较高的话，可能会有一些安全容器，比如 Kata Containers 等等。K8s 在 Slave 上部署的应用程序，都是用容器化的方式去做分发和管理，同时用容器化的技术做隔离。

它是一个声明式的 API，我们只需要告诉 K8s 集群需要创建一个 Deployment，设置的副本数量，需要达到一个什么样的状态，调度系统也就是 K8s 就会帮助我们维持状态，直到达到设置的状态为止。如果中间发生了一些 failover 或者发生了一些失败，它会自动地将任务迁移到其他的机器上，来满足当前的调度。

目前几乎所有的云厂商都已经提供了 K8s 服务支持，包括国内的阿里、国际上的 Amazon、Google 等等，包括传统的微软都已经提供了对于 K8s 的 Managed 服务或者是 Unmanaged 服务。随着目前 Lambda 表达式或者 Function 计算的应用， Serverless 方式也变得更加流行。除了传统的部署小集群以外，通过云产生一个 manager，构建一个大的 Serverless 集群，然后用户按需进行计算资源付费，这也是一种新的模式。

Kubernetes 的架构

Kubernetes架构图

上图是 K8s 基本的架构，这是一个非常典型的 Master-Slave 的架构。

  1. 在 Master 上，是由 Controller，API Server，Scheduler 以及包括做存储的 Etcd 等构成。Etcd 可以算成 Master，也可以作为独立于 Master 之外的存储来对待。Master 的 Controller、API Server、Scheduler 都是单独的进程模式。这和 Yarn 有一些不同，Yarn 的整个 Master 是一个单进程的模式。K8s 的 Master 还可以在多个 Master 之间完成自发的选举，然后由 active 状态的 Master 对外提供服务。
  2. 在 Slave 上，它主要是包括 Kube proxy、Kubelet，以及 Docker 等相关的组件，每个 Node 上部署的相关组件都是类似的，通过它来管理上面运行的多个 Pod。
  3. 根据不同用户的习惯，可以通过 UI 或者 CLI 的方式向 K8s 提交任务。用户可以通过 K8s 提供的 Dashboard Web UI 的方式将任务进行提交，也可以通过 Kubectl 命令行的方式进行提交。

Kubernetes 的一些概念

ConfigMap 是一个 K-V 数据结构。通常的用法是将 ConfigMap 挂载到 Pod ，作为配置文件提供 Pod 里新的进程使用。在 Flink 中可以将 Log4j 文件或者是 flink-conf 文件写到 ConfigMap 里面，在 JobManager 或者 TaskManger 起来之前将它挂载到 Pod 里，然后 JobManager 去读取相应的 conf 文件，加载其配置，进而再正确地拉起 JobManager 一些相应的组件。

一种对外暴露服务的方式。如果现在内部有一个服务，需要在 K8s 外部进行访问，此时可以通过 Service，然后用 LoadBalancer 或者 NodePort 的方式将其暴露出去。

如果有一个 Service，不希望或不需要将其对外暴露，可以把它设置为 Cluster IP 或者是 None 这种 Headless 的模式。这个时候，它可以用于服务之间相互连接，例如传统的前端去联后端服务，或者是在 Flink 中非 HA 的情况下，TaskManager 去连 JobManager 等等。

Pod 是 K8s 里最小的调度单元。K8s 都是以 Pod 进行调度的。每个 Pod 可以包含一个或者多个 Container。每个 Container 都会有自己的资源，相互之间资源也是已经隔离的，但是所有 Container 共享同一个网络，这就意味着所有的 Container 可以通过 localhost 直接进行通信。

同时，Container 之间可以通过 Volume 共享一些文件。比如 JobManager 或 TaskManager 的 Pod 里产生了一些日志，在同一个 Pod 里再去起另外一个进程收集不符合 K8s 的原生语义。可以通过 SideCar 的方式去起另外一个 Container，把 JobManager 产生的日志收走。这就是一个 Pod 多个 Container 的具体用途。

因为 Pod 是可以随时被终止的，所以当 Pod 终止之后，就无法再拉起来去做 failover 等其他相关操作。Deployment 是在 Pod 之上提供了更高一层的抽象。Deployment 可以设置 Pod 的状态，比如需要起 5 个 TaskManager，Deployment 会维持当前状态。当有 TaskManager 挂了以后，它会起新的 TaskManager，来补上。这样可以避免自己汇报 Pod 的状态，可以去做一些更复杂的管理 failover 等等。这也是最基础的概念——运维自动化。

目前使用情况

目前都有什么样的任务在 K8s 上运行？

除了传统的 Web 以及移动端一些无状态的如 MySQL、Kafka 等存储相关的任务外，有状态的服务也不断地在 K8s 上做适配和运行。除此之外，深度学习框架 Tensorflow 原生即可在 K8s 上运行，包括 Spark、Flink 等等，一些大数据相关的框架也在不断地去兼容，不断地去适配，以便让更多的大数据服务可以更好地在 K8s 上运行。

从这一点我们可以看出， K8s 相比于 Yarn 或传统的 Hadoop 具有更好的包容性，它可以把存储、深度学习、大数据包括 OLAP 分析等多种计算框架、引擎都运行在 K8s 之上。这样就会带来一个很大的好处，整个公司只需要去管理一个调度架构，就可以把所有的存储，实时计算，批量计算，包括深度学习，OLAP 分析等等，都在一个集群里面运行。除了管理更方便以外，也可以达到更好的集群利用率。

Flink On Kubernetes 的部署演进

Flink 在 K8s 上最简单的方式是以 Standalone 方式进行部署。这种方式部署的好处在于不需要对 Flink 做任何改动，同时 Flink 对 K8s 集群是无感知的，通过外部手段即可让 Flink 运行起来。

Standalone Session On K8s

Standalone方式在k8s运行步骤：

Standalone Session On K8s方式

如图所示：

Standalone perjob on K8s

现在我们看一下 Perjob 的部署，因为 Session Cluster 和 Perjob 分别都有不同的适用场景，一个 Session 里面可以跑多个任务，但是每个任务之间没有办法达到更好的隔离性。而 Perjob 的方式，每个job都会有一个自己独立的 Flink Cluster 去运行，它们之间相互独立。

■ Perjob 的特点:

  1. 用户的 Jar 和依赖都是在镜像里提前编译好，或者通过 Init Container 方式，在真正 Container 启动之前进行初始化。
  2. 每个 Job 都会启动一个新的 Cluster。
  3. 一步提交，不需要像 Session Cluster 一样先启动集群再提交任务。
  4. 用户的 main 方法是在 Cluster 里运行。在特殊网络环境情况下，main 方法需要在 Cluster 里运行的话，Session 方式是无法做到的，而 Perjob 方式是可以执行的。
Standalone perjob on K8s方式

■ 执行步骤：

由 Standalone JobCluster EntryPoint 执行，从 classpath 找到用户 Jar，执行它的 main 方法得到 JobGrapth 。再提交到 Dispathcher，这时候走 Recover Job 的逻辑，提交到 JobMaster。JobMaster 向 ResourceManager 申请资源，请求 slot，执行 Job。

Helm Chart 方式

Helm 类似于 Linux 上的 Yum。

K8s 里的 Helm 是一个包管理工具，可以很方便的安装一个包。部署一个 Flink 集群等操作，只需要 helm install 就可以将之前很多步的安装操作，一步去完成。本质上没有什么差别，只是它用 Helm 重新组织，包括一些模板等等，用起来会更加方便。

Flink Kubernetes Operator

Flink Kubernetes Operator方式

使用 Operator 的方式来管理 Flink，主要是来管理多个 Cluster 的情况，可起到任务生命周期管理的作用。它和 Standalone、Native 的方式，本质上不是在一个层次上，它类似于一个更上层的做任务管理的工具。

之前去创建一个 Perjob Cluster，可能需要部署多次，如果任务要做升级，甚至可能需要把之前的删掉，然后修改配置，再重新部署。

引入 K8s Operator 就只需要做一些简单操作。比如 Operator 中有自己的一套 yaml 描述方式，修改其中某一个字段，如修改 image 的 version 字段，此时后台会自动触发一些重启，包括对目前正在执行的任务做 savepoint，然后把 Cluster 销毁掉，再进行新的定向就可以将集群拉起，等一系列自动化的操作。对 Flink 的配置做修改等也都可以在后台自动化完成。

目前 Operater 有 Lyft 和 Google 两个开源的 operator，他们在功能上类似，而且都是已经经过生产检验，与目前的 Standalone Cluster 结合的比较好的，已经达到生产可用的标准。

参考：
1.lyft/flinkk8soperator
https://github.com/lyft/flinkk8soperator
2.GoogleCloudPlatform/flink-on-k8s-operator
https://github.com/GoogleCloudPlatform/flink-on-k8s-operator

总结

当然，Flink on K8s 当前也存在一些不足：

基于这几点，我们在社区推进了一个 Native 的集成方案，这个 Native 类似于 Yarn 这种原生的集成，就是让 Flink 原生的感知到下层 Cluster 的存在。

Navtive Integration 的技术细节

为什么叫 Native 方式？包括如下几个含义。

具体如何工作？主要分 Session 和 Perjob 两个方面来给大家介绍。

Native Kubernetes Session 方式

Native Kubernetes Session方式

首先 Session 的方式。

Native Kubernetes Perjob 方式

Native Kubernetes Perjob方式

我们再来看一下 Perjob 的方式，如图所示，Perjob 方式其实和之前是有一些类似，差别在于不需要先去起一个 Session Cluster，再提交任务，而是一步的。

Session 与 Perjob 方式的不同

我们来看一下 Session 和 Perjob 方式有哪些不同？

Demo 演示

Session

  1. 启动 Session

注意：image 需要替换为自己的镜像或者使用 docker hub 上的 Flink 官方镜像库。

./bin/kubernetes-session.sh \
-Dkubernetes.cluster-id=k8s-session-1 \
-Dkubernetes.container.image=<ImageName> \
-Dkubernetes.container.image.pull-policy=Always \
-Djobmanager.heap.size=4096m \
-Dtaskmanager.memory.process.size=4096m \
-Dtaskmanager.numberOfTaskSlots=4 \
-Dkubernetes.jobmanager.cpu=1 -Dkubernetes.taskmanager.cpu=2
  1. 提交 job 到 Session
./bin/flink run -d -p 10 -e kubernetes-session -Dkubernetes.cluster-id=k8s-session-1 examples/streaming/WindowJoin.jar
  1. 停止 Session
echo 'stop' | ./bin/kubernetes-session.sh -Dkubernetes.cluster-id=k8s-session-1 -Dexecution.attached=true

Application

Application 模式是 FLIP-85 引入的一种新的模式，长远来看是用于替换社区目前的 perjob 模式的。目前 application 模式和 perjob 模式最大的区别是用户代码在 client 端还是 jobmanager 端运行。在 K8s 部署上，由于用户的 jar 和依赖都可以提前打在镜像里面，所以支持 application 模式就变得非常容易。

注意：Application 模式是在 Flink 1.11 中才支持的功能，需要使用对应的 Flink client 和镜像。可以参考社区文档来构建自己的镜像。

./bin/flink run-application -p 10 -t kubernetes-application \
-Dkubernetes.cluster-id=k8s-app1 \
-Dkubernetes.container.image=<ImageName> \
-Dkubernetes.container.image.pull-policy=Always \
-Djobmanager.heap.size=4096m -Dtaskmanager.memory.process.size=4096m \
-Dkubernetes.jobmanager.cpu=1 -Dkubernetes.taskmanager.cpu=2 \
-Dtaskmanager.numberOfTaskSlots=4 \
local:///opt/flink/examples/streaming/WindowJoin.jar

目前功能的状态

FLINK-9953：已经在 Flink 1.10发布：

https://issues.apache.org/jira/browse/FLINK-9953

FLINK-10934：计划在 Flink 1.11发布

https://issues.apache.org/jira/browse/FLINK-10934

FLINK-12884：目前高可用方式是基于 zk 实现的，未来是希望不依赖于外部组件，基于 K8s 的 ConfigMap 去做 Meta 存储和 Leader 选举。目前已经有内部实现，未来将会贡献给社区。

https://issues.apache.org/jira/browse/FLINK-12884

FLINK-14460：计划陆续在 Flink 1.11/1.12两个版本中发布和完善

https://issues.apache.org/jira/browse/FLINK-14460

包括内容如下:

目前社区的 Flink on K8s 的 native 方案还在快速发展和完善，希望大家多多试用并且提出反馈意见，如果有兴趣也非常欢迎一起参与进来开发。

上一篇 下一篇

猜你喜欢

热点阅读