大数据计算的兴起起源于Google的MapReduce纸。 MapReduce的原理非常简单，其过程核心是MAP和READE之间的两个阶段数据交换，即洗牌。

洗牌对大数据计算具有很大的影响。从公共信息来看：来自Facebook [1]，LinkedIn [2]，Alibaba [3]等公司的数据，Shuffle具有很大比例的任务和任务计算时间。从Oppo的在线任务来看，有68的Spark任务具有洗牌计算。

大数据计算引擎的技术演变与随机的优化一直是不可分割的。无论是在执行计划方面进行优化，都必须尽可能避免洗牌操作员的演变或各种洗牌机制的演变，以最大程度地减少耗时的耗时。

洗牌不仅会影响工作的运行效率，而且对计算稳定性产生了很大的影响。开发大数据的学生通常会有这种经验：莫名其妙的散装提取失败错误，甚至任务也会经常失败，他们必须优化任务计算逻辑。

背景

Shuffle Shuffle成为大数据计算效率和稳定性的关键因素的原因是什么？

我们认为有两个要点：

1。磁盘的片段读取和写作，Spill多次写入磁盘，仅将部分分区数据拉动，从而影响效率。

2。减少读取地图侧的本地数据，并且需要MXR远程网络读数，这会影响稳定性。

MAPREDUCE洗牌图[4]

沿解决上述两个问题的解决方案也促进了改进技术的演化。还有两个里程碑：

ESS：外部洗牌服务。 ESS的原则是，地图任务在计算节点上本地合并了相同的分区数据。

RSS：远程洗牌服务。 RSS原理是地图任务将同一分区数据推向远程RSS，RSS合并了同一分区的数据。

ESS vs RSS图

ESS和RSS均旨在解决片段读写的问题以及前面提到的过度RPC连接的问题。 ESS减轻了这种情况，没有RSS就无法彻底解决。

Spark社区提供了远程洗牌服务的接口，公司可以实施自己的RSS。因此，Spark平台上的RSS技术解决方案在过去两年中雨水像蘑菇一样迅速涌现，并被公开揭幕。

相关工作

让我们首先看一下每个公司的解决方案。目前，公共信息和源代码的主要解决方案是：

Uber\’s RSS [5]：开放源代码在2020年，基础存储基于本地磁盘。 Shuffle Server提供读写数据功能，对性能产生一定的影响。此外，它是较早的开源，但维护较少。

Tencent的Firestorm [6]：开源于2021年11月，基础存储使用HDF，几乎没有考虑稳定性和性能优化设计的考虑因素。

阿里巴巴云EMR-RSS [7]：2022年1月的开源。基础存储是基于本地磁盘，并且对本地IO进行了深入的优化。但是，这种基于本地存储的洗牌服务具有自然限制。

LinkedIn磁铁[2]：磁铁严格不是真正的RSS，而是只能被视为基于推动的弹药。磁铁将数据推向远程NodeManager的ESS，而本机洗牌数据被删除在磁盘中。相同的数据将两次删除，这实际上会增加集群的IO压力。但是，磁铁已被纳入Spark 3.2版本中，鉴于此，磁铁的散装制成了这样的设计。

OPPO 解决方案-Shuttle

整体架构

首先，让我们介绍班车的整体建筑：

穿梭架构图

班车主要由两个角色组成，即Shufflemaster和Shuffleworker。

Shufflemaster管理Shuffleworker的状态，并将可用的Shuffleworker分发到任务。

Shuffleworker负责接收ShuffleWriter发送的数据，从同一分区汇总数据，并将其写入分布式存储。

为了确保大师的高可用性，集群有两个主人，一个活跃的和一个备用大师。

如图所示，Activecluster和StandbyCluster分别具有两个主人。

为什么有两个群集活跃和待机？这也是为了服务稳定性，主要用于热升级，这将在下面详细介绍。

架构设计考量

当我们设计分布式的洗牌服务系统时，我们考虑以下方面：

1。数据正确性

数据正确性是生命线。洗牌数据能否围绕远程洗牌服务系统进行，以确保数据没有问题？

我们通过校验和机制确保数据的正确性。写给航天飞机的每个数据都将计算出校验和值。阅读数据时，将计算每个读取的数据。最后，将与校验和检查进行比较，以确保正确读取每个数据，并且仅读取一次。

2。稳定性

稳定性是分布式系统的基石，在分布式系统中，各种问题是不可避免的。

稳定性的保证是系统的问题，而不是可以解决所有稳定问题的功能或设计。我们从以下各个方面讨论班车的稳定性结构：

A.节点/任务控制

