oppo 开源高可用、高性能的 spark remote shuffle service-金马国际

大数据计算的兴起，源于 google 的 mapreduce 论文，mapreduce 的原理很简单，其流程核心则是 map 和 reduce 两阶段数据交换，也即 shuffle。

shuffle 对大数据计算影响很大，从公开的资料：facebook[1]、linkedin[2]、阿里[3] 等公司的数据看，shuffle 影响的任务和任务计算时间上都有较高占比。从 oppo 的线上任务看，68%的 spark 任务都有 shuffle 计算。

大数据计算引擎的技术演进，一直离不开对 shuffle 的优化，无论是从执行计划方面优化，尽量避免 shuffle 算子还是各种 shuffle 机制的演进，都是为了尽量缩短 shuffle 的耗时。

shuffle 不仅影响作业运行效率，对计算稳定性也有较大影响，大数据开发的同学一般都有这样的经历：莫名的 shuffle fetch fail 错误，甚至任务会因此频繁失败，不得不优化任务计算逻辑。

背景

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

我们认为主要有两点：

1、磁盘的碎片读写，spill 多次写磁盘和 reduce 只拉取部分 partition 数据，影响效率。

2、reduce 读取 map 端本地数据，需要 mxr 次远程网络读，影响稳定性。

mapreduce shuffle 示意图[4]

shuffle 技术演进，主线也是沿着解决上面两个问题推进。比较有里程碑意义的有两个方向：

ess：external shuffle service，ess 原理是 map 任务在计算节点本地将相同 partition 数据合并到一起；

rss：remote shuffle service，rss 原理是 map 任务将相同 partition 数据 push 到远端的 rss，rss 将同一 partition 的数据合并。

ess vs rss 示意图

ess 和 rss 都是为了解决前面我们提到的碎片读写和 rpc 连接过多的问题，ess 是缓解了这种情况，没有 rss 解决的彻底。

spark 社区提供了 remote shuffle service 的接口，各家公司可以自己实现自己的 rss。所以，近两年在 spark 平台的 rss 技术方案如雨后春笋，纷纷公开亮相。

相关工作

我们先看一下各家的金马国际的解决方案，目前公开资料和源码的方案主要有：

• uber 的 rss [5]：2020 年开源，底层存储基于本地磁盘，shuffle server 提供读写数据功能，对性能有一定的影响，另外，开源时间比较早，但维护较少。

• 腾讯的 firestorm [6]：2021 年 11 月开源，底层存储使用 hdfs，对稳定性以及性能优化设计考虑较少。

• 阿里云 emr-rss [7]：2022 年 1 月开源，底层存储基于本地磁盘，对本地 io 做了深入的优化，不过这种基于本地存储的 shuffle service，有着天然的限制。

• linkedin magnet [2]：magnet 严格来说不算真正意义的 rss，只能算是 push based 的 shuffle。magnet 在 spark 原生 shuffle 数据落盘的同时把数据 push 到远端 nodemanager 的 ess 上，同一份数据，会落盘两次，这样其实会增加集群的 io 压力。不过，magnet 已经合入到 spark3.2 版本，鉴于此，magnet 的 shuffle 才做了这样的设计。

oppo 金马国际的解决方案-shuttle

整体架构

首先，介绍一下 shuttle 的整体架构：

shuttle 架构图

shuttle 主要由两个角色组成，shufflemaster 和 shuffleworker。

shufflemaster 负责管理 shuffleworker 的状态，向任务分发可用的 shuffleworker。

shuffleworker 负责接收 shufflewriter 发送的数据，并将同一 partition 的数据聚合，写入分布式存储。

为保障 master 高可用，一个集群有两个 master，一个 active 和一个 backup master。

如图所示，activecluster 和 standbycluster 分别有两个 master。

为什么会有 active 和 standby 两个 cluster ？这也是为了服务的稳定性考虑，主要用于热升级，下面会详细介绍。

架构设计考量

我们在设计一个分布式的 shuffle service 系统的时候，从下面几个方面考虑：

1.数据正确性

数据正确性是生命线，shuffle 数据在 remote shuffle service 系统走一圈，能否保障数据不出问题？

我们通过 checksum 机制保障数据的正确性。每一条写入 shuttle 的数据，都会计算一个 checksum 值，最后读数据的时候同样对读取的每一条数据计算 checksum，最后对比 checksum，保证每条数据都被正确读到且只被读一次。

2.稳定性

