如何实现Spark on Kubernetes？

简介： 大数据时代，以Oracle为表明的数据库中间件已经逐渐没法适应企业数字化转型的需求，Spark将会是比较好的大数据批处理引擎。而随着Kubernetes愈来愈火，不少数字化企业已经把在线业务搬到了Kubernetes之上，并但愿在此之上建设一套统一的、完整的大数据基础架构。那么Spark on Kubernetes面临哪些挑战？又该如何解决？

云原生背景介绍与思考

“数据湖”正在被愈来愈多人提起，尽管定义并不统一，但企业已纷纷投入实践，不管是在云上自建仍是使用云产品。

阿里云大数据团队认为：数据湖是大数据和AI时代融合存储和计算的全新体系。为何这么说？在数据量爆发式增加的今天，数字化转型成为IT行业的热点，数据须要更深度的价值挖掘，所以确保数据中保留的原始信息不丢失，应对将来不断变化的需求。当前以Oracle为表明的数据库中间件已经逐渐没法适应这样的需求，因而业界也不断地产生新的计算引擎，以便应对大数据时代的到来。企业开始纷纷自建开源Hadoop数据湖架构，原始数据统一存放在对象存储OSS或HDFS系统上，引擎以Hadoop和Spark开源生态为主，存储和计算一体。

图1是基于ECS底座的EMR架构，这是一套很是完整的开源大数据生态，也是近10年来每一个数字化企业必不可少的开源大数据解决方案。

主要分为如下几层：

• ECS物理资源层，也就是Iaas层。
• 数据接入层，例如实时的Kafka，离线的Sqoop。
• 存储层，包括HDFS和OSS，以及EMR自研的缓存加速JindoFS。
• 计算引擎层，包括熟知的Spark，Presto、Flink等这些计算引擎。
• 数据应用层，如阿里自研的Dataworks、PAI以及开源的Zeppelin，Jupyter。

每一层都有比较多的开源组件与之对应，这些层级组成了最经典的大数据解决方案，也就是EMR的架构。咱们对此有如下思考：

• 是否可以作到更好用的弹性，也就是客户能够彻底按照本身业务实际的峰值和低谷进行弹性扩容和缩容，保证速度足够快，资源足够充足。
• 不考虑现有情况，看将来几年的发展方向，是否还须要支持全部的计算引擎和存储引擎。这个问题也很是实际，一方面是客户是否有能力维护这么多的引擎，另外一方面是某些场景下是否用一种通用的引擎便可解决全部问题。举个例子来讲，Hive和Mapreduce，诚然现有的一些客户还在用Hive on Mapreduce，并且规模也确实不小，可是将来Spark会是一个很好的替代品。
• 存储与计算分离架构，这是公认的将来大方向，存算分离提供了独立的扩展性，客户能够作到数据入湖，计算引擎按需扩容，这样的解耦方式会获得更高的性价比。

基于以上这些思考，咱们考虑一下云原生的这个概念，云原生比较有前景的实现就是Kubernetes，因此有时候咱们一提到云原生，几乎就等价因而Kubernetes。随着Kubernetes的概念愈来愈火，客户也对该技术充满了兴趣，不少客户已经把在线的业务搬到了Kubernetes之上，而且但愿在这种相似操做系统上，建设一套统一的、完整的大数据基础架构。因此咱们总结为如下几个特征：

但愿可以基于Kubernetes来包容在线服务、大数据、AI等基础架构，作到运维体系统一化。

存算分离架构，这个是大数据引擎能够在Kubernetes部署的前提，将来的趋势也都在向这个方向走。

经过Kubernetes的天生隔离特性，更好的实现离线与在线混部，达到降本增效目的。

Kubernetes生态提供了很是丰富的工具，咱们能够省去不少时间搞基础运维工做，从而能够专心来作业务。

EMR计算引擎 on ACK

图2是EMR计算引擎 on ACK的架构。ACK就是阿里云版本的Kubernetes，在兼容社区版本的API同时，作了大量的优化。在本文中不会区分ACK和Kubernetes这两个词，能够认为表明同一个概念。

基于最开始的讨论，咱们认为比较有但愿的大数据批处理引擎是Spark和Presto，固然咱们也会随着版本迭代逐步的加入一些比较有前景的引擎。

EMR计算引擎提供以Kubernetes为底座的产品形态，本质上来讲是基于CRD+Operator的组合，这也是云原生最基本的哲学。咱们针对组件进行分类，分为service组件和批处理组件，好比Hive Metastore就是service组件，Spark就是批处理组件。

图中绿色部分是各类Operator，技术层面在开源的基础上进行了多方面的改进，产品层面针对ACK底座进行了各方面的兼容，可以保证用户在集群中很方便的进行管控操做。右边的部分，包括Log、监控、数据开发、ECM管控等组件，这里主要是集成了阿里云的一些基础设施。咱们再来看下边的部分：

• 引入了JindoFS做为OSS缓存加速层，作计算与存储分离的架构。
• 打通了现有EMR集群的HDFS，方便客户利用已有的EMR集群数据。
• 引入Shuffle Service来作Shuffle 数据的解耦，这也是EMR容器化区别于开源方案的比较大的亮点，以后会重点讲到。

这里明确一下，因为自己Presto是无状态的MPP架构，在ACK中部署是相对容易的事情，本文主要讨论Spark on ACK的解决方案。

Spark on Kubernetes的挑战

总体看，Spark on Kubernetes面临如下问题：

• 我我的认为最重要的，就是Shuffle的流程，按照目前的Shuffle方式，咱们是没办法打开动态资源特性的。并且还须要挂载云盘，云盘面临着Shuffle数据量的问题，挂的比较大会很浪费，挂的比较小又支持不了Shuffle Heavy的任务。
• 调度和队列管理问题，调度性能的衡量指标是，要确保当大量做业同时启动时，不该该有性能瓶颈。做业队列这一律念对于大数据领域的同窗应该很是熟悉，他提供了一种管理资源的视图，有助于咱们在队列之间控制资源和共享资源。
• 读写数据湖相比较HDFS，在大量的Rename，List等场景下性能会有所降低，同时OSS带宽也是一个不可避免的问题。

针对以上问题，咱们分别来看下解决方案。

Spark on Kubernetes的解决方案

Remote Shuffle Service架构

Spark Shuffle的问题，咱们设计了Shuffle 读写分离架构，称为Remote Shuffle Service。首先探讨一下为何Kubernetes不但愿挂盘呢，咱们看一下可能的选项：

• 若是用是Docker的文件系统，问题是显而易见的，由于性能慢不说，容量也是极其有限，对于Shuffle过程是十分不友好的。
• 若是用Hostpath，熟悉Spark的同窗应该知道，是不可以启动动态资源特性的，这个对于Spark资源是一个很大的浪费，并且若是考虑到后续迁移到Serverless K8s，那么从架构上自己就是不支持Hostpath的。
• 若是是Executor挂云盘的PV，一样道理，也是不支持动态资源的，并且须要提早知道每一个Executor的Shuffle数据量，挂的大比较浪费空间，挂的小Shuffle数据又不必定可以容纳下。

因此Remote Shuffle架构针对这一痛点、对现有的Shuffle机制有比较大的优化，图3中间有很是多的控制流，咱们不作具体的讨论。主要来看数据流，对于Executor全部的Mapper和Reducer，也就是图中的蓝色部分是跑在了K8s容器里，中间的架构是Remote Shuffle Service，蓝色部分的Mapper将Shuffle数据远程写入service里边，消除了Executor的Task对于本地磁盘的依赖。Shuffle Service会对来自不一样Mapper的同一partition的数据进行merge操做，而后写入到分布式文件系统中。等到Reduce阶段，Reducer经过读取顺序的文件，能够很好地提高性能。这套系统最主要的实现难点就是控制流的设计，还有各方面的容错，数据去重，元数据管理等等工做。

简而言之，咱们总结为如下3点：

• Shuffle数据经过网络写出，中间数据计算与存储分离架构
• DFS 2副本，消除Fetch Failed引发的重算，Shuffle Heavy做业更加稳定
• Reduce阶段顺序读磁盘，避免现有版本的随机IO，大幅提高性能

Remote Shuffle Service性能

咱们在这里展现一下关于性能的成绩，图4和图5是Terasort Benchmark。之因此选取Terasrot这种workload来测试，是由于它只有3个stage，并且是一个大Shuffle的任务，你们能够很是有体感的看出关于Shuffle性能的变化。

图4中，蓝色部分是Shuffle Service版本的运行时间，橙色部分是原版Shuffle的运行时间。咱们测试了2T，4T，10T的数据，能够观察到随着数据量愈来愈大，Shuffle Service优点就愈来愈明显。图5红框部分说明了做业的性能提高主要体如今Reduce阶段，可见10T的Reduce Read从1.6小时降低到了1小时。缘由前边已经解释的很清楚了，熟悉Spark shuffle机制的同窗知道，原版的sort shuffle是M*N次的随机IO，在这个例子中，M是12000，N是8000，而Remote Shuffle就只有N次顺序IO，这个例子中是8000次，因此这是提高性能的根本所在。

图4 Remote Shuffle Service性能

图5 Terasort做业的stage图

其余方面的重点优化

这里讲一下EMR在其余方面作的优化。

调度性能优化，咱们解决了开源的Spark Operator的一些不足，对于Executor pod的不少配置Spark Operator把他作到了Webhook里边，这个对调度来讲是十分不友好的，由于至关于在API Server上绕了一圈，实际测试下来性能损耗很大。咱们把这些工做直接作到Spark内核里边，这样避免了这方面的调度性能瓶颈。而后从调度器层面上，咱们保留了两种选择给客户，包括ACK提供的Scheduler FrameworkV2实现方式和Yunicorn实现方式。

读写OSS性能优化，咱们这里引入了JindoFS做为缓存，解决带宽问题，同时EMR为OSS场景提供了Jindo Job Committer，专门优化了job commit的过程，大大减小了Rename和List等耗时操做。

针对Spark自己，EMR积累了多年的技术优点，也在TPCDS官方测试中，取得了很好的成绩，包括Spark性能、稳定性，还有Delta lake的优化也都有集成进来。

咱们提供了一站式的管控，包括Notebook做业开发，监控日志报警等服务，还继承了NameSpace的ResourceQuota等等。

整体来讲，EMR版本的Spark on ACK，不管在架构上、性能上、稳定性上、易用性方面都有着很大的提高。

Spark云原生后续展望

从个人视角来观察，Spark云原生容器化后续的方向，一方面是达到运维一体化，另外一方面主要但愿作到更好的性价比。咱们总结为如下几点：

• 能够将Kubernetes计算资源分为固定集群和Serverless集群的混合架构，固定集群主要是一些包年包月、资源使用率很高的集群，Serverless是作到按需弹性。
• 能够经过调度算法，灵活的把一些SLA不高的任务调度到Spot实例上，就是支持抢占式ECI容器，这样能够进一步下降成本。
• 因为提供了Remote Shuffle Service集群，充分让Spark在架构上解耦本地盘，只要Executor空闲就能够销毁。配合上OSS提供的存算分离，一定是后续的主流方向。
• 对于调度能力，这方面须要特别的加强，作到可以让客户感觉不到性能瓶颈，短期内调度起来大量的做业。同时对于做业队列管理方面，但愿作到更强的资源控制和资源共享。

随着阿里云2.0时代的到来，云原生已经逐渐成为新的主角，愈来愈多的企业将会享受到云原生所带来的红利。数据湖计算引擎云原生容器化，可以极大地方便客户上云，提高性价比。可是总体上来说：大数据云原生的落地是十分具备挑战性的，阿里云EMR团队会和社区同伴、合做伙伴一块儿打造技术和生态，共同打造“存算分离，按需扩展”、“极致弹性，随用随得”、“运维闭环，高性价比”的愿景。