Firestorm 0.2.0发布:首个支持混合存储的开源Remote Shuffle Service

原创
2022/02/11 19:20
阅读数 2.4K

一、背景

Firestorm自2021年11月上线开源 0.1.0 版本后,该项目受到了业界的广泛关注。 Firestorm是为了加速分布式计算引擎能上云的重要组件,同时也能解决在大Shuffle场景下,计算任务由于Shuffle过程异常而导致的任务失败。(更详细的背景可以参考此文:[ Firestorm - 腾讯自研Remote Shuffle Service在Spark云原生场景的实践]

目前Firestorm迎来了0.2.0 版本的正式发布,而Firestorm也成为了第一个支持混合存储的开源Remote Shuffle Service 方案。本文将重点介绍 Firestorm 0.2.0 版本的最新特性及性能分析。

二.版本新特性—支持混合存储

什么是混合存储

在Firestorm初始版本中,Shuffle数据只能存储在Shuffle Server的本地盘,或者分布式存储系统。而混合存储则充分利用了Shuffle Server的内存资源,并结合本地文件和分布式存储系统,使得Shuffle数据能存储在多个介质中。

为什么需要混合存储

在实际的生产过程中,由于Shuffle数据的块大小不一致,小的只有几KB,甚至几十Byte,**而大的能达到256MB以上。这样的场景下,对于HDFS这样的分布式存储非常不友好,大量的小数据块的写入会导致集群响应过慢,严重影响计算任务的效率。**虽然使用Shuffle Server

磁盘能很好的缓解该问题,但随之而来的问题是,Shuffle Server必须具备大量的磁盘空间来承载PB级别的Shuffle数据,这样的绑定不利于现在的云原生的大环境。同时,在Shuffle数据写入过程中,必须要等待数据都写入存储后,才能进行下一步,在存储繁忙时,对于计算任务的性能有较大的影响。为了解决上述提到的问题,基于内存,本地文件和分布式存储相结合的混合存储的方案就油然而生了。

混合存储实现原理

以Spark为例,先看下基于单一存储的方案是如何对Shuffle数据进行读写的:

在上图写的过程中,Shuffle数据在经过步骤1,2,3的计算,缓存等操作,在步骤4发送到了Shuffle Server侧,再经过步骤5,6的缓存,数据聚合等操作,最终通过步骤7写入存储介质。所有任务结束后,会发送Commit命令给Shuffle Server,如果是最后一个任务,则必须等待相关数据都写入存储后,才能完成,而Commit操作后等待写入存储的过程对于任务的整体性能影响较大。

在写入完成后,读取过程则较为简单,基于存储介质,选择从Shuffle Server读取或直接从分布式存储读取。 了解完之前的方案后,再来看下混合存储是如何实现:

相比之前,有3个主要的变化:

1.首先,步骤5的Flush方案进行了优化: 之前的Flush方案是,每个Shuffle Partition数据达到阈值或整个缓存空间达到阈值,则将这部分数据写入存储介质,而现在则设置了缓存空间的上下水位,到达上水位则进行Flush操作,直至缓存空间到达低水位。同时,在Flush数据选择上,优先选择数据量多的Partition。通过上水位的控制,保证了在Flush过程中,缓存依然有足够的空间接受新的数据,而通过下水位及Flush数据的选择,保障了数据量较少的Shuffle数据能驻留在内存,降低存储写入小文件的概率。

2.其次,对步骤7进行了重构: 支持基于写入数据块大小对存储介质进行选择,如,大于32MB的数据块写入分布式存储,而其它的则写入本地存储。这样的策略是为了更好匹配分布式存储的写入模式,达到更好的写入性能。同时,也观察到在实际任务运行过程中,大数据块的数量虽然占比不高,如,30%,但是,大数据块的数据总量占比更高,如,70%。基于这样的存储方案,可以降低对于本地盘容量的依赖,便于Firestorm在各种环境下进行部署,甚至云上部署。

3.最后,去除了步骤8的Commit操作: Commit操作存在的意义在于读取数据时保证数据都能被读取到。由于内存也作为了混合存储的一部分,且Shuffle Server侧在存储介质正常的情况下能保证Shuffle数据要么在内存中,要么在存储介质中,那么Commit操作也失去了存在的意义。从下图可以看到,BufferManager包含多个Buffer,每个Buffer存储了单个Partition的Shuffle数据,且存储CachedData中。当BufferManager达到高水位时,CachedData的数据会转移到InFlushData,直到存储写入完成,同时,CachedData还能接收新的Shuffle数据。这样的策略保证了Shuffle数据未写入存储前也不会丢失。

下图则展示了数据在Shuffle Server的内存区域中是如何流转的,写入前先申请内存空间并占据PreAllocation区域,接收到数据后内存使用转移到CachedData区域,在Flush后进一步移动到InFlushData区域,最后写入存储中,并清理掉内存空间。

了解了写入过程,再看读取过程的变化则更容易了,相比之前的单一存储的读取方案,基于混合存储方案读取时,会按序从Shuffle Server Memory, Shuffle Server本地存储及分布式存储读取Shuffle数据。

混合存储的优势

上文已经介绍了混合存储解决的问题及相关实现,这里再做下总结,引入混合存储可以带来如下收益: 1.基于写入数据块大小选择存储介质,提升DFS的写入性能 2.降低对于Shuffle Server本地磁盘容量的依赖,在云原生环境下,更容易部署 3.降低写入Shuffle Server本地磁盘的数据量,当采用SSD作为本地存储时,增加SSD使用寿命,降低存储成本 4.引入内存作为存储,提升计算任务性能

混合存储的使用方式

由于内存的Flush策略进行了变更,Shuffle Server引入的相关配置如下:

#基于rss.server.buffer.capacity值的低水位百分比

rss.server.memory.shuffle.lowWaterMark.percentage 25.0

#基于rss.server.buffer.capacity值的高水位百分比

rss.server.memory.shuffle.highWaterMark.percentage 75.0

目前支持的混合存储类型有:

Shuffle Server端:

注意:由于使用了本地文件和HDFS混合存储,需要增加rss.server.flush.cold.storage.threshold.size该配置,设定单次写入数据量阈值,大于该值将写入HDFS,其余的写入本地文件

rss.storage.type MEMORY_LOCALFILE_HDFS

rss.storage.basePath /path1,/path2

rss.server.hdfs.base.path hdfs://ip:port/path

rss.server.flush.cold.storage.threshold.size 32m

Spark Client端:

spark.rss.storage.type MEMORY_LOCALFILE_HDFS

spark.rss.base.path hdfs://ip:port/path

支持数据过滤

在读取Shuffle数据的过程中,会先读取所有的元数据信息,如,BlockId,TaskId,Length等,再基于元数据信息读取Shuffle数据。由于分布式计算任务的Shuffle数据会产生冗余,如,Spark的推测执行等。为了减少数据的无效读取,更合理的利用系统资源,增加了读取Shuffle数据时的过滤功能。优化的场景如下: 1.Spark AQE 需要读取指定的上游数据 2.Spark 推测执行产生的冗余数据 3.混合存储场景下,数据已从内存读取,又被写入存储而产生的冗余数据

其它特性

除了上述的主要特性,版本还有如下改动:

1.新增对于Spark版本的支持,目前已能支持,Spark2.3, Spark2.4, Spark3.0, Spark3.1

2.优化Shuffle数据读取策略,改为先读取Index文件,再读取Data文件

3.新增GRPC相关指标

4.修复已知缺陷

三、版本性能测试

由于新版本在存储架构上有了较大的变动,以下是性能测试的相关信息

测试环境

硬件环境

1.每台服务器为 176 cores,256G内存,4T * 12 HDD,网络带宽 10GB/s

2.Hadoop Yarn集群:1 * ResourceManager + 6 * NodeManager, 4T * 10 HDD 写临时数据

3.Firestorm集群:1 * Coordinator + 6 * Shuffle Server,4T * 10 HDD 写Shuffle数据

软件环境

1.Hadoop版本2.8.5

2.Spark版本

3. 2.4.Spark相关配置:  

spark.executor.instances 100

spark.executor.cores 4

spark.executor.memory 9g

spark.executor.memoryOverhead 1024

spark.shuffle.manager org.apache.spark.shuffle.RssShuffleManager

spark.rss.storage.type MEMORY_LOCALFILE

4.Firestorm Shuffle Server相关配置:

rss.storage.type MEMORY_LOCALFILE rss.server.buffer.capacity 50g

测试场景:

TPC-DS 基于1TB数据量的TPC-DS,对Spark原生Shuffle,Firestorm 0.1.0,Firestorm 0.2.0进行了对比性能测试,以下是相关测试结果:

从测试结果可以看到,Firestorm0.2.0版本比上一个版本有了30%左右的提升,但对于Spark原生Shuffle并无任何优势,这个结果是符合预期的,原因有如下几点:

1.即使是1TB的TPC-DS测试,query的Shuffle数据量普遍较小,使得磁盘可以忽略由于随机读写而产生的性能下降

2.由于考虑到高并发场景下的网络连接数过多问题,每个Executor和Shuffle Server之间仅存在一个RPC连接,串行发送数据的模式降低了性能

3.客户端在发送完数据后,会每隔一定时间检查发送成功与否,这个间隔时间也增加了任务运行的性能损耗

从性能角度看,Firestorm的优势主要在于减少了存储随机读写带来的性能损耗,由于RPC在实现上更多的考虑稳定性及高并发场景,相比原生Shuffle方案有额外的性能开销,最终导致了在没有磁盘随机IO的场景下,Firestorm性能不如原生Shuffle。 **但是,在磁盘有随机IO的场景下,Firestorm还是具备性能优势的,**为了验证这个结论,将10块HDD降低为2块HDD,选取Shuffle数据量较多的query23a进行测试,测试结果如下:

可以很明显的看到,当HDD数量从10下降到2以后,对于原生Spark的Shuffle Read性能影响严重,读取时间上升了5倍,而对于Firestorm来说,由于随机读写问题不突出,Shuffle Read性能基本没有损耗。 测试场景: TeraSort 基于1TB数据集,对原生Spark Shuffle,Firestorm 进行性能对比测试,结果如下:

由于Shuffle数据量在500GB,从测试结果可以明显看出即使拥有10块HDD,原生Spark的磁盘随机读取造成的Shuffle Read性能下降还是非常明显的。而Firestorm不管是哪个版本,在Shuffle Read的性能上远优于原生Spark。对于Firestorm-0.2.0版本,由于混合存储的存在,Commit操作不再需要,可以看到已经不需要在最后个任务完成后等待Shuffle数据写入存储了。

四、总结

本文介绍了Firestorm 0.2.0版本对于存储侧的一系列改进,其中最为重要的是引入了混合存储功能,利用了内存,本地磁盘,远程存储等资源更合理的分配存储策略。除了提高了性能,还降低了对本地磁盘的依赖,能更好的在云原生的环境下进行部署使用。

附上开源地址,欢迎共建: https://github.com/Tencent/Firestorm

展开阅读全文
加载中
点击引领话题📣 发布并加入讨论🔥
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部