Shufflemaster和Shuffleworker在管理和控制方面都有自己的机制。

Shufflemaster用于节点/任务控件的主要功能是：

节点自我修复：Shuffleworker通过心跳向Shufflemaster报告了自己的“健康”信息。如果心跳是定时的，或者“健康”信息异常，则洗牌者将暂停将新任务数据流量分配给节点。工人节点还原“健康”后，它将为修改的节点分配任务。

负载平衡：Spark Tasks请求Shufflemaster的可用Shuffleworker。主决定哪些随机工人根据群集负载分配；同时，分配工人的算法实现是插件，并且可以自定义各种不同的分配策略。

例外拦截：Shufflemaster将在短时间内积极拦截大量相同的任务，以避免影响集群的整体稳定性。

当任务数据量突然增加时，Shuffleworker流量控制机制可确保工人的稳定性。流控制机制主要限制两个方面：

记忆量：流量控制发生时，当Shuffleworker进程使用总存储器超过阈值

连接数：连接数量同时将数据发送给Shuffworker的连接数，如果该连接超过阈值，则将发生流量控制。

B.多机械切换

当地图将数据发送给Shuffleworker时，将有多个Shuffleworker可供选择。当工人遇到问题时（例如工人流控制或断开节点），您可以切换到替代工人继续发送。

如图所示，当ShuffleWriter将数据发送到Shuffleworker A时，节点A失败，然后ShuffleWriter切换到节点B以继续发送数据。

C.分布式存储

Shuttle使用分布式文件系统作为存储码头。

今天，当分布式存储技术如此先进时，我们不需要花费太多精力来优化存储。

将专业的事情留给专业的“人”，主要好处是：

1。降低穿梭系统本身的复杂性并提高自身稳定性

2。分布式文件系统本身具有良好的优势，例如稳定性，可扩展性，负载平衡等。

3.适应多个分布式文件存储，选择多元化并充分利用不同系统的优势

4。使Shuffleworker解除本地存储功能，单独的存储和计算，从而更容易在云上部署

行业中的主流分布式文件系统已完全优化了读写性能。

此外，我们还使用了Cubefs [8]，这是该公司存储团队开发的分布式文件系统。 Cubefs对洗牌方案进行了定制的优化。让我简要介绍Cubefs的优势：

Cubefs体系结构图

Cubefs是CNCF的新一代云本地分布式存储产品，与S3，HDFS和POSIX Access协议兼容，并提供两个存储引擎，即多个复制品和擦除编码，并支持多租户和多AZ部署。

Cubefs创新地采用了用于存储和计算的分离体系结构，提供可扩展的元数据服务，具有低成本的配备模型的擦除编码引擎以及自适应的多级缓存特性，使CUBEF在稳定性，可扩展性，性能和成本，操作和维护等方面都出色。当地对各种访问协议的支持很好，并且扩大了Cubefs产品生态系统。 CUBEF已用于OPPO的各种核心业务，例如大数据存储，大数据洗牌，人工智能，Elasticsearch，MySQL，Data Backup等，这强烈支持各种业务数据的巨大存储需求。

D.热升级

ShuffleService启动后，它将为大量任务提供洗牌服务，并且无法停止服务。同时，系统升级和迭代将不断重新启动服务。为此，系统必须具有热升级的能力。

班车有两个热升级模式：

1。滚动升级：通过Shufflemaster逐一添加黑色- 重新启动Shuffleworker。

该方法对于小型系统也可行。对于相对较大的洗牌服务系统，可以考虑第二个模型。

2。群集切换：Shuffleworker进程绑定机器IP和端口。一台机器可以部署多个工艺过程。因此，我们在同一批机器上在线上部署了两组ShuffleServices，并且可以在升级期间直接切换服务。

自推出以来，它已经经历了许多在线升级和变化，并且没有升级引起的故障情况。

3。性能优化

A.异步传输

数据传输和消息处理都使用Netty异步处理机制，与同步处理机制相比，该机制具有明显的性能优势。同时，消息采用PB格式来改善消息序列化和挑选性能。

B.并发阅读和写作

Shufflewriter和读者使用多线程并发处理进行数据读写。 ringbuffer用作读取器侧基础存储的缓冲区，并且读取过程是异步的。

C.自定义线程池

Shuffleworker将同时处理由不同地图发送的数据。使用Java本地线程池将引入太多的同步机制，从而影响处理数据的速度。为此，我们自定义线程池，以确保将同一分区的数据移交给单个线程进行处理，从而大大降低同步操作并提高处理速度。

不仅如此，为了优化数据传输效率，我们还根据网络MTU自定义数据包大小，并努力争取卓越。

