Spark 颠覆 MapReduce 保持的排序记录

标签: 综合新闻 | 发表时间:2014-10-11 22:25 | 作者:
分享到:
出处:http://www.oschina.net/?from=rss

在过去几年,Apache Spark的采用以惊人的速度增加着,通常被作为MapReduce后继,可以支撑数千节点规模的集群部署。在内存中数 据处理上,Apache Spark比MapReduce更加高效已经得到广泛认识;但是当数据量远超内存容量时,我们也听到了一些机构在Spark使用 上的困扰。因此,我们与Spark社区一起,投入了大量的精力做Spark稳定性、扩展性、性能等方面的提升。既然Spark在GB或TB级别数据上运行 良好,那么它在PB级数据上也应当同样如此。

为了评估这些工作,最近我们与AWS一起完成了一个Sort Benchmark(Daytona Gray类别)测试,一个考量系统排序 100TB数据(万亿条记录)速度的行业基准测试。在此之前,这项基准测试的世界记录保持者是雅虎,使用2100节点的Hadoop MapReduce 集群在72分钟内完成计算。而根据测试结果得知,在使用了206个EC2节点的情况下,Spark将排序用时缩短到了23分钟。这意味着在使用十分之一计 算资源的情况下,相同数据的排序上,Spark比MapReduce快3倍!

此外,在没有官方PB排序对比的情况下,我们首次将Spark推到了1PB数据(十万亿条记录)的排序。这个测试的结果是,在使用190个节点的情 况下,工作负载在短短不到4小时内完成,同样远超雅虎之前使用3800台主机耗时16个小时的记录。同时,据我们所知,这也是公用云环境首次完成的PB级 排序测试。


Hadoop World Record Spark 100 TB Spark 1 PB
Data Size 102.5 TB 100 TB 1000 TB
Elapsed Time 72 mins 23 mins 234 mins
# Nodes 2100 206 190
# Cores 50400 6592 6080
# Reducers 10,000 29,000 250,000

1.42 TB/min 4.27 TB/min 4.27 TB/min
Rate/node 0.67 GB/min 20.7 GB/min 22.5 GB/min
Sort Benchmark Daytona Rules Yes Yes No
Environment dedicated data center EC2 (i2.8xlarge) EC2 (i2.8xlarge)

为什么会选择排序?    

排序的核心是shuffle操作,数据的传输会横跨集群中所有主机。Shuffle基本支持了所有的分布式数据处理负载。举个例子,在一个 连接了两个不同数据源的SQL查询中,会使用shuffle将需要连接数据的元组移动到同一台主机;同时,类似ALS等协同过滤算法同样需要依赖 shuffle在网络中发送用户或产品的评级(ratings)和权重(weights)。

大部分数据管道开始时都会有大量的原始数据,但是在管道处理过程中,随着越来越多不相干数据被过滤,或者中间数据被更简洁的表示,数据量必 然会减少。在100TB原始数据的查询上,网络上shuffle的数据可能只有100TB的一小部分,这种模式也体现在MapReduce的命名。

然而,排序却是非常有挑战的,因为数据管道中的数据量并不会减少。如果对100TB的原始数据进行排序,网络中shuffle的数据必然也 是100TB。同时,在Daytona类型的基准测试中,为了容错,不管是输入数据还是输出数据都需要做备份。实际上,在100TB的数据排序上,我们可 能会产生500TB的磁盘I/O及200TB的网络I/O。

因此,基于上述原因,当我们寻找Spark的测量标准和提升办法时,排序这个最苛刻的工作负载成为了对比的不二之选。

产生如此结果的技术实现    

在超大规模工作负载上,我们投入了大量的精力来提升Spark。从细节上看,与这个基准测试高度相关的工作主要有3个: 

首先及最关键的,在Spark 1.1中我们引入了一个全新的shuffle实现,也就是基于排序的 shuffle(SPARK­2045)。在此之前,Spark做的是基于哈希的shuffle实现,它需要在内存中同时保持P(reduce的分割数 量)个缓冲区。而在基于排序的shuffle下,任何时候系统只使用一个缓冲区。这个操作将显著地减少内存开销,因此同一个场景下可以支撑数十万任务(我 们在PB排序中使用了2.5万个任务)。

其次,我们修订了Spark的网络模型,通过JNI(SPARK­2468)使用基于Netty的Epoll本地端口传输。同时,新的模型还拥有了独立的内存池,绕过了JVM的内存分配器,从而减少垃圾回收造成的影响。

最后但同样重要的是,我们建立了一个外部shuffle服务(SPARK­3796),它与Spark本身的执行器完全解耦。这个新的服务基于上文所述的网络模型,同时,在Spark本身的执行器忙于GC处理时,它仍然可以保证shuffle文件处理的继续执行。

                           
   

通过这三项改变,我们的Spark集群在map阶段单 节点可以支撑每秒3GB的IO吞吐,在reduce阶段单节点可以支撑1.1GB,从而榨干这些机器间10Gbps的网络带宽。

更多的技术细节    

TimSort:在Spark 1.1版本中,我们将默认排序算法从 quicksort转换到TimSort,它是合并排 序和嵌入排序的一个衍生。在大部分现实世界数据集中,TimSort比quicksort更加高效,在部分排序数据中表现则更为优秀。不管在map阶段还 是Reduce阶段,我们都使用了TimSort。

缓存位置的利用:在排序基准测试中,每条记录的大小都是100字节,而被排序的键是前10个字节。在排序项目的性能分析阶 段,我们注意到缓存命中率不如人意,因为每次比较都需要进行一个随机的对象指针查询。为此,我们重新设计了记录在内存的布局,用16字节长度(两个长整 形)的记录来表示每条记录。在这里,前10个字节代表了排序的键,后4个字节则代表了记录的位置(鉴于字节顺序和符号,这点并不容易发现)。这样一来,每 个比较只需要做一次缓存查询,而且它们都是连续的,从而避免了随机的内存查询。

使用TimSort和新的布局方式来利用缓存命中,排序所占用的CPU时间足足减少了5倍。

大规模下的容错机制:在大规模下,许多问题都会暴露。在这个测试过程中,我们看到因为网络连通问题出现的节点丢失,Linux内核自旋,以及因为内存碎片整理造成的节点停滞。幸运的是,Spark的容错机制非常好,并且顺利的进行故障恢复。

AWS的能量:如上文所述,我们使用了206个i2.8xlarge实例来跑这个I/O密集测试。通过SSD,这些实例交付了非常高的I/O吞吐量。我们将这些实例放到一个VPC放置组中,从而通过单SR-IOV增强网络性能,以获得高性能(10Gbps)、低延时和低抖动。        
   

Spark只能在内存中大放异彩?    

这个误解一直围绕着Spark,特别是刚进入社区中的新人更是如此认为。不错,Spark因为内存计算的高性能闻名,然而Spark的设计 初衷和理念却是一个通用的大数据处理平台——不管是使用内存还是磁盘。在数据无法完全放入内存时,基本上所有的Spark运算符都会做一些额外的处理。通 俗来说,Spark运算符是MapReduce的超集。

如本次测试所示,Spark可以胜任集群内存大小N倍的数据集处理。

总结    

击败Hadoop MapReduce集群创造的大规模数据处理记录不仅是对我们工作的一个证明,也是对Spark承诺的一个验证——在任何数据体积,Spark在性能和扩展性上都更具优势。同时,我们也希望在用户使用过程中,Spark可以带来时间和开销上的双节省。

博文链接: Spark Breaks Previous Large-Scale Sort RecordCSDN 翻译/童阳 责编/仲浩)

相关 [spark mapreduce 排序] 推荐:

Spark与Mapreduce?

- - 崔永键的博客
我本人是类似Hive平台的系统工程师,我对MapReduce的熟悉程度是一般,它是我的底层框架. 我隔壁组在实验Spark,想将一部分计算迁移到Spark上. 年初的时候,看Spark的评价,几乎一致表示,Spark是小数据集上处理复杂迭代的交互系统,并不擅长大数据集,也没有稳定性. 但是最近的风评已经变化,尤其是14年10月他们完成了Peta sort的实验,这标志着Spark越来越接近替代Hadoop MapReduce了.

Spark 颠覆 MapReduce 保持的排序记录

- - 开源中国社区最新新闻
在过去几年,Apache Spark的采用以惊人的速度增加着,通常被作为MapReduce后继,可以支撑数千节点规模的集群部署. 在内存中数 据处理上,Apache Spark比MapReduce更加高效已经得到广泛认识;但是当数据量远超内存容量时,我们也听到了一些机构在Spark使用 上的困扰. 因此,我们与Spark社区一起,投入了大量的精力做Spark稳定性、扩展性、性能等方面的提升.

mapreduce倒排序索引

- - CSDN博客云计算推荐文章
  private Text word_filepath = new Text();//文件路径.   private  Text one  = new Text("1");//个数.      job.setCombinerClass(IndexCombiner.class);//优化.

MapReduce之二次排序

- - CSDN博客云计算推荐文章
     面试官问了一个MapReduce问题:“如何用MapReduce实现两个表的连接”.      我说“用两个job实现,第一个. 于是这两天回来看了下MapReduce的二次排序.      在某些情况下,需要对reduce中的value进行排序. 二次排序,可以将根据key聚合起来的valueList根据value进行排序.

【原】MapReduce的排序和二次排序

- - ITeye博客
自己学习排序和二次排序的知识整理如下. 1.Hadoop的序列化格式介绍:Writable. 2.Hadoop的key排序逻辑. 4.如何自定义自己的Writable类型. 1.Hadoop的序列化格式介绍:Writable. 要了解和编写MR实现排序必须要知道的第一个知识点就是Writable相关的接口和类,这些是HADOOP自己的序列化格式.

利用hadoop mapreduce 做数据排序

- - zzm
我们的需求是想统计一个文件中用IK分词后每个词出现的次数,然后按照出现的次数降序排列. 由于hadoop在reduce之后就不能对结果做什么了,所以只能分为两个job完成,第一个job统计次数,第二个job对第一个job的结果排序. 第一个job的就是hadoop最简单的例子countwords,我要说的是用hadoop对结果排序.

Mapreduce小结

- MAGI-CASPER/Peter Pan - 博客园-唯有前进值得敬仰
读完mapreduce论文小结一下. 1.MapReduce是一个编程模型,封装了并行计算、容错、数据分布、负载均衡等细节问题. 输入是一个key-value对的集合,中间输出也是key-value对的集合,用户使用两个函数:Map和Reduce. Map函数接受一个输入的key-value对,然后产生一个中间key-value 对的集合.

Hadoop MapReduce技巧

- - 简单文本
我在使用Hadoop编写MapReduce程序时,遇到了一些问题,通过在Google上查询资料,并结合自己对Hadoop的理解,逐一解决了这些问题. Hadoop对MapReduce中Key与Value的类型是有要求的,简单说来,这些类型必须支持Hadoop的序列化. 为了提高序列化的性能,Hadoop还为Java中常见的基本类型提供了相应地支持序列化的类型,如IntWritable,LongWritable,并为String类型提供了Text类型.

MapReduce原理

- - C++博客-牵着老婆满街逛
       MapReduce 是由Google公司的Jeffrey Dean 和 Sanjay Ghemawat 开发的一个针对大规模群组中的海量数据处理的分布式编程模型. MapReduce实现了两个功能. Map把一个函数应用于集合中的所有成员,然后返回一个基于这个处理的结果集. 而Reduce是把从两个或更多个Map中,通过多个线程,进程或者独立系统并行执行处理的结果集进行分类和归纳.

MapReduce优化

- - 行业应用 - ITeye博客
相信每个程序员在 编程时都会问自己两个问题“我如何完成这个任务”,以及“怎么能让程序运行得更快”. 同样,MapReduce计算模型的多次优化也是为了更好地解答这两个问题. MapReduce计算模型的优化涉及了方方面面的内容,但是主要集中在两个方面:一是计算性能方面的优化;二是I/O操作方面的优化.