Spark on K8s 在vivo大数据平台的混部实战

QUUd39buBuWg

作者：vivo 互联网大数据团队- Qin Yehai
 
在离线混部可以提高整体的资源利用率，不过离线Spark任务部署到混部容器集群需要做一定的改造，本文将从在离线混部中的离线任务的角度，讲述离线任务是如何进行容器化、平台上的离线任务如何平滑地提交到混部集群、离线任务在混部集群中如何调度的完整实现以及过程中的问题解决。

 
一、在离线业务差异

 
互联网数据业务服务一般可以分为在线服务和离线任务两大类，在线服务是指那些长时间运行、随时响应对实时性要求高、负载压力随着接收流量起伏的服务，如电商、游戏等服务，离线任务是指运行周期短、可执行时间提交对实时性要求低、有一定容错性、负载压力基本可控的服务，如离线计算任务、模型训练等。一般在线服务在白天时段繁忙，离线任务在凌晨繁忙，两者的业务高峰期存在错峰现象，如果按传统方式在线和离线都是分别独立机器部署，业务高峰时期需要更多机器来支持，业务低峰期又存在部分机器空闲，整体资源利用率都不高。因此行业提出来在离线混部的解决方案，在线和离线业务通过混部系统部署在同一批机器，实现共享资源并错峰互补，提高整体的资源利用率。目前业内利用混部技术可以将数据中心的CPU利用率提升至40%左右，vivo在2023年混部平台投入生产也已经将部分混部集群的CPU利用率提升至30%左右，整体收益也是可观的。
 
混部系统需要有强大的隔离能力，绝大部分都是基于容器，所以混部的前提是在线和离线业务都容器化，对于容器管理工具如K8s来说是更适应于运行时间长、启停次数少、容器数量少的在线服务，在线服务也能比较容易地上容器，而对于运行时间短、启停频繁、容器数量大的离线任务，对K8s来说不是天然地适应，但容器化已是大势所趋，K8s也推出了性能更好的调度器、用于离线任务的控制器，Spark在2.3版本后也支持容器化，诸多技术的发展也推动离线任务实现容器化以及在离线混部的落地。
 
本文将从在离线混部中的离线任务的角度，讲述离线任务是如何进行容器化、平台上的离线任务如何平滑地提交到混部集群、离线任务在混部集群中如何调度的完整实现以及过程中的问题解决。
 
二、离线任务容器化

 
2.1 Spark Operator 方案

2.1.1 方案对比

 
vivo离线任务大部分任务是以Spark作为执行引擎，Spark任务运行在K8s上，目前业界有两种架构的方案：Spark on K8s及Yarn on K8s。两者部分优缺点对比如下：
 

 
Spark on K8s是Spark容器化，由K8s直接创建Driver和Executor的Pod来运行Spark作业，Yarn on K8s是Yarn的容器化，由K8s创建RM和NM的Pod，Spark的Driver和Executor运行在NM Pod的container中，正是由于两种架构方案的区别，它们各自也会存在优缺点。
 
Yarn on K8s方案可以支持原生的Hive、Spark、Flink等引擎，它仅需要创建一定数量的NodeManager Pod来满足作业需求，Pod运行相对稳定因此对K8s的压力比较小，本身Yarn支持调度性能和调度策略也是专门为离线任务设计的，调度性能比K8s的强很多。由于NodeManager ESS服务是对磁盘有容量和读写性能要求的，混部机器的磁盘一般难以满足，所以也需要能支持不同引擎的Remote Shuffle Service。在资源利用上，NodeManager需要满足多个作业的资源，最小单位是Container，Pod的资源粒度比较大，自身也会占用一些资源，如果资源粒度得不到有效地弹性伸缩，也会造成资源的浪费，因此需要引入额外的组件来协调,根据Kubernetes集群节点的剩余资源，动态调整NodeManager的CPU和内存，然而这也需要一定的改造成本。在资源紧张的情况下，NodeManager Pod如果被驱逐也就意味着整个NodeManager被销毁，将会影响多个任务。
 
Spark on K8s方案目前在Spark 3.1以上版本才正式可用，它需要频繁的创建、查询、销毁大量的Executor Pod，对K8s的ApiServer和ETCD等组件都会造成比较大的压力，K8s的调度器也不是专门为离线的大批量任务设计的，调度性能也比较弱。另一方面，Spark on K8s虽然只能支持Spark3.X的RSS，不过目前有较多的开源产品可选择。在资源利用上，最小单位是Driver和Executor的Pod，资源粒度小，可以填充到更多的碎片资源，调度时直接与K8s对接，资源的弹性调度更多由K8s来承担，不需要额外的组件，改造成本比较低。在资源紧张的情况下，Executor、Driver的Pod将依次逐个被驱逐，任务的稳定性会更高。
 
而对于Spark on K8s方案，还细分2种实现方案：Spark Submit on K8s和Spark Operator on K8s。
 

SparkOnK8s架构图
(图片来源：Spark官网) 
 

Spark Operator架构图
(图片来源：Spark Operator开源项目)
 
以spark-submit方式提交到K8s集群是Spark在2.3版本后提供的原生功能，客户端通过spark-submit设置K8s的相关参数，内部再调用K8sApi在K8s集群中创建Driver Pod，Driver再调用K8sApi创建需要的Executor Pod，共同组成Spark Application，作业结束后Executor Pod会被Driver Pod销毁，而Driver Pod则继续存在直到被清理。使用spark-submit方式的最大好处是由spark-submit来与K8s的进行交换，提交作业的方式几乎保持一致。但是因为使用的便利性所需要的封装也会带来一些缺点，spark-submit是通过K8sApi创建Pod，使用非声明式的提交接口，如果需要修改K8s配置就需要重新开发新接口，二次开发复杂繁琐，虽然Spark提供了大量的K8s配置参数，但也远比不了K8s YAML的声明式的提交方式更加灵活，而且Spark Application和K8s Workload的生命周期还不能较好地对应起来，生命周期不能灵活控制，任务监控也比较难接入Prometheus集群监控。虽然Spark社区也不断地在推出新特性来和K8s集成地更加灵活，不过对于些复杂场景需要定制开发，spark-submit的封装性也会成为阻碍。
 
spark-submit还是离线任务提交的思维，而Spark Operator方式就更倾向于K8s作业的思维，作为K8s的自定义控制器，在集成了原生的Spark on K8s的基础上利用K8s原生能力提供了更全面管控功能。Spark Operator使用声明式的YAML提交Spark作业，并提供额外组件来管理Spark作业的生命周期，SparkApplication控制器，负责SparkApplicationObject的创建、更新和删除，同时处理各种事件和作业状态，Submission Runner, 负责调用spark-submit提交Spark作业，Driver和Executor的运行流程是一致的，Spark Pod Monitor，负责监控和同步Spark作业相关Pod的状态。Spark Operator最大的好处是为在K8s中的Spark作业提供了更好的控制、管理和监控的功能，可以更加紧密地与K8s结合并能灵活使用K8s各种特性来满足复杂场景，例如混部场景，而相对地它也不再像spark-submit那样方便地提交任务，所以如何使用Spark Operator优雅提交任务将是在离线混部中一项重要的工作。
 
2.1.2 最终选项

在大的架构选型上，我们选择了Spark on K8s，一方面因为Spark3.X是vivo当前及未来2~3年的主流离线引擎，另一方面vivo有比较完善的K8s生态体系，内部对K8s研发也比较深入，环境和能力都能很好地支持，在应用的小方向上，我们选择了Spark Operator，因为它在混部这种复杂场景下使用更加灵活、扩展性更强、改造成本更低，我们最终决定使用Spark Operator方案。
 
2.2 Spark优化

2.2.1 Spark镜像

Spark任务容器化的第一步就是构建具有Spark相关环境的镜像，Spark任务类型主要分为sql任务和jar任务，在实践的过程中我们发现Spark的镜像构建需要注意几个问题：

• Spark环境的完整性：镜像中除了打入自研的Spark包以外，还需要打入相应的依赖如Hadoop、ZSTD、RSS等包，对于SparkJar任务还有直接调用Hadoop客户端的，因此Hadoop客户端也需要打入镜像中。
  

 

• JDK版本问题：K8s使用的Spark是基于3.2.0版本，镜像打包工具默认使用JDK11，而自研的Spark用的JDK1.8，由于在Yarn和K8s上使用的JDK版本不同，导致在双跑验证数据一致性时发现了hash函数、时间戳不一致的问题，因此Spark镜像中的JDK版本需要和Yarn保持一致。
  

 

• 环境变量问题：镜像生成容器后需要预置如Spark、Hadoop的环境变量，如果镜像中相关目录的位置不能完全和Yarn的提交节点保持一致，则需要检查各启动脚本，如spark-env.sh中的环境变量的路径是否存在，发生冲突时可以修改为绝对路径。
  

 
Spark镜像构建完成后，区分SparkSql任务和SparkJar任务实质就是启动命令的不同，事实上SparkSql任务也就是SparkJar任务的一种，只是启动的主类是固定的，两者的启动参数如下：
 
SparkSql任务：
driver --class org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver -f {sql文件} 
SparkJar任务：
driver --class {jar任务主类} {jar任务jar包} {参数} 
早期不仅构建了Spark镜像，还构建了Spark日志镜像，容器组成结构会复杂一些。如图例如Driver容器，我们将Spark、Hadoop等配置文件构建了configMap，启动initContainer来拉取从configMap拉取配置文件，然后启动Driver容器执行Spark任务，同时也使用sidecar创建日志上报的容器，在Spark任务运行完成后上报Driver和Executor日志到Spark HistoryServer。这样的方案看似充分应用了K8s技术，但是在实践的过程中这些技术却被一一弃用，转而逐步地把各种功能集中到了一个Driver容器上。
 

 
具体演进如下：

