您的位置:  首页 > 技术杂谈 > 正文

Firestorm - 腾讯自研Remote Shuffle Service在Spark云原生场景的实践

2021-11-19 15:00 https://my.oschina.net/u/4956788/blog/5312813 腾讯大数据官方 次阅读 条评论

图片来源:pexels

背景

Shuffle是分布式计算框架用来衔接上下游任务的数据重分布过程,在分布式计算中所有涉及到数据上下游衔接的过程都可以理解为shuffle。针对不同的分布式框架,shuffle有几种实现形态:

  1. 基于文件的pull based shuffle,如MapReduce、Spark。这种shuffle方式多用于类MR的框架,比如MapReduce、Spark,它的特点是具有较高的容错性,适合较大规模的批处理作业。由于实现的是基于文件的shuffle方案,因此失败重跑时只须重跑失败的task、stage,而无须重跑整个job。
  2. 基于管道的push based shuffle,比如Flink、Storm等。基于管道的push based shuffle的实现方式多用于Flink、Storm等流式框架,或是一些MPP框架,如Presto、Greenplum等,它的特点是具有较低的延迟和较高的性能,但是比较大的问题是由于没有将shuffle数据持久化下来,因此任务的失败会导致整个作业的重跑。

Shuffle是分布式框架里面最为重要的一个环节,shuffle的性能和稳定性直接影响到了整个框架的性能和稳定性,因此改进shuffle框架是非常有必要的。

业务痛点

Spark在云原生场景下的挑战

基于本地磁盘的shuffle方式,使得Spark在云原生、存储计算分离、在离线环境中有极大的使用限制:

  1. 在云原生环境中,serverless化是服务部署的一个目标,但是由于弹性或是抢占情况的发生,节点或是容器被抢占导致executor被kill是一种常态,现有的shuffle无法使计算做到serverless,在节点/容器被抢占时往往需要重新计算shuffle数据,有很高的代价。
  2. 在线集群通常只有少量的本地磁盘和大量的CPU core,因此其计算和IO是不平衡的,在这样的集群中根据算力去调度作业时非常容易将磁盘写满。
  3. 现在越来越多的数据中心架构采用了存储计算分离的部署方式,在这样的部署方式下基于本地磁盘的shuffle方式首先会遇到的问题是由于本地磁盘的不足导致无法存放shuffle数据;其次,虽然可以通过块存储(RBD)的方式来解决本地存储,但是对于shuffle这样的IO使用模式,使用块存储会带来极大的网络开销和性能问题。

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量)的作业,非常难以顺利跑过,这里面的问题有:

  1. shuffle数据非常容易将磁盘写满。只有通过反复调整和重试使executor尽量分布到多的节点(anti-affinity)上避免这个问题。
  2. 海量的shuffle partition导致非常多的shuffle连接,使得shuffle框架极容易发生超时问题,以及非常高的随机访问IO所导致的问题。

基于本地磁盘的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,能带来几点业务价值:

  • 云原生架构的支持:现有的分布式计算框架(如Spark需要依赖本地磁盘存储Shuffle数据)极大地限制了云原生的部署模式。使用Remote Shuffle Service可以有效减少对本地磁盘的部分依赖,支持集群的多种部署模式,提升资源利用率,助力云原生架构。
  • 提升Spark作业shuffle稳定性:对于shuffle数据量达到TB,甚至10TB级别,这类任务会对磁盘空间造成很大压力,同时,Task数据较多还造成了网络访问的压力,最终导致失败率较高,而Remote Shuffle Service能更好的解决这些问题,使得业务方能平稳运行这类任务。

Firestorm介绍

目标

在腾讯内部每天有上百万的Spark任务在运行,上述各类Shuffle问题也经常遇到。同时,为了更好的利用硬件资源,计算存储分离的部署模式也在逐步推进。因此,我们进行了Firestorm的研发,该项目的目标如下:

  • 支持大Shuffle量的任务(如,TeraSort 40T+)
  • 支持云原生的部署模式(如,计算存储分离部署模式)
  • 支持多种存储系统(LocalFile, HDFS, COS等)
  • 支持数据完整性校验
  • 性能与计算引擎的原生方案接近

架构设计方案

Remote Shuffle Service的架构如下:

其中,各个组件的功能如下:

  • Coordinator,基于心跳机制管理Shuffle Server,存储Shuffle Server的资源使用等元数据信息,还承担任务分配职责,根据Shuffle Server的负载,分配合适的Shuffle Server给Spark应用处理不同的Partition数据。
  • Shuffle Server,主要负责接收Shuffle数据,聚合后再写入存储中,基于不同的存储方式,还能用来读取Shuffle数据(如LocalFile存储模式)。
  • Shuffle Client,主要负责和Coordinator和Shuffle Server通讯,发送Shuffle数据的读写请求,保持应用和Coordinator的心跳等。
  • 在Shuffle Server和Storage交互过程中,解耦了Storage Handler组件,基于这个组件可以灵活接入不同的存储,满足各种存储需求。

架构设计差异点

