信也容器云揭秘02-Flink上云

Flink作为大数据和实时数据部门重要的框架和引擎，扮演着重要的角色，Flink的使用也越来越多，集群管理也变得越来越不容易。为了支持Flink上云，容器云团队也对其做了大量的探索工作，以保证Flink能够更好的且平稳的进行容器化。

一：部署方式选择

目前信也上云的Flink版本是Flink 1.11，Flink 1.11基于kubernetes的部署模式有：Session、Per-job、Application三种模式，下面说明三种部署模式的对比

这三种模式的区别在于：

• 集群生命周期和资源隔离保证
• 应用程序的main()方法是在客户端还是在集群上执行

​ 图：1-1

Application模式

用户程序的 main 方法将在集群中而不是客户端运行，用户将程序逻辑和依赖打包进一个可执行的 jar 包里，集群的入口程序 (ApplicationClusterEntryPoint) 负责调用其中的 main 方法来生成 JobGraph。Application 模式为每个提交的应用程序创建一个集群，该集群可以看作是在特定应用程序的作业之间共享的会话集群，并在应用程序完成时终止。在这种体系结构中，Application 模式在不同应用之间提供了资源隔离和负载平衡保证。在特定一个应用程序上，JobManager 执行 main() 可以节省所需的 CPU 周期，还可以节省本地下载依赖项所需的带宽。

Per-Job 模式

per-job模式是为每个提交的作业启动Flink集群，拥有自己的jobmanager和taskmanager。所以在启动的时候该作业模式可能延迟会比较高点。作业完成后，集群将销毁，并清理所有的资源。此模式允许更好的资源隔离，因为运行有问题的作业也不会影响到其它作业。

Session模式

session模式假定一个已经在运行的集群，并使用该集群的资源来执行任何提交的应用程序。在同一个（会话）集群中执行的应用程序使用相同的资源，并因此竞争相同的资源。你并没有为每一项工作付出全部的开销。但是，如果其中一个作业行为不当或导致TaskManager崩溃，则该TaskManager上运行的所有作业都将受到该故障的影响。除了对导致失败的作业造成负面影响外，这意味着一个潜在的大规模恢复过程，即所有重新启动的作业同时访问文件系统，并使其对其他服务不可用。另外，让一个集群运行多个作业意味着JobManager的负载更大，JobManager负责记录集群中的所有作业。这种模式非常适合于启动延迟非常重要的短作业，例如交互式查询。

目前信也科技在服务的容器化方面的支持已经很成熟，有一套完善的构建发布流程，所以通过对比Flink的几种部署模式的优缺点，最终我们采用了Application的部署方式，该方式相比于其它两种模式优点明显，拥有更好的隔离性，同时对资源的利用率也高，也更符合我们现有的发布流程规范。

二：Flink on k8s

​ 图：2-1

创建Flink Kubernetes集群时，Flink客户端将首先连接到Kubernetes ApiServer提交集群，包括ConfigMap，Job Manager。然后，Kubernetes将创建JobManager，在此期间，Kubelet将拉取镜像，准备并挂载，然后执行启动命令。启动JobManager命令后，Dispatcher和KubernetesResourceManager可用，并且群集已准备好接受一个或多个作业。当用户通过Flink客户端提交作业时，客户端将生成 Job graph ，并将其与用户jar一起上传到Dispatcher。JobManager向KubernetesResourceManager请求slots资源。如果没有可用的slots，资源管理器生成TaskManager并在集群中注册。这是Flink 在在kubernetes内部的简要交互方式。

三：构建发布

信也采用的Flink版本为1.11,部署模式是Application。我们把每个job抽象成一个应用。所以每个job的发布流程也就是信也普通应用一样的发布流程:

• 申请Flink job相关的应用
• 非1.11版本的job升级到1.11版本，并集成maven 镜像构建打包插件
• 通过aladdin打包平台，打包镜像。通过stargate平台选择相应应用和打包的镜像版本进行job的发布

​ 图：3-1

​ 图：3-2

在程序进行升级的时候，停止job可以采用savepoint的机制来保持作业的完整快照，在下次启动的时候可以利用保存的savepoint来进行作业的恢复

四：监控告警

Flink部署在kubernetes上后，job的监控和运维也需要相应的配套设施才行。 当Flink job在运行过程中挂掉了，我们怎么才能监控到并产生告警？job在运行过程中可能会出现不健康的运行，比如checkpoint时间过长、gc频繁、或者发生了重启。这些我们又如何监控？

1. 配置探针

Flink job在运行过程中由jobmanager进行资源管理、作业调度，所以我们为每个Flink job中的jobmanager添加探针，检测job是否正常运行，当健康检测不通过，我们通过zmon平台进行告警

 readinessProbe:
        httpGet:
          path: /jobs/overview
          port: 8081
          scheme: HTTP
        initialDelaySeconds: 30
        timeoutSeconds: 3
        periodSeconds: 5
        successThreshold: 3
        failureThreshold: 12

2.指标收集

目前公司上云的应用都是采用prometheus进行指标进行收集，所以Flink的指标收集我们仍然采用prometheus进行收集。利用grafana进行展示（下图进展示部分）

​ 图：4-1

1.对于系统指标最常关注的是作业的可用性，如 uptime (作业持续运行的时间)、fullRestarts (作业重启的次数)
2.另外是 checkpoint 相关信息，例如最近完成的 checkpoint 的时长(lastCheckpointDuration )、最近完成的 checkpoint 的大小(lastCheckpointSize )、作业失败后恢复的能力(lastCheckpointRestoreTimestamp )、成功和失败的 checkpoint 数目(numberOfCompletedCheckpoints、numberOfFailedCheckpoints)以及在 Exactly once 模式下 barrier 对齐时间(checkpointAlignmentTime)

五：高可用

作业可能由于机器维护、硬件故障导致挂掉，这时候如何快速恢复作业也成为了需要思考的问题。目前我们采用的方式是可以通过程序的方式自动恢复作业

​ 图：5-1

因为部分job并不适用这种方式来恢复job，所以在平台可以设置，如果job设置了启用高可用，默认情况下，检测到job挂掉，会采用checkpoint的机制来直接恢复job，如果没有设置，job挂掉后会通知对应的job负责人，负责人收到告警后，需要手动来恢复job

六：遇到的问题

1. 网络问题

Flink在部署的过程中需要访问Apiserver,这时候job需要通过clusterip来访问ApiServer，而公司集群采用的是macvlan网络是无法访问clusterip的，所以为了支持Flink部署，我们在大数据集群通过给对应pod添加双网卡的方式（一块是macvlan，一块是bridge）

2. ip管理

因为Flink部署在kubernetes是采用的deployment方式部署，这种方式部署的pod，我们无法管理相关Flink pod的ip，所以容器云团队，部署了一个webhook的服务，来监听集群pod的相关事件，来分配和释放相关的ip

​ 图：6-1

3.配置问题

Flink在运行过程中在做checkpoint和savepoint时都依赖于hdfs，所以pod在运行过程中是需要访问hdfs服务的，我们是通过，将hdfs-site.xml、core-site.xml相关配置提前配置到kubernetes集群Configmap里，并在Flink启动过程中指定相应的ConfigMap

​ 图：6-2

信也容器云平台

stargate作为信也科技的容器云平台，是基于kubernetes开发的一套企业级容器管理平台，具有业务场景覆盖全面、生态丰富的特点，目前已经开源

开源地址：https://github.com/ppdaicorp/stargate

扫描加群讨论（备注：stargate）