4。可扩展性

答：多集群路由

Shufflemaster可以配置任务路由规则和在线服务的多个群集，并且可以随时切换流量。当集群中发生异常时，任务可以选择切换到普通群集。

B.多店共存

当前，Shuttle支持分布式存储系统，例如HDFS，Cubefs，Alluxio，S3等，并且多个仓库可以同时在线提供服务，无论是在云上还是由自行构建的集群构建，都可以处理。

同时，Shuttle Design考虑了Spark3.x的AQE功能支持。我们在线运行Spark2.4和Spark3.1.2版本的班车。

业界相关技术对比

我们在稳定性，数据正确性保证和性能优化方面比较了与行业相关的工作。

班车在稳定性和性能优化方面都提出了很多考虑。系统启动后，它一直提供稳定的服务。在此期间，它已经升级了很多次，但是没有任何任务失败。以下将引入我们的性能测试效果。

测试效果

在第[3]中，EMR-RSS已与其他开源产品进行了详细比较和测试，并且在性能方面具有明显的优势，因此我们直接将EMR-RS比较和测试。

测试环境

硬件环境：20台物理机器

机器配置：HDD的24个块，384GB内存，CPU 48 CORE。

软件配置：

穿梭使用HDFS存储，均使用默认配置

EMR-RSS使用本地存储，配置使用所有磁盘。 rss.shuffle.writer.mode被配置为排序（默认哈希）

测试任务：Terasort Spark任务

静态资源分配，执行程序800，1000个分区，其他分区使用默认配置。

测试结果

EMR-RSS 1TB Terasort：

穿梭1TB Terasort：

EMR-RSS 5TB Terasort：

班车5TB Terasort：

注意：不同规格的任务的运行时间，两种技术解决方案分别运行5次以找到抽奖值比较

总体而言，与Terasort任务相比，班车和EMR-RSS在几种不同尺寸的数据中的性能提高了4-8。

测试分析

班车的阅读数据显然很快，分析原因如下：

1。从HDFS读取航天飞机读取数据，并且不占据Shuffleworker Process Resources；

2。穿梭数据阅读方法是异步管道方法。

但是，我们还看到，班车比EMR-RS较慢，以撰写数据，分析原因如下：

1。航天飞机的流控制机制，每次发送数据包时，都会首先获得令牌并更多地交互网络。

2. Shuttle的校验和机制，在发送每个分区数据后，将发送额外的校验和数据包，最后一个校验和数据包在同步通信中。

基于上述分析，Shuttle在确保稳定性和数据准确性方面取得了一些性能权衡。但是，由于阅读数据的速度更快，因此它不仅可以弥补编写数据引起的性能差距，而且总体性能也得到了改善。

线上效果

目前，Oppo Group的大数据计算任务中的30的混洗数据已连接到航天飞机，最佳大型任务的执行效率增加了50+；总体效应数据如下图所示：

未来展望

为了使班车能够影响更多的计算，我们决定开源班车项目[9]。

对于技术进化的方向，我们计划从三个方向做到这一点：

1。访问更多的计算机，例如Flink，Trino等。

2。依靠现有的分布式存储，优化基础存储并适应洗牌方案的特殊需求。

3。提供更多的计算服务，不仅限于远程洗牌服务。

David FU：OPPO大数据计算平台架构师。负责大数据计算平台技术发展的设计和开发。他曾经在Alibaba Cloud和Qunar.com大数据平台上工作。他拥有10年的大数据架构和开发经验。

XUEN：OPPO的高级数据平台工程师，目前在OPPO数据架构团队中工作，主要负责Spark Computing Engine和Shuttle的开发，并且在大数据架构和开发方面具有丰富的经验。

附录

[1] Haoyu Zhang，Brian Cho，Ergin Seyfe。 Riffle:优化的大规模数据分析的洗牌服务。 ACM 2018

[2] Min Shen，Ye Ye Zhou，Chandni Singh。 Magnet:基于推动的洗牌服务，用于大规模数据处理。 VLDB 2020

[3] Alibaba Cloud EMR远程随机服务在小米中的实践。

https://MP.Weixin.qq.com/s/xdbmkkkl4nw7eefnmdxxyq

[4] 《Hadoop 权威指南》

[5] Ubser Spark RSS: https://Github.com/uber/RemoteshuffleService

[6] Tencent Spark RSS Firestorm：https://github.com/tencent/firestorm

[7] Alibaba Cloud Spark RSS：https://Github.com/alibaba/remoteshuffleservice

[8] Cubefs：https://github.com/cubefs/cubefs