相较于业界的其它方案,Firestorm有其独特的地方:

  • 架构方面,引入Coordinator组件,更好的管理Shuffle Server并基于Shuffle Server的节点状态合理分配Shuffle任务,集群本身也支持灵活的横向扩展以满足生产需要
  • 技术方面,解耦了存储模块,支持新的Shuffle数据存储方式仅需要实现相关接口。作为整个系统最重要的数据校验部分,除了CRC,数据去重等实现外,还增加了读写数据一致性校验,使得数据在传输过程中更安全可靠。
  • 运营方面,Firestorm提供了各类运行统计数据并接入内部监控平台,便于观察集群整体状况,了解性能瓶颈,并能在异常状况下及时收到告警信息。

整体流程

基于Firestorm的整体Shuffle流程如下:

  1. Driver从Coordinator获取分配信息
  2. Driver向Shuffle Server注册Shuffle信息
  3. 基于分配信息,Executor将Shuffle数据以Block的形式发送到Shuffle Server
  4. Shuffle Server将数据写入存储
  5. 写任务结束后,Executor向Drive更新结果
  6. 读任务从Driver侧获取成功的写Task信息
  7. 读任务从Shuffle Server获得Shuffle元数据(如,所有blockId)
  8. 基于存储模式,读任务从存储侧读取Shuffle数据

写流程

在写Shuffle数据的时候,需要考虑内存合理使用,文件的异步写入,Shuffle数据的合并等,具体流程如下:

  1. Task基于PartitionId,将数据发送到对应Buffer中
  2. 当Buffer到达阀值时,将Buffer 的数据发送到数据队列里
  3. 不断的从数据队列里获取数据,并提交给发送线程
  4. 发送线程先向Shuffle Server请求内存空间,再将数据发送至Shuffle Server的缓冲区。
  5. Shuffle Server缓冲区到达阀值后,将Shuffle数据发送至写队列
  6. 不断从写队列中获取数据,并提交给写入线程
  7. 基于Shuffle数据信息(ApplicationId, ShuffleId, PartitionId)获取存储路径,将Shuffle数据写入Index文件和Data文件中
  8. Task写入完成后,告知Shuffle Server任务已完成并获取当前所有任务完成数,假如任务完成数小于预期值,则进入下一步,假如任务完成数大于预期值,则发送信息给Shuffle Server将缓冲区相关信息写入存储,并等待写入结果,成功后进入下一步
  9. Task完成后,将TaskId记录在MapStatus中,并发送到Driver,该步骤用来支持Spark推测执行功能

读流程

在读Shuffle数据的时候,主要考虑数据的完整性,具体流程如下:

  1. 从Driver侧获取Write阶段所有成功TaskId
  2. 读取shuffle数据,先读取Index文件,校验BlockId是否都存在,基于Index文件Offset信息,再读取Data文件,获取shuffle数据
    · 如果Storage是HDFS,则直接从HDFS读取
    · 如果Storage是Local File,则需要通过Shuffle Server读取文件

Shuffle文件

对于Shuffle数据,存储为Index文件和Data文件,其中实际的Shuffle数据以Block形式存储在Data文件中,而Index文件则存储每个Block的元数据,具体存储信息如下:

  • BlockId: 每个Block的唯一标识,long型,前19位为自增Int,中间20位为PartitionId,最后24位为TaskId
  • Offset: Block在Data文件里的偏移量
  • Crc: Block的Crc校验值,该值在Block生成时计算并最终存储在Index文件,读取Block时用来验证数据完整性
  • CompressLength: Block压缩后的数据长度
  • UnCompressLength: Block未压缩的数据长度,用来提高读取时解压效率
  • TaskId: 用来过滤无效的block数据

数据校验

数据的正确性对于Shuffle过程来说是最为关键的,下面介绍了Firestorm如何保障数据的正确性:

  1. 写任务为每个Block数据计算CRC校验值,读任务会对每个Block基于CRC进行校验,避免数据不一致
  2. 每个BlockId存储在Shuffle Server侧,读取数据的时候,将验证所有BlockId都被处理,避免数据丢失
  3. 成功的Task信息将记录在Driver侧,读取时过滤冗余的Block,避免推测执行导致的数据不一致问题

支持多存储

由于存储的选择较多,LocalFile,HDFS,OZONE,COS等,为了能方便接入各类存储,在设计上对存储做了解耦,抽象了读写接口。对于不同的存储只需要实现相关接口,即可作为Shuffle数据的后端存储使用。

Firestorm 收益

支撑云原生的部署模式

Firestorm目前在腾讯内部已经落地于近万个节点的在离线混布集群,每天支撑近5W的分布式计算作业,每天的Shuffle数据量接近2PB,任务失败率从原先的14%降低到了9%,已经达到了初期制定的第一阶段目标,助力分布式计算上云。

改善Shuffle阶段的稳定性及性能

基于TPC-DS 1TB数据量,我们对使用原生Spark Shuffle 和 使用Firestorm进行了性能对比测试,测试环境如下:

  • 3台服务器作为计算节点,80 core + 256G + HDD
  • 3台服务器作为ShuffleServer,112core + 128G + HDD存储Shuffle数据

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%的提升。

总结

本文介绍了现有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_31

  • 0
    感动
  • 0
    路过
  • 0
    高兴
  • 0
    难过
  • 0
    搞笑
  • 0
    无聊
  • 0
    愤怒
  • 0
    同情
热度排行
友情链接