图片来源:pexels
背景
Firestorm
Shuffle是分布式计算框架用来衔接上下游任务的数据重分布过程,在分布式计算中所有涉及到数据上下游衔接的过程都可以理解为shuffle。针对不同的分布式框架,shuffle有几种实现形态:
Shuffle是分布式框架里面最为重要的一个环节,shuffle的性能和稳定性直接影响到了整个框架的性能和稳定性,因此改进shuffle框架是非常有必要的。
业务痛点
Spark在云原生场景下的挑战
基于本地磁盘的shuffle方式,使得Spark在云原生、存储计算分离、在离线环境中有极大的使用限制:
Spark在生产环境的挑战
当前分布式计算平台上大多数的批处理作业是Spark作业,少量是MR作业,相比于MR作业,Spark作业的稳定性较差,而稳定性的问题中至少有一半是由于shuffle的失败造成的。
Shuffle失败导致的任务陷入重试,严重拖慢作业。shuffle fetch失败会导致map任务重跑重新生成shuffle数据,然后再重跑reduce任务,如果reduce任务反复失败会导致map任务需要反复重跑,在集群压力较高的情况下重跑的代价很高,会严重影响作业。
邵铮在SPARK-1529中就有相应的评论,地址如下:
https://issues.apache.org/jira/browse/SPARK-1529
对于超大规模的shuffle数据(T级别以上的shuffle量)的作业,非常难以顺利跑过,这里面的问题有:
基于本地磁盘的shuffle方式有较为严重的写放大问题和随机IO问题,当任务数量达到10K乃至100K以上时,随机IO的问题非常严重,严重影响了集群的性能和稳定性。
因此实现一个更好的、能解决上述业务痛点的shuffle框架显得尤为重要。
业界趋势
业界在shuffle[1]上也有了多年的探索,围绕各自的业务场景构建了相应的能力,这里罗列一下主流公司在shuffle上所做的工作。
百度DCE shuffle
百度DCE shuffle是较早在业内实践并大规模使用的remote shuffle service方案,它设计的初衷是为了解决几个问题,一是在离线混部,二是提高MR作业的稳定性和处理规模。百度内部的MR作业已经改造接入DCE shuffle并使用多年,现在Spark批处理作业也已经改造使用DCE shuffle做为其shuffle引擎。
Facebook Cosco Shuffle[2]
Facebook Cosco Shuffle的设计初衷和百度非常接近,Facebook数据中心的构建是存储计算分离,因此传统的基于本地文件的shuffle方式有较大的开销,同时在Facebook中最大的作业规模shuffle量达100T,这对shuffle有极大的挑战,因此Facebook实现了基于HDFS的remote shuffle service - Cosco Shuffle。
Google Dataflow Shuffle[3]
Google Dataflow Shuffle是Google在Google Cloud上的Shuffle服务,针对云上的弹性易失环境,Google开发了一套Dataflow Shuffle服务供Google Cloud的大数据服务使用。Dataflow Shuffle也是一套remote shuffle service,将shuffle存储移到了VM之外,提供了计算作业更大的弹性。
Uber Zeus[4]
Uber 为了解决上述提到的Shuffle痛点,也实现了Zeus这个Remote Shuffle Service,该项目已经开源。从设计文档及实现看,他们部署了多台Shuffle Server用来接收并聚合Shuffle数据,采用SSD作为存储介质来提升Shuffle性能。
阿里ESS[5]
阿里的ESS(EMR Remote Shuffle Service)主要是为了解决Spark on Kubernetes面临的计算存储分离问题,使得Spark能够适配云原生环境。
业务价值
实现Remote Shuffle Service,能带来几点业务价值:
Firestorm介绍
Firestorm
目标
在腾讯内部每天有上百万的Spark任务在运行,上述各类Shuffle问题也经常遇到。同时,为了更好的利用硬件资源,计算存储分离的部署模式也在逐步推进。因此,我们进行了Firestorm的研发,该项目的目标如下:
架构设计方案
Remote Shuffle Service的架构如下:
其中,各个组件的功能如下:
架构设计差异点
相较于业界的其它方案,Firestorm有其独特的地方:
整体流程
基于Firestorm的整体Shuffle流程如下:
写流程
在写Shuffle数据的时候,需要考虑内存合理使用,文件的异步写入,Shuffle数据的合并等,具体流程如下:
读流程
在读Shuffle数据的时候,主要考虑数据的完整性,具体流程如下:
Shuffle文件
对于Shuffle数据,存储为Index文件和Data文件,其中实际的Shuffle数据以Block形式存储在Data文件中,而Index文件则存储每个Block的元数据,具体存储信息如下:
数据校验
数据的正确性对于Shuffle过程来说是最为关键的,下面介绍了Firestorm如何保障数据的正确性:
支持多存储
由于存储的选择较多,LocalFile,HDFS,OZONE,COS等,为了能方便接入各类存储,在设计上对存储做了解耦,抽象了读写接口。对于不同的存储只需要实现相关接口,即可作为Shuffle数据的后端存储使用。
Firestorm 收益
支撑云原生的部署模式
Firestorm目前在腾讯内部已经落地于近万个节点的在离线混布集群,每天支撑近5W的分布式计算作业,每天的Shuffle数据量接近2PB,任务失败率从原先的14%降低到了9%,已经达到了初期制定的第一阶段目标,助力分布式计算上云。
改善Shuffle阶段的稳定性及性能
基于TPC-DS 1TB数据量,我们对使用原生Spark Shuffle 和 使用Firestorm进行了性能对比测试,测试环境如下:
TPC-DS的SQL复杂程度不一,对于简单的SQL,由于Shuffle数据量较少,原生Spark Shuffle表现更好,但是性能优势并不明显,而对于复杂的SQL,涉及到大量的partition的Shuffle过程,则Firestorm表现更稳定,且性能有大幅提升,下面将分别描述这2种场景:
场景1,简单SQL,以query43为例,下图是query43的Stage图,由2个stage组成,shuffle数据量非常少,使用原生Spark Shuffle整个query运行耗时12秒左右,而使用Remote Shuffle Service则需要15秒左右。
那这个时间是损耗在哪里了呢?下图展示了第一个stage的相关耗时,可以看到在write time这一列的统计上,原生Spark Shuffle是有性能优势的,耗时都在毫秒级别,而使用Firestorm,由于在shuffle write阶段增加了RPC通信,导致耗时增加,再加上任务数量需要分多批次跑完,每批次都会产生几百毫秒的差值,最终造成了原生Spark Shuffle在这个query上有了3秒左右的性能优势。
随着SQL的执行时间增加,这类性能优势会逐步下降,几乎可以忽略不计,这一类的SQL有query1, query3等等,这里就不一一列举了。
场景2,复杂SQL,以query17为例,下图展示了分别使用不同shuffle模式的Stage图,从图中可以看到这个SQL的stage数量多,且shuffle数据量大,执行的耗时上使用原生Spark Shuffle为8分钟左右,而使用Remote Shuffle Service仅为3分钟左右。
将耗时最长的Stage展开,进一步看下具体的耗时比对,先看下Shuffle Read的耗时,由于原生Spark Shuffle需要从各个Executor上拉取数据,涉及到大量的网络开销以及磁盘的随机IO,耗时非常长,甚至达到了2分钟,而Remote Shuffle Service由于读取时降低了网络开销,且读取的是整块Shuffle数据,所以耗时短且较为稳定。
再来看下Shuffle Write的耗时,原生Spark Shuffle依然耗时长,且不稳定,这个主要是由于这个时间点,计算节点同时处理Shuffle Read和Shuffle Write,都需要对本地磁盘频繁访问,且数据量较大,最终导致了耗时大幅增长,而Remote Shuffle Service在读写机制上很好的规避了这类问题,所以整体性能有了大幅提升且更稳定。
这类SQL还有query25,query29等待,这里就不一一举例了。
除了上述2个场景外,还有部分query由于shuffle数据量更大,使用原生Spark Shuffle不能正常跑出结果,而使用Remote Shuffle Service则可以顺利运行,如,query64,query67。
总的来说,在Shuffle数据量较小的场景下,相比原生Spark Shuffle,Remote Shuffle Service并无优势,性能有5%-10%的小幅下降或基本持平,而在Shuffle数据量大的场景下,Remote Shuffle Service则优势明显,基于TPC-DS的部分SQL测试结果显示,性能有50% - 100%的提升。
总结
Firestorm
本文介绍了现有Spark Shuffle实现的各类问题及业界的应对方式,结合腾讯公司内部的Spark任务实际运行状况,介绍了我们自研的Firestorm的架构,设计,性能,应用等。希望在云原生的场景中,Firestorm能更好的协助分布式计算引擎上云。
开源版本地址:
https://github.com/Tencent/Firestorm
欢迎大家关注和star,同时也欢迎各位优秀的开发者加入腾讯大数据研发团队。
附录
[1]https://issues.apache.org/jira/browse/SPARK-25299
[2]https://www.slideshare.net/databricks/cosco-an-efficient-facebookscale-shuffle-service
[3]https://cloud.google.com/dataflow/docs/guides/deploying-a-pipeline#cloud-dataflow-shuffle
[4]https://github.com/uber/RemoteShuffleService
[5]https://developer.aliyun.com/article/772328
[6]https://www.sohu.com/a/447193430_315839