Spark原理 | Cosco:高效的Spark Shuffle服务

背景

Spark应用中，Shuffle服务的可靠性和性能直接影响了Spark应用的执行效率，来自Facebook的Brian Cho与Dmitry Borovsky在今年4月份旧金山举行的Spark AI峰会上，分享了他们为Spark/Hive Shuffle优化做的工作，这些工作中的很大部分已经应用于Facebook的大数据平台生产环境，对于超大规模的 Spark 数据处理优化，有一定的参考借鉴价值。

从Facebook这类超大数据处理平台需求角度来看，分析计算与存储的分离带来的好处包括：

• 计算节点和存储节点对于服务器硬件的配置需求是不太一样的，可以针对不同的功能需求，合理优化硬件配置；

• 对于计算能力和存储能力的管理与资源分配，会带来明显的好处，特别是在多应用多租户场景下；

• 计算节点和存储节点在硬件升级更新换代的周期也明显不同，对于成本控制也会有极大的好处；

此外，计算节点往往需要更多的CPU计算能力和内存，但是没有长时间数据存储的需求，存储节点则往往使用的是机械磁盘；持久化存储的数据访问模式是大量文件的顺序读写，而Spark计算产生临时数据，其数据访问模式往往呈现的是小片数据的随机读写。

对于机械磁盘每次I/O请求，读取数据量大小与磁盘吞吐量的量化关系，简而言之，如果每次I/O请求，读取数据量较小，整体的数据读的吞吐量会急剧下降，如上面图标所示，加载同样的数据量，如果每次I/O请求只读取64KB，耗时140秒，相反每次I/O请求读取4MB，则只需要8秒完成数据加载。小数据I/O请求，大量的时间都浪费在磁盘寻道上。

另外，要想维持相同的读写吞吐率，当磁盘的容量加大时，必须加大每次I/O请求的数据块大小。

磁盘访问的几个关键指标：

• 磁盘数据读取时间（Disk Service Time），与之对应的是磁盘寻道（Disk Seek Time）的时间，我们期待数据读取的时间（Service Time)占比高，才能提高磁盘的访问效率；

• 平均每次I/O请求读取数据量大小（取决与应用程序本身）；

• 磁盘本身的WA（写放大效应）也会导致磁盘效率使用的变化；

在Spark Shuffle过程中，Map output和下一个Stage读取数据时，当数据量足够大导致数据Spill时，也会导致额外的数据读写，可导致实际写的数据量为Shuffle数据量的3倍。

Spark Shuffle过程中还会有大量的小数据块I/O请求，我们知道Mapper生成的每一个partition数据，都会被下一个stage的所有的reducer来读取，在大规模数据处理应用中，Reducer数量往往很大，facebook工程师观察到他们的典型应用场景，平均每个Reducer只会读取每个Map Output输出文件中的200k数据。

Facebook工程师去年Spark AI峰会也分享过Spark Shuffle的优化，内部项目名称叫SOS，其原理是在Map输出的时候，将多个Mapper输出的文件先在Map端做有一个预聚合，10路Mapper输出数据聚合后，Reducer读取Map输出数据时，平均I/O请求的数据大小为2MB（从200KB到2MB），极大提高了数据访问效率。

今年的Spark AI峰会，facebook工程团队带来了SOS项目的升级版本Cosco，让所有的Mapper输出文件，对应同一个reduce partition都能共享同一个Write-ahead Buffer，确保每一个Reduce Partition都能生成经可能少的文件。让Reducer读取文件完全变成顺序读取，极大减小了小数据量I/O请求问题，顺序读的平均每次I/O请求读取数据量从200KB到2.5MB，同时因为减小了数据Spill的可能性，Shuffle过程中的写放大（WA）从原先3倍变成1.2倍的Shuffle数据量。

Cosco项目，不仅仅可以服务Spark，也可以为Hive（MapReduce）作业的shuffle提供服务，目前已经在facebook的生产环境中有一年多的实践经验，平均提高磁盘3.2倍的效率；在Spark应用中，目前还没有正式在生产环境中引入Cosco优化，不过目前测试的结果来看，提高了3.5倍的磁盘利用效率，并且没有观察到明显的CPU利用率增加。

正式开始对Cosco项目进行分享之前，对于Facebook的大数据计算中的Shuffle场景进行总结：

• Shuffle的数据交换发生在计算与存储已经分离的环境中；

• Shuffle的数据量通常都在PB级别，涉及到10万个Mapper以及10万个Reducer；

• 由于 Shuffle 数据量巨大，Shuffle过程中Mapper阶段与Reducer阶段的数据Spill不可避免，因此，实际写的数据量是Shuffle数据量的3倍（也就是3PB的数据量会写到磁盘）；

• Shuffle阶段平均每次I/O请求，读取200KB数据；

• Shuffle阶段，I/O请求基本上是同时发生的（下一个Stage刚刚开始时，基本上所有的Reducer都同时读取Mapper的输出数据），更加降低量磁盘的随机读写效率；

• Cosco是一个独立的外部服务；

Cosco受到2012年的一篇论文的启发，Shuffle数据性能瓶颈和前文提到的原因基本差不多，而论文的核心观点是将Shuffle的Reduce Partition的数据尽可能在Shuffle的过程中进行聚合。

核心概念如上图：不同Mappers的输出数据，对应同一个Reduce Partition的，共享同一个Write-ahead buffer，这个Write-ahead buffer由外部独立Shuffle Service (Cosco)来维护，Write-ahead buffer会尽可能归并（Merge）足够大的Mapper生成的连续数据块，并写到外部的分布式文件系统（DFS），一个Write-ahead buffer会生成多个文件，每个文件内部为有序状态，文件之间为无序，所以当Reducer读取Write-ahead buffer输出的文件时，需要做一个归并（Merge Sort)，从这个过程中可以看到Cosco提供的Shuffle服务，通过两次全局Reduce Partition归并排序（排序这事不是必须的，与当前执行算子有关），一方面减小了Shuffle时，由于Spill数据带来的写放大效应，另外一方面也规避了Reduce读取数据时高并发读取小数据块带来的磁盘随机访问的低效率。

不过由于Cosco是独立的外部Shuffle服务，更多的问题体现在独立服务自身的可靠性和容错性，因此原文中对于容错和可靠性进一步展开，欢迎阅读原文及视频链接：

【视频】

https://pan.baidu.com/s/10HmEy1zbVnfsZQrllTwl8A?fid=625499763685421

【原文】

https://www.slidestalk.com/s/Cosco_An_Efficient_Facebook_Scale_Shuffle_Service

-- 结束 --

公益技术沙龙

大家工作学习遇到HBase技术问题，把问题发布到HBase技术社区论坛http://hbase.group，欢迎大家论坛上面提问留言讨论。想了解更多HBase技术关注HBase技术社区公众号(微信号:hbasegroup)，非常欢迎大家积极投稿。

本群为HBase+Spark技术交流讨论，整合最优质的专家资源和技术资料会定期开展线下技术沙龙，专家技术直播，专家答疑活动

点击链接钉钉入群：https://dwz.cn/Fvqv066s或扫码进群

本群为Cassandra技术交流讨论，整合最优质的专家资源和技术资料会定期开展线下技术沙龙，专家技术直播，专家答疑活动

Cassandra 社区钉钉大群：https://c.tb.cn/F3.ZRTY0o

Cassandra 技术社区微信公众号：