一、背景

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，

4.T * 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