稳定性是分布式系统的基石，在分布式系统中，出现各种问题是必然。

稳定性的保障，是一个系统性的问题，不是某一个 feature 或者设计能解决所有稳定性问题，我们从以下几个方面讨论 shuttle 的稳定性建设：

a、节点/任务管控

shufflemaster 和 shuffleworker 在管控方面都有自己的机制。

shufflemaster 对节点/任务管控的功能主要有：

节点自愈：shuffleworker 通过心跳向 shufflemaster 上报自身的“健康”信息。心跳超时或者“健康”信息异常，shufflemaster 会暂停向该节点分配新的任务数据流量，worker 节点恢复“健康”后，再向改节点分配任务。

负载均衡：spark 任务向 shufflemaster 请求可用的 shuffleworker，master 根据集群负载决定分配哪些 shuffleworker；同时，分配 worker 的算法实现是插件式的，可以定制多种不同的分配策略。

异常拦截：对于用户短时间提交的大量相同任务，shufflemaster 会主动拦截，避免影响集群整体稳定性。

shuffleworker 流控机制，当任务数据量突增场景下，流控保障 worker 的稳定性。流控机制主要从两方面限制：

内存量：shuffleworker 进程使用总内存超过阈值即发生流控

连接数：同时向 shuffleworker 发送数据的连接数，超过阈值即发生流控

b、多机切换

map 向 shuffleworker 发送数据，会有多个 shuffleworker 可供选择，当某个 worker 出问题（比如 worker 发生流控，或节点掉线），可以切换到备选 worker 继续发送。

如图所示，shufflewriter 在向 shuffleworker a 发送数据的时候，a 节点出现故障，shufflewriter 切换到 b 节点继续发送数据。

c、分布式存储

shuttle 采用分布式文件系统作为存储底座。

在分布式存储技术如此发达的今天，我们不需要花费过多精力优化存储。

专业的事情交给专业的“人”来做，这样的好处主要有：

1、降低 shuttle 系统本身的复杂度，提升自身稳定性

2、分布式文件系统自身具有良好的稳定性，扩展性，负载均衡等优势

3、适配多种分布式文件存储，选择多样化，充分利用不同系统优势

4、使得 shuffleworker 解耦本地存储能力，存算分离，更易于云上部署

业界主流的分布式文件系统，本身对读写性能都做了充分的优化。

另外，我们大量使用了公司存储团队自研的分布式文件系统 cubefs[8]，cubefs 针对 shuffle 场景做了定制化的优化，简单介绍一下 cubefs 的优势：

cubefs 架构图

cubefs 是 cncf 新一代云原生分布式存储产品，兼容 s3、hdfs、posix 多种接入协议，提供多副本和纠删码两种存储引擎，支持多租户、多 az 部署。

cubefs 创新性采用存算分离架构，提供可扩展的元数据服务，低成本的模式可配的纠删码引擎，自适应多级缓存特性，使得 cubefs 在稳定性、扩展性、性能与成本、可运维性等方面均表现优秀；对多种接入协议的原生支持，与容器兼容性好，拓宽了 cubefs 产品生态；cubefs 已经被用于 oppo 各个核心业务，如大数据存储，大数据 shuffle、人工智能、elasticsearch、mysql、数据备份等，有力支撑各类业务数据海量存储需求。

d、热升级

shuffleservice 一旦上线，会为大量任务提供 shuffle 服务，不能停服，同时，系统的升级迭代会不断需要重启服务。为此，系统必须具备热升级的能力。

shuttle 有两种热升级模式：

1、滚动升级：通过 shufflemaster 逐一加黑-重启 shuffleworker。

这种方式针对小规模系统还可行，对于规模比较大的 shuffleservice 系统，可以考虑第二种模式。

2、集群切换：shuffleworker 进程绑定机器 ip 和端口，一台机器可以部署多个 worker 进程，因此我们在线上同一批机器部署两套 shuffleservice，升级的时候可以直接整体切换服务。

上线以来，经历线上多次升级变更，无一例因为升级导致的失败 case。

3.性能优化

a、异步传输

数据传输和消息处理，均使用 netty 异步处理机制，对比同步处理机制，性能有明显优势。同时，消息采用 pb 格式，提升消息序列化和反序列化性能。

b、并发读写

shufflewriter 和 reader 对于数据的读写均采用多线程并发处理，在 reader 端使用 ringbuffer 作为底层存储的缓冲，读过程异步化。