• 移除initContainer，拉取Spark等配置文件步骤写在启动命令中，Spark作业执行前执行下载配置，原因在多个namespace下不方便统一管理，而且configmap内容较大，会导致Pod启动时配置加载的延迟增加，影响了Pod创建速度，同时K8s的内存和CPU资源占用增加，对kube-apiserver、ETCD负载有一些影响。去掉initContainer还有个重要的好处就是减小ETCD的存储压力，事实上我们在移除initContainer拉取配置的功能后的一段时间内还保留着initContainer，在任务逐渐上量后发现ETCD的存储比较满，分析后发现Spark作业中的一个Pod生命周期大约8次更新，其中initContainer更新会占用2次，移除了之后理论上是可以减少1/4的ETCD存储，实际应用中完全去除了initContainer也确实能减小了ETCD的存储压力。
  

 

• 移除sidecar创建日志上报的容器，Driver和Executor日志上报步骤写在启动命令中，Spark作业执行完后再执行脚本上报，原因是sidecar在同一个Pod中与主容器共享相同的生命周期，不使用sidecar方式就能更快创建Pod，Spark任务执行完成后能更快释放资源。
  

 
对于Spark作业会频繁创建、更新和销毁大量的Pod，所以去除非必要的容器，提高Pod生命周期流转速度，就能降低kube-apiserver、ETCD工作负载，也能提高Spark的作业效率。
 
2.2.2 Spark改造

Spark任务运行在K8s上，对于一些使用的兼容问题也进行了相关改造。
 

• HistoryServer改造，因为Spark Operator没有存储已结束作业的日志，因此参考了on Yarn的方式，在Spark作业结束后，通过日志上传脚本把Driver和Executor的日志上传HDFS，与Yarn日志聚合类似，同时也在Spark HistoryServer做了二次开发工作，增加了on K8s方式的日志查看接口，用户查看已完成的Executor日志时，不再请求JobHistory Server，而是请求Spark HistoryServer接口。但日志上传方式需要Executor执行完才能查看到日志，为了能实时查看到执行中的日志，可以在Executor内部实现一个HTTP服务，根据Pod以及端口信息拼接出日志请求URL，Executor启动一个Servlet自动获取本地日志并返回。日志查看体验上做到了基本与Yarn一致。
  

 

• 主机ip通信，Spark Driver和Executor之间的通信通常是通过主机名进行的，不过随着Spark任务增多，CoreDNS因为频繁的域名解释请求导致压力增大，甚至会影响到在线服务，因此我们将Hadoop的配置文件改为ip格式、设置Driver和Executor使用ip地址，同时去除了对应的K8s Service，通过访问ip而不是域名的方式来规避这个问题。
  

 

• 文件参数兼容，Spark Driver在K8s上是运行在某一个Pod中的，所以文件需要是全局可视的，如HDFS文件，否则就会报文件未找到的错误，但Spark作业运行在大数据作业平台时有的任务使用的上传的本地文件，因此对于提交到K8s的任务，第一步是要把上传到大数据作业平台的文件再次上传到HDFS，第二步是改造add jar和--file等命令逻辑，Spark任务在未能读取本地文件后将再尝试读取二次上传到HDFS的文件，实现任务无需修改成全局可视的文件路径也能读取到文件。
  

 

• non-daemon线程终止，在K8s上运行的Spark任务是指定Client模式，Client模式下Driver遇到异常时停掉SparkContxet，等所有non-daemon线程结束后，Driver才会退出，但如果存在一直运行的non-daemon线程，那么Driver一直不退出，任务就一直处于执行中。因此需要改造成Cluster模式的异常退出机制，即异常时以非0退出码退出，不再等待其他的non-daemon线程结束，Driver直接终止，以确保Driver Pod的正常结束。
  

 
2.3 Spark Operator优化

随着在K8s上运行的Spark任务不断增加，K8s集群的负载也逐渐显现。因此，需要对Spark Operator进行一系列优化，以减轻K8s集群的压力。
 

• 离线使用独立的kube-apiserver，混部集群中离线容器占了很大一部分，而且离线任务由于生命周期短，容器创建销毁更加频繁，这对kube-apiserver造成了很大的压力，然而在线业务需要更高的稳定性，为了减少离线对在线业务的影响，我们拆分了kube-apiserver，离线任务通过指定master参数来使用独立的kube-apiserver。
  

 

• 使用K8s的HostNetwork网络模式，在K8s上启动Driver与Executor虽然使用的是独立ip+固定端口，但频繁的ip申请和释放也对kube-apiserver造成了一定的压力，因此我们改为使用HostNetwork网络模式，同时不指定端口避免端口冲突。
  

 

• 优化Spark Operator控制器的队列，在任务量比较大的情况下，Spark Operator对Pod创建消耗效率会遇到瓶颈，排查后发现是Spark Operator的事件处理队列的并发数和限速桶的默认配置地太小，因此我们调低Spark maxPendingPods参数，调高schedulerBacklogTimeout、 sustainedSchedulerBacklogTimeout参数，减少Pending Pod个数，使Pod的处理效率符合集群的承载水平。
  

 

• 优化Spark Driver List Pod接口，使用kube-apiserver缓存，避免对ETCD产生影响，同时修改Spark Driver清理Executor逻辑，直接Delete，减少List Pod对kube-apiserver压力。
  

 

• 存储emptydir + log lv存储优化，开发CSI插件，Spark任务的离线日志单独存储，避免对在线业务pod的影响和磁盘负载高等问题。
  

 

• Spark Secret标记immutable，减少kubelet watch secret请求，降低kube-apiserver的负载。
  

 
三、离线任务提交

3.1 平台任务提交平滑切换

离线任务容器化方案确定后就要落地到生产，目前有SparkSql和SparkJar两种离线任务实现了容器化，这里以SparkSql任务为例描述Spark提交到混部K8s集群的流程并达到与传统客户端提交任务几乎无差异的平滑切换。目前vivo的离线任务都是通过大数据平台进行提交和调度的，平台会把主要的提交流程进行封装形成简单操作的功能，例如在平台上提交SparkSql任务流程一般是编写sql、提交任务、查看Driver日志或在跳转到SparkUI、执行完成后获取结果以及更新任务状态。
 
在平台内部，SparkSql任务使用传统的spark-submit提交流程是：

• 用户编写好的sql上传到提交节点生成一个sql文件；
  
• 在提交节点使用Spark客户端执行该sql文件启动SparkSql任务；
  
• 任务启动后，通过不断地tail操作查询日志转存到HBase方便在平台页面上查询到Driver日志；
  
• 任务结束后，再查询输出结果转存到HBase方便在平台页面上查询到执行结果；
  
• 根据提交sql任务命令的返回码来更新任务状态。
  

 
传统Spark客户端提交任务大部分只会涉及到提交节点的客户端与平台服务器之间的交互，而SparkSql任务提交到混部K8s集群，从上节的Spark容器化方案的原理可知最终目的是要将Spark任务的任务参数按一定的格式封装好传入Spark Operator控制器来创建相关的容器，平台需要通过会调用容器团队提供的封装好K8sApi的统一接入层来创建Spark容器。
 

 
在平台内部，SparkSql任务提交到混部K8s集群的完整流程为：

• 用户编写好的sql上传到HDFS生成一个远程可访问的HDFS文件；
  
• SparkSql任务参数封装好传入容器接入层的createSpark接口来调用Spark Operator控制器容器，再由Spark Operator控制器创建Driver Pod，最后由Driver Pod根据Spark任务需要创建多个Executor Pod，这些Driver、Executor的Pod相当于Driver和Executor的角色，共同配合执行Spark作业；
  
• 任务启动后，通过容器接入层的getDriverLog接口周期性地查询Driver日志，实质上是查询Driver容器的日志，查询到的Driver日志会转存到HBase方便在平台页面上查询；
  
• 任务结束后，一方面通过Spark启动脚本中的日志上传命令，把Driver和Executor的日志上传HDFS，可以在改造后的Spark HistoryServer直接查看，另一方面执行结果也会先输出到HDFS，再从HDFS转存到HBase方便在平台页面上查询到执行结果；
  
•  通过轮询接入层的getSpark接口根据返回的状态码来更新任务状态，在任务结束后，此时Driver Pod不会主动退出，首先将任务状态更新为成功，在日志和结果都存储完成后，再调用deleteSpark接口主动地杀死Driver Pod释放资源，完成整个Spark任务流程。
  

 
可以看出SparkSql任务提交到混部K8s的执行主体是容器，因此需要增加容器接入层来管理Spark相关的容器，同时容器的使用更倾向于存算分离的效果，因此需要使用HDFS作为远程文件中转。
 
大数据平台上传统使用spark-submit和onK8s使用spark-operator的SparkSql任务执行流程对比如下：
 

 
3.2 混部任务的资源参数调整

Spark任务的Driver和Executor，在Yarn上执行实质是运行在NodeManager节点上的，而在K8s上执行实质是运行在对应的Pod中的，由于Spark on K8s的提交方式和运行环境都不同于on Yarn，任务的资源参数不能直接套用，需要做一些参数调整才能提交到K8s上。
 
1、资源参数提取和转换
SparkSql任务在Yarn上可以灵活地调整sql中的配置来满足不同特性的任务，sql中的资源配置会覆盖客户端启动时的全局配置，因为Executor是运行在NodeManager节点上的，资源会相对充裕能满足Executor的资源需求，与此不同的是Spark on K8s的Executor是运行在Executor Pod中的，使用的资源会受到Pod资源规格大小的限制，而spark-operator的提交方式是要先获取Executor全局资源规格并生成相应资源规格大小的Executor Pod，所以在提交Spark任务到K8s前就要准确地获取任务真正生效的资源参数。在大数据平台中资源参数会存在多中类型的参数中，参数的优先级为：任务配置参数 < 任务模板参数 < sql中设置参数 < HBO优化参数 < 平台统一参数，按此优先级顺序依次提取最终的资源参数并传入容器接入层创建Spark作业。另外容器接入层对于Spark的arguments和sparkConf参数都是要求以字符数组的方式传入，需要做好对原任务参数中的单引号、双引号、反斜杠和回车等符号以及分段落的处理和转换。
 
2、overheadMemory的计算
在Yarn上Executor是运行在NodeManager节点上的，节点的资源一般都大于并能满足container申请的资源，所以在Yarn上只需要关心container本身申请的资源即可，而在K8s上Executor运行在对应的Pod中，可以把Pod理解为只一台独立的节点，除了要满足container申请的资源量，还需要一些Pod容运行时网络、存储等基础设施的自身开销资源，如果把Spark任务中Driver和Executor申请的资源直接设置为K8s中Driver Pod和Executor Pod的资源规格，有可能出现OOM情况，另外还要考虑非JVM内存，Spark默认会把申请的Executor内存乘以一个系数或者至少预留384 MiB内存作为额外的非JVM内存缓冲区，用于堆外内存分配、非JVM任务以及各类系统进程的使用，可以通过设置overheadMemory进行覆盖。因此K8s的Pod除了要满足申请的Memory和运行时需要的overheadMemory的资源，还会再添加100M资源用于Pod运行的自身开销。
 
pod的资源规格 = memory + pod overheadMemory
 
对于overheadMemory也需要先获取到并加到Pod的资源规格，如果任务有配置就直接使用配置的overheadMemory，如果没有配置值则按一定计算公式来计算得到。
 
有配置：
pod overheadMemory = overheadMemory + 100M
 
无配置：
pod overheadMemory = (max(384M，0.1*memory))向上取整到512MB的整数倍 + 100M
 
不过在实际应用中发现对于个别任务，即使K8s上配置的overheadMemory比在Yarn的配置多100M，完全一样的任务在K8s上则有较多的Executor OOM情况，而在Yarn上却完全没有，目前排查到的现象是有JVM堆外的内存无法回收，如果任务需要较多的对外内存，堆外内存一直增长最终导致OOM，但哪些内存无法回收的还未排查到。目前对于这些OOM过多且实际影响到运行效率的任务，在原overheadMemory基础上再增加512M后就没有OOM情况了，同时也有采用了大数据平台的HBO能力自动调整内存参数来事后规避这个问题。
 
3、CPU超分配置
Spark任务申请的CPU使用一般不会使用完，事实上Executor Pod的CPU利用率也并不是很高，比如Executor申请1个核，通常只能利用0.6个核，存在CPU浪费的现象。Executor Pod的资源规格是创建的时候分配的，利用容器的能力，可以采取CPU超分的方式提高CPU的利用率，例如Executor申请1核，实际用0.6核，如果Pod分配1核，那利用率就只有60%，但如果Pod只分配0.8核，那利用率就有75%了，所以超分的策略就是申请了1核只给0.8核，但还是要按1核的申请量来运行任务。目前平台使用的是静态的固定比例超分设置为0.8，实施超分配置策略后Pod的实际CPU利用率打到80%以上。
 

 
3.3 混部任务的筛选提交

经过上面的任务提交方式的改造和任务资源参数的调整，原SparkSql和SparkJar任务就可以平滑切换提交到混部K8s上执行了，但在大规模切换之前平台还做了比较长期的双跑验证工作，在执行成功率、数据一致性和执行时效等方案都进行了双跑比较，双跑通过的任务才能切换到K8s上执行。除了双跑通过，前期还设置了其他的筛选条件如下。
 

 
前期按这些条件筛选出可以提交到K8s的任务，然后分批的进行K8s任务的参数标记，并把标记的这批任务添加监控进行跟踪。经过双跑验证、任务筛选、批量标记、监控跟踪和问题解决这一整套SparkSql任务上量K8s的流程，K8s上的任务运行逐步稳定，K8s的兼容问题也基本解决，因此目前取消了双跑通过的这一条件，主要保留了任务重要性、运行时长和重试次数这几个筛选指标。随着SparkSql任务上量和稳定，提交到K8s的任务类型也增加了SparkJar任务，SparkJar任务无法进行双跑验证，所以在各种K8s兼容问题解决后再推进会更加稳妥。
 
目前大数据平台会定期筛选和标记一批SparkSql和SparkJar任务允许提交到混部K8s，用户也可以自行开启，在任务配置页面只显示已开启混部，则该任务就有机会被提交到混部K8s上执行。当然，用户也可以手动关闭这一开关，并且手动操作的优先级最高，手动关闭后平台的自动开启功能将不再生效。
 
四、弹性调度系统

4.1 弹性调度功能矩阵

Spark任务开启了混部也不是必定能提交到混部，最终能不能在混部集群上执行，还要根据当时混部集群的资源和运行情况等来确定，为了更好地协调离线任务和混部集群的供需关系，大数据平台构建了离线任务混部弹性调度系统。弹性调度系统的设计目是混部集群有资源了就调度离线任务，但在生产环境中不管是混部集群还是离线任务都会各自的问题需要解决和优化的需求，弹性调度系统也逐步演变成了全面管理离线任务提交到混部以实现混部资源最大化利用的功能矩阵。
 
4.1.1 资源水位线调度

 

 
弹性调度的流程，任务按调度时间以任务流的形式过来，如果任务标记了允许提交到混部，那就会先去查询K8s的各个集群，如果某一个集群资源充足就直接提交到K8s，如果当时没有足够资源就等待资源再判断，这里分为有三类任务，第一类是一直等K8s资源，永不超时，只会提交到K8s；第二类是长时间等待，超时时间在1到5分钟，可以等久一点；第三类是短时等待，超时时间为30-60秒，稍微等一下，如果K8s没有资源就回到Yarn上执行，目前平台标记的任务大部分任务都是第三类短时等待。
 
混部集群提供给离线任务的资源是呈潮汐波动的，使用百分比的水位线方式才能更好地贴合资源的波动情况。混部集群提供的资源是指CPU和内存，但离线任务一般不能百分之百地获取到这部分资源，需要设置一个折算比例也就是水位线来计算出离线任务能使用的真正资源是多少，水位线的设置需要考虑几个因素：
1、混部集群的碎片化率，混部集群中的机器规格和正在运行的业务占用量都是不确定的，但一般大规格的机器多的集群碎片化率较低，所以小规格的机器多的集群的水位线要设置低一点。
 
2、资源动态分配容纳率，对于开启了动态分配的Spark任务，无法提前知道任务所需的资源，需要留有一部分资源用于动态分配的消耗，如果同样的水位线资源规模大的混部集群容纳率会高，所以资源规模小的集群的水位线要设置低一点。
 
3、资源配比的均衡性，不同的集群或者同一集群的不同时间段的CPU和内存配比可能会存在很大的差异，例如Spark任务的CPU和内存的平均比例是1核6G，即1:6，如果有CPU和内存比为1:2的，内存会被用完而CPU有剩余，此时为了内存留有部分余量，水位线要设置低一点。
 
混部资源可用量 = 混部资源提供量 * 资源水位线
 
资源水位线有CPU水位线和内存水位线，设计时以CPU或内存中的最低水位线为准，哪个资源先分配完就停止提交任务，不过在实际生产中大部分混部集群都是受内存限制较多，个别时段CPU比内存多但通过其他的限制手段即使CPU满载对任务影响不大，因此目前只开启了内存资源水位线。以上提到的3点可以当成集群的固有消耗需要保留有一定的余量，为了直观地控制混部资源使用率和引入优先策略，计算方式调整为：
混部资源可用量 = 混部资源提供量 * (1-余量水位线) * 优先水位线
 
余量水位线根据各个集群来调整，一般为0.05，优先水位线的范围可以在0-1之间。优先水位线的作用是对于一些符合优先条件的任务可以优先提交，但是任务调度是一有任务就要调度的流式调度，不能够先集中再挑选优先任务而是先到先得，所以要为优先任务预留一部分资源，例如优先水位线为0.8，混部资源使用到0.8以下的时候任何任务都可以调度上来，但使用量超过了0.8，那只有优先任务能调上来，也就是为优先任务预留了0.2的资源，当然即使资源使用量达到了1，由于余量水位线的存在，实际的使用量为0.95，混部集群仍有资源维持周转。优先水位线是最常用的调整参数，它实质就是控制混部任务提交量，不仅能调整混部资源的使用量，还在灰度测试、压力测试和问题排查等事项起到了灵活调节的作用。
 
4.1.2 其他调度能力

1.多集群管理：混部集群通常会有多个，vivo目前就有多个生产环境的混部集群，各混部集群由于建设周期、机器规格和业务接入的不同，混部资源的规模和变化趋势都会呈现比较大的差异，因此每个集群的调度策略配置都需要做到能独立调整来适应各自的资源特点。
 
2.分时段控制：每个混部集群上的在线业务一般是潮汐波动的，给到离线任务的资源也是潮汐波动的，因此每个集群需要做到在每天不同时段可以调整不同的调度策略，尤其在波峰波谷差异较大的时间段各自调整配置的差异会更大。
 
3.分散namespace：Spark任务的Driver Pod和Executor Pod都会放在一个namespace中管理，如果所有任务都由一个namespace管理，那需要管理的pod数量会达到数十万的级别，会对K8s集群的性能和稳定性产生影响。因此需要将Spark任务平均分配到多个namespace，采用的方案是轮询填充，任务优先分配到多个namespace中任务最少namespace。
 
4.失败回退Yarn：离线任务混部推进的过程中还有会有Spark兼容问题、混部集群异常和平台变更等问题导致的离线任务在混部K8s上运行失败，为了减少失败对任务的影响，任务在K8s上首次执行失败后就会自动回到Yarn重新执行。
 
5.资源准入粒度：各混部集群的机器规格和碎片率是不一样的，如executorMemory=2G这样较小粒度的Spark任务即使碎片率较高的混部集群可以填充，而对于executorMemory=16G这样较大粒度的Spark任务，机器规格大的集群才更容易获取到资源，因此不同混部集群可以设置不同的准入粒度，小规格和碎片率高的集群准入粒度可以设置小一些。
 
6.任务偏好配置：对于一些灰度任务和特殊要求的任务，例如只有在0到8点才允许提交到混部、只提交到某几个指定的混部集群等调度要求，需要支持任务偏好配置，在任务参数中调整混部控制参数实现相应的调度需求。
 
4.2 弹性调度策略优化

弹性调度的核心是通过资源水位线的调节，有混部资源就调度离线任务，但实际生产中还要考虑混部集群的运行情况，是否能稳定地接收和消化离线任务，同时在存在多个差异较大的集群时提交到哪个集群最优。
 
4.2.1 任务调度稳定优化

大数据平台的离线任务提交高峰在凌晨时段而且调度时间集中在整点半点，还有5分和10分这样的整分，例如03:00调度的任务达1000个，但在03:01调度的任务只有10个，过于集中地提交任务会导致混部集群Pending Pod数量急剧上升，这是因为无论是查询集群资源还是Pending数的接口，更新数据都需要一定的周期时间，而且离线任务提交上去到获取资源也受K8s的调度时间的影响，所以获取集群运行情况总会滞后于任务提交。例如03:00查询集群是有资源的并且是健康的，由于任务开启了动态分配所以不能确定需要多少资源，此时集中提交了1000个任务，这1000个任务首先会创建1000个Driver Pod，集群资源还是能满足的并且优先创建，假如每个Driver需要创建100个Executor，如果集群没有这么多资源，那就会产生大量的Penging Pod，严重影响集群的性能和稳定以及任务的执行效率，因此需要对弹性调度的稳定性进行优化。
 
短时提交限制：避免集中提交任务的直接方案就是根据各混部集群的资源规模设置短时提交的任务数量限制，例如1分钟内只能提交100个任务，集群短时间内Pending Pod数量会增加但仍在可以承受范围内，集群和任务都会稳定运行。短时提交限制相当于拦截并舍弃了部分某个时间点集中提交的任务，这里相当于舍弃了900个任务，那么提交的总任务量就减少了。
 
延迟打散提交：为解决短时提交限制导致舍弃部分任务的问题，增加了短时延迟打散提交，例如03:00提交的1000个任务，随机打散到03:00到03:03的3分钟内，即使有短时提交限制，这3分钟内也可以提交300个任务。理论上将集中提交的任务延迟更久，能提交到混部的任务会更多，但是增加延迟时长就等于增加任务的执行时长，会影响到业务数据产出的及时性，因此延迟打散提交策略只能是短时的，进一步的优化是执行时长更久的任务延迟更久一点，但根本解决方案还是用户能将调度时间尽量打散。
 
集群反馈限制：短时提交限制和延迟打散提交都属于静态限制，需要人为地根据各个混部集群的情况去判断和设置限制值，因此需要做到动态限制，就需要获取集群的运行情况并根据运行情况进行限制。事实上K8s的调度性能相比于Yarn还是有差距的，从提交的Spark任务到获取到资源运行Pod有一定的滞后时间差，这段时间查询内还是有剩余资源，但如果还继续提交新任务就会产生更多Pending Pod，因此需要做集群运行情况的反馈控制，例如查询Pending Pod数、等待的SparkApp数，当数量达到一定数量就不再提交新任务。
 
集群反馈限制虽然是动态的能根据混部集群情况进行反馈调节，但是查询集群状态是滞后的，这种滞后的控制就容易被集中提交给打垮，所以要加上短时提交限制来上一道保险，为缓解短时提交限制造成的任务损失，就引入了延迟打散提交，而在延时打散的过程中集群能逐步消化任务，查询集群状态逐步接近真实情况，这时又可以交给集群反馈限制来动态调节，逐步从突增恢复到稳定，三个调度稳定优化策略相辅相成。
 
4.2.2 集群分配均匀优化

离线任务会调度到多个混部集群，每个集群的资源总量和可用资源量，以及集群运行状况都不相同，为保证离线任务的运行稳定和执行效率，需要在多个混部集群中选择一个最合适的集群。各个集群会按一定的规则进行排序，离线任务会按这个排序依次轮询各个集群，只要集群剩余资源满足且没有被短时提交限制、集群反馈限制等拒绝，离线任务就提交到该集群。集群排序的演化顺序如下：
 
①初始方案
排队队列+轮询
剩余资源量多的优先
 

 
优点
离线任务优先提交到资源最多的集群，保证离线任务运行稳定
 
缺点
对于小集群剩余资源量很小一直分配不到任务容易“饿死”（事实上有的小集群全部资源量都达不到一个大集群的20%）
 
② 优化方案
随机队列+排序队列+轮询
将资源使用量超过一定比例的集群放到排序队列，剩余的集群放到随机队列
 

 
优点
离线任务优先提交到资源较多的集群，即保证任务的运行稳定，随机的方式也能均匀“喂饱”每个集群
 
缺点
随机分配在大任务量时相当于是平均分配，每个集群都会调度差不多的任务量，当前情况会存在整点集中提交大量任务，小集群接收和大集群同样任务量会抗不住，影响任务执行稳定和效率，小集群容易“撑死”
 
③再优化方案
加权随机队列+排序队列+轮询
按剩余资源进行加权随机，剩余资源多的集群有更多概率分配到任务
 

 
优点
离线任务优先提交到资源较多的集群，“大集群多吃，小集群少吃”，每个集群都能填充同时保证任务的运行稳定
 
④ 最终方案
优先队列（排序）+加权随机队列+排序队列+轮询
考虑优先队列，无视其他排序规则，优先队列里的集群将最优先，在优先队列中的集群再按资源排序
 

 
优点
继承上一方案的优点，同时对于特定项目或机房的离线任务，能优先调度到某些特定的集群
 
目前只以内存作为资源水位线的衡量标准，这里的资源量指的是内存量。最开始方案是按集群的剩余资源排序，内存资源剩余多的集群优先，缺点是小集群一直分配不到任务容易“饿死”，然后使用随机的方式也能均匀“喂饱”每个集群，但小集群接收同样任务量时容易“撑死”，于是随机队列按剩余资源进行加权随机，剩余资源多的集群有更多概率分配到任务，这样离线任务优先提交到资源较多的集群，“大集群多吃，小集群少吃”，每个集群都能填充同时保证任务的运行稳定，在此基础上增加优先队列，无视其他排序规则，优先队列里的集群将最优先，在优先队列中的集群再按资源排序，能优先调度到某些特定的集群，形成最终集群选择排序方案。
 
五、混部的效果与未来规划

经过以上的对Spark组件、K8s混部系统、大数据平台以及弹性调度系统的改造和优化，目前混部集群及提交混部的离线任务运行持续稳定，每天任务调度到混部的次数达10+万次，在凌晨的高峰期通过混部能为离线任务额外增加数百TB内存的计算资源，部分混部集群的CPU利用率提升至30%左右，整体收益也是可观的。
 
虽然目前vivo的在离线混部达到了一定的规模，但未来要继续提高混部的规模和收益，还有规划一些改进工作。
 
1、提高离线任务混部规模。

离线任务混部的节点是在线业务提供的，节点规模取决于在线业务峰值，峰值越高那么在业务低峰期能提供给离线混部资源就越多，因此提高混部规模的重要因素是提交更多的离线任务。然而目前采用的Spark Operator方案能提交的离线任务只有标准的SparkSql和SparkJar任务，而对于非标准的任务如脚本任务，脚本中除了调用spark-submit提交Spark作业还有额外的处理逻辑，这类任务还不能直接以Spark Operator的方式提交。事实上Spark作业更多是来自脚本任务的非标准任务，如果要继续增加离线任务的量，就必须把非标准任务也提交到混部，因此后续是选择改造spark-submit客户端支持Spark Operator，或是选择使用Yarn on K8s，还需要综合评估。
 
2、提高离线任务混部收益。

目前混部节点CPU的平均利用率达到30%，但仍有提升空间。从离线任务的角度来看，一方面是要增加错峰互补的时间段，例如离线任务的高峰期是02:00到08:00，在线业务的高峰期是06:00到23:00，在06:00后在线业务逐步上量开始回收资源，所以离线任务能显著提高混部集群CPU利用率的黄金时间是有02:00到06:00这4个小时，因此如果能把离线任务高峰期提前到00:00到06:00，混部提效的黄金时间就能达到6小时。所以需要推动离线任务高峰期的前移，对于有依赖链路的任务，尽量减少调度时间的间隔，上游任务完成后能尽快调起下游任务，而对于没有依赖的任务，可以尽量提前调度时间，不过这两种调整都需要推动业务方来调整，平台也可以给予一定的计算成本优惠作为激励。另一方面是要提高混部资源的填充率，Spark任务需要创建大量的Executor Pod，目前混部集群的调度器为了保证调度效率就没有开启预选、优先策略，事实上Spark的资源粒度比较小更适合填充资源碎片，所以在不影响K8s调度效率的情况下优化资源调配策略，把合适的资源粒度的Pod分配到合适的混部节点，也是提高混部收益的方向。