c、定制线程池

shuffleworker 会并发处理不同的 map 发送的数据，使用 java 原生线程池会引入过多的同步机制，影响处理数据速度。为此，我们定制线程池，确保同一 partition 的数据交由单一线程处理，显著降低同步操作，提升处理速度。

不仅如此，为优化数据传输效率，我们根据网络 mtu 定制数据包大小，精益求精。

4.扩展性

a、多集群路由

shufflemaster 可配置任务路由规则，多个集群在线服务，随时可以切换流量。在集群出现异常，任务可以选择切换到正常的集群。

b、多存储共存

目前 shuttle 支持 hdfs、cubefs、alluxio、s3 等分布式存储系统，多种存储可以同时在线提供服务，无论是云上还是自建集群，均可应对。

同时，shuttle 设计就考虑到 spark3.x 的 aqe 特性支持，我们线上同时运行着支持 spark2.4 和 spark3.1.2 版本的 shuttle。

业界相关技术对比

针对稳定性，数据正确性保障，性能优化方面，我们跟业界相关工作做了对比。

shuttle 在稳定性和性能优化方面做了很多考量，系统上线后一直提供稳定服务，期间多次升级，无一任务因此失败，下面会介绍一下我们的性能测试效果。

测试效果

文章[3]中，emr-rss 已经跟其他的开源产品做了详细的对比测试，且在性能上有明显的优势，所以，我们直接跟 emr-rss 对比测试。

测试环境

硬件环境：20 台物理机

机器配置：24 块 hdd，内存 384gb，cpu 48 核心。

软件配置：

shuttle 使用 hdfs 存储，均使用默认配置

emr-rss 使用本地存储，配置使用所有磁盘。rss.shuffle.writer.mode 配置为 sort（默认为 hash）

测试任务：terasort spark 任务

静态资源分配，executor 800，分区数 1000，其他使用默认配置。

测试结果

emr-rss 1tb terasort：

shuttle 1tb terasort：

emr-rss 5tb terasort：

shuttle 5tb terasort：

注：不同规格任务运行时间，两种技术方案分别运行 5 次求平局值对比

整体看，shuttle 和 emr-rss 对比 terasort 任务在几个不同规模数据量上有 4%-8%的性能提升。

测试分析

shuttle 的读数据明显快，分析原因如下：

1、shuttle 读数据从 hdfs 读取，不占用 shuffleworker 进程资源；

2、shuttle 读数据方式是异步流水线方式。

但是，我们也看到 shuttle 在写数据要比 emr-rss 慢，分析原因如下：

1、shuttle 的流控机制，在每次发送数据包会先获取一次令牌，多一次网络交互。

2、shuttle 的 checksum 机制，在每个分区数据发送完毕后，会多发一个 checksum 包，且最后的 checksum 包是同步方式通信。

由上分析，shuttle 在保障稳定性和数据正确性上做了一些性能取舍。但是，由于读数据的 速度更快，不仅弥补了写数据导致的性能 gap，整体性能还是有提升。

线上效果

目前，oppo 集团大数据计算任务 30%的 shuffle 数据已经接入 shuttle，效果最好的大任务执行效率提升 50% ；整体效果数据见下图：

未来展望

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

对于技术演进方向，我们计划从三个方向进行：

1、接入更多的计算引擎，比如 flink、trino 等。

2、依托现有的分布式存储，优化底层存储，适应 shuffle 场景的特殊需求。

3、提供更多的计算服务，不局限于 remote shuffle 服务。

关于作者：

david fu ：oppo 大数据计算平台架构师。负责大数据计算平台技术演进设计开发，曾供职于阿里云，去哪儿网大数据平台，拥有 10 年大数据架构，开发经验。

xuen：oppo 高级数据平台工程师，目前就职于 oppo 数据架构团队，主要负责 spark 计算引擎和 shuttle 的开发，拥有丰富大数据架构和开发经验。

附录

[1] haoyu zhang, brian cho, ergin seyfe. riffle: optimized shuffle service for large-scale data analytics. acm 2018

[2] min shen, ye zhou, chandni singh. magnet: push-based shuffle service for large-scale data processing. vldb 2020

[3] 阿里云 emr remote shuffle service 在小米的实践。

[4] 《hadoop 权威指南》

[5] ubser spark rss:

[6] 腾讯 spark rss firestorm：

[7] 阿里云 spark rss：

[8] cubefs：

[9] shuttle: