MapReduce Shuffle过程详解

Shuffle过程是MapReduce的核心,也被称为奇迹发生的地方。要想理解MapReduce,Shuffle是必须要了解的。我看过很多相关方面的资料,但每次看完都云里雾里的绕着,很难理清大致的逻辑,反而越搅越乱。前端时间在做MapReduce job性能调优的工作,需要深入代码研究MapReduce的运行机制,这才对Shuffle探了个究竟。考虑到之前我在看相关资料而看不懂时很恼火,所以在这里我尽最大的可能试着把Shuffle说清楚,让每一位想了解它原理的朋友都能有所收获。如果你对这篇文章有任何疑问或建议请留言到后面,谢谢!

Shuffle的正常意思是洗牌或弄乱,可能大家更熟悉的是JAVA API里的Collections.shuffle(List)方法,它会随机地打乱参数list里的元素顺序。如果你不知道MapReduce里Shuffle是什么,那么请看这张图:

这张是官方对Shuffle过程的描述。但我可以肯定的是,但从这张图你基本不可能明白Shuffle的过程,因为它与事实相差挺多,细节也是挺乱的。后面我会具体描述Shuffle的事实情况,所以这里你只要清楚Shuffle的大致范围就OK,Shuffle的大致范围是Map端输出结果到数据传送到Reduce端。也可以这样理解,Shuffle描述着数据从MapTask输出到ReduceTask输入的这段过程。

在Hadoop这样的集群环境中,大部分MapTask与ReduceTask的执行是在不同的节点上的,当然很多情况下Reduce执行时需要跨节点去拉取其他节点上MapTask的结果。如果集群正在运行的job很多,那么Task的正常执行对集群内部的网络资源消耗会很严重。这种网络消耗是正常的,我们不能限制,能做的就是最大化的减少不必要的消耗。还有在节点内,相比与内存,磁盘IO对Job完成时间影响也是可观的。从最基本的要求来说,我们对Shuffle过程的期望可以有:

(1):完整地从Map Task端拉取数据到Reduce端。

(2):在跨节点拉取数据时,尽可能地减少对带宽的不必要消耗。

(3):减少磁盘IO对Task执行的影响。

OK,看到这里时,大家可以先停下来想想,如果是自己来设计这段Shuffle过程,那么你的实际目标是什么。我能优化的地方主要在于减少拉取数据的量及尽量使用内存而不是磁盘。

我的分析基于Hadoop0.21.0的源码,如果与你所认识的Shuffle过程有差别,不吝指出。我会以WordCount为例,并假设它有8个MapTask和3个Reduce Task。从上图看出,Shuffle过程横跨Map与Reduce两端,所以下面我也会分两部分来展开。

先看看Map端的情况,如下图:

上图可能是某个MapTask的运行情况。拿它与官方的左半边比较,会发现很多不一致。官方图没有清楚的说明Partition,Sort与Combiner到底作用在哪个阶段。我画了这张图,希望让大家清晰地了解从Map数据输入到Map端所有数据准备好的全过程。

整个流程我分了四步。简单些可以这样说,每个MapTask都有一个内存缓冲区,存储着map()函数的输出结果,当缓冲区快满的时候需要将缓冲区的数据以一个临时文件的形式放到磁盘,当整个MapTask结束后再对磁盘中这个MapTask产生的所有临时文件做合并,生成最终的正式输出文件,然后等待ReduceTask来拉取数据。当然这里的每一步都可能包含着多个步骤与细节,下面我对细节来一一说明:

1.在MapTask执行时,它的输入数据来源于HDFS的block,当然在MapReduce概念中MapTask只读取split。Split与block的对应关系可能是多对一,默认是一对一。在WordCount例子里,假设的输入数据都是像"aaa"这样的字符串。

2.在经过Mapper的运行后,我们得知Mapper的输出是这样的一个Key/Value对:Key是"aaa",Value是数值1。因为当前Map端只做加1的操作,在ReduceTask里才会去合并结果集。前面我们知道这个Job有3个ReduceTask,到底当前的"aaa"应该交由哪个Reduce去做呢?是需要现在决定的。

MapReduce提供Partition接口,它的作用就是根据Key或Value及Reduce的数量来决定当前的这对输出数据最终应该交由哪个ReduceTask处理。默认对Key Hash然后再对ReduceTask数量取模。默认的取模方式知识为了平均Reduce的处理能力,如果用户自己对Partitioner有需求,可以订制并设置到Job上。

在我们的例子中,"aaa"经过Partitioner后返回0,也就是这对值应当交由第一个Reducer来处理。接下来,需要将数据写入内存缓冲区中,缓冲区的作用是批量收集Map结果,我们的Key/value对以及Partition的结果都会被写入缓冲区。当然写入之前,Key与Value值都会被序列化成字节数组。

整个内存缓冲区就是一个字节数组,它的字节索引及Key/Value存储结构我没有研究过。如果有朋友对它有研究,那么请大致描述一下它的细节吧。

3.这个内存缓冲区是有大小限制的,默认是100MB。当MapTask的输出结果很多时,就可能会撑爆内存,所以需要在一定条件下将缓冲区中的数据临时写入磁盘,然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程称为spill,中文可译为溢写,字面意思很直观。这个溢写是由单独线程来完成的,不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出,所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8,也就是当缓冲区的数据已经达到阈值(buffer size*spill percent=100MB*0.8=80MB),溢写线程启动,锁定这80MB的内存,执行溢写过程。MapTask的输出结果还可以往剩下的20MB内存中写,互不影响。

当溢写线程启动后,需要对着80MB空间内的Key做排序(Sort)。排序是MapReduce模型默认的行为,这里的排序也是对序列化的字节做的排序。

在这里我们可以想象,因为MapTask的输出是需要发送到不同的Reduce端去,而内存缓冲区没有对将发送到相同Reduce端的数据做合并,那么这种合并应该是体现是磁盘文件中的。从官方图上也可以看到写到磁盘中的溢写文件是对不同的Reduce端的数值做过合并。所以溢写过程一个很重要的细节在于,如果有很多的Key/Value对需要发送到某个Reduce端去,那么需要将这些Key/Value值拼接到一块,减少与Partion相关的索引记录。

在针对每个Reduce端而合并数据时,有些数据可能像这样:"aaa"/1,"aaa"/1.对于WordCount例子,就是简单地统计单词出现的次数,如果在同一个MapTask的结果中有很多个像"aaa"一样出现多次的Key,我们就应该把它们的值合并到一块,这个过程叫Reduce也叫Combine。但MapReduce的术语中,Reduce只指Reduce端执行从多个MapTask取数据做计算的过程。除Reduce外,非正式地合并数据只能算做Combine了,其实大家都知道,MapReduce中将Combine等同于Reducer。

如果Client设置过Combiner,那么现在就是使用Combiner的时候了。将有相同Key的Key/Value对的Value加起来,减少溢写到磁盘的数据量。Combiner会优化MapReduce的中间结果,所以它在整个模型中会多次使用。那哪些场景才能使用Combiner呢?从这里分析,Combiner的输出是Reducer的输入,Combiner决不能改变最终的计算结果。所以从我的想法来看,Combiner只应该用于那种Reduce的输入Key/Value与输出Key/Value类型完全一致,且不影响最终结果的场景。比如累加,最大值等。Combiner的使用一定要慎重,如果用好,它对Job执行效率有帮助,反之会影响Reduce的最终结果。

4.每次溢写会在磁盘上生成一个溢写文件,如果Map的输出结果真的很大,有多次这样的溢写发生,磁盘上相应的就会有多个溢写文件存在。当MapTask真正完成时,内存缓冲区中的数据也会全部溢写到磁盘中形成一个溢写文件。最终磁盘中会至少有一个这样的溢写文件存在(如果Map的输出结果很少,当Map执行完成时,只会产生一个溢写文件),因为最终的只有一个,所以需要将这些溢写文件归并到一起,这个过程叫做Merge。Merge是怎样的?如前面的例子,"aaa"从某个MapTask的一个溢写文件中读取过来时候值是5,从另外一个MapTask的溢写文件读取时值为8,因为它们有相同的Key,所以得merge成group。什么是group。对于"aaa"就是像这样得:{"aaa",[5,8,2,...]},数组中的值就是从不同溢写文件中读取出来的,然后再把这些值加起来。请注意,因为merge是将多个溢写文件合并到一个文件,所以可能也有相同的key存在,在这个过程中如果Client设置过Combiner,也会使用Combiner来合并相同的Key。

至此,Map端的所有工作都已结束,最终生成的这个文件也存放在TaskTracker够得着的某个本地目录内。每个ReduceTask不断地通过RPC从JobTracker那里获取MapTask是否完成的信息,如果Reduce Task得到通知,获知某个TaskTracker上的MapTask执行完成,Shuffle的后半段过程开始启动。

简单的说,ReduceTask在执行之前的工作就是不断拉取当前job里每个MapTask的最终结果,然后对从不地方拉取过来的数据不断地做merge,也最终形成一个文件作为ReduceTask的输入文件。见下图:

如Map端的细节图,Shuffle在Reduce端的过程也能用图上表明的三点来概括。当前Reduce copy数据的前提是它要从JobTracker获得要哪些MapTask以执行结束。Reduce真正运行之前,所有的时间都是在拉取数据,做merge,且不断重复地在做。如前面的方式一样,下面我也分段地描述Reduce端的Shuffle细节。

1.Copy过程,简单地拉取数据。Reduce进程启动一些数据Copy线程(Fetcher),通过HTTP方式请求MapTask所在的TaskTracker获取MapTask的输出文件。因为MapTask早已结束,这些文件就归TaskTracker管理在本地磁盘中。

2.Merge阶段。这里的merge如Map端的merge动作,只是数据中存放的是不同Map端copy来的数值。Copy过来的数据会先放入内存缓冲区中,这里的缓冲区大小要比Map端的更为灵活,它基于JVM的heap size设置,因为Shuffle阶段Reducer不运行,所以应该把绝大部分的内存都给Shuffle用。这里需要强调的是,merge有三种形式:1)内存到内存 2)内存到磁盘 3)磁盘到磁盘。默认情况下第一种形式不启用,让人比较困惑,是吧。当内存中的数据量达到一定阈值,就启动内存到磁盘的merge。与Map端类似,这也是溢写的过程,这个过程中如果你设置有Combiner,也是会启用的,然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行,直到没有map端的数据时才结束,然后启动第三种磁盘到磁盘的merge方式生成最终的那个文件。

3.Reducer的输入文件。不断的merge后,最后会生成一个"最终文件"。为什么加上引号?因为这个文件可能存在于磁盘上,也可能存在于内存中。对我们来说,当然希望它存放于内存中,直接作为Reducer的输入,但默认情况下,这个文件是存放于磁盘中的。至于怎样才能让这个文件出现在内存中,之后的性能调优我再说。当Reducer的输入文件已定,整个Shuffle才最终结束。然后Reducer执行,把结果放在HDFS上。

上面就是整个Shuffle的过程。细节很多,我很多都略过了,只试着把要点说明白。当然,我可能也有理解或表述上的很多问题,不吝指点,我希望不断完善和修改这篇文章,能让它通俗易懂,看完就能知道Shuffle的方方面面。至于具体的实现原理,各位有兴趣就自己去探索,如果不方便的话,留言给我,我再来研究并反馈。

时间: 2024-10-12 10:31:26

MapReduce Shuffle过程详解的相关文章

MapReduce阶段源码分析以及shuffle过程详解

MapReducer工作流程图: 1. MapReduce阶段源码分析 1)客户端提交源码分析 解释:   - 判断是否打印日志   - 判断是否使用新的API,检查连接   - 在检查连接时,检查输入输出路径,计算切片,将jar.配置文件复制到HDFS   - 计算切片时,计算最小切片数(默认为1,可自定义)和最大切片数(默认是long的最大值,可以自定义)   - 查看给定的是否是文件,如果是否目录计算目录下所有文件的切片   - 通过block大小和最小切片数.最大切片数计算出切片大小  

Hadoop MapReduce执行过程详解(带hadoop例子)

https://my.oschina.net/itblog/blog/275294 摘要: 本文通过一个例子,详细介绍Hadoop 的 MapReduce过程. 分析MapReduce执行过程 MapReduce运行的时候,会通过Mapper运行的任务读取HDFS中的数据文件,然后调用自己的方法,处理数据,最后输出.Reducer任务会接收Mapper任务输出的数据,作为自己的输入数据,调用自己的方法,最后输出到HDFS的文件中.整个流程如图: Mapper任务的执行过程详解 每个Mapper任

Hadoop学习之MapReduce执行过程详解

转自:http://my.oschina.net/itblog/blog/275294 分析MapReduce执行过程 MapReduce运行的时候,会通过Mapper运行的任务读取HDFS中的数据文件,然后调用自己的方法,处理数据,最后输出.Reducer任务会接收Mapper任务输出的数据,作为自己的输入数据,调用自己的方法,最后输出到HDFS的文件中.整个流程如图: Mapper任务的执行过程详解 每个Mapper任务是一个java进程,它会读取HDFS中的文件,解析成很多的键值对,经过我

MapReduce和spark的shuffle过程详解

面试常见问题,必备答案. 参考:https://blog.csdn.net/u010697988/article/details/70173104 mapReducehe和Spark之间的最大区别是前者较偏向于离线处理,而后者重视实效性,下面主要介绍mapReducehe和Spark两者的shuffle过程. MapReduce的Shuffle过程 MapReduce计算模型一般包括两个重要的阶段:Map是映射,负责数据的过滤分发:Reduce是规约,负责数据的计算归并.Reduce的数据来源于

MapReduce shuffle阶段详解

在Mapreduce中,Shuffle过程是Mapreduce的核心,它分布在Mapreduce的map阶段和reduce阶段,共可分为6个详细的阶段: 1).Collect阶段:将MapTask的结果输出到默认大小为100M的MapOutputBuffer内部环形内存缓冲区,保存的是key/value,Partition分区 2).Spill阶段:当内存中的数据量达到一定的阀值的时候,就会将数据写入本地磁盘,在将数据写入磁盘之前需要对数据进行一次排序的操作,先是对partition分区号进行排

MapReduce Shuffle过程

MapReduce Shuffle 过程详解 一.MapReduce Shuffle过程 1. Map Shuffle过程 2. Reduce Shuffle过程 二.Map Shuffle过程 1.   环形缓冲区 Map输出结果是先放入内存中的一个环形缓冲区,这个环形缓冲区默认大小为100M(这个大小可以在io.sort.mb属性中设置),当环形缓冲区里的数据量达到阀值时(这个值可以在io.sort.spill.percent属性中设置)就会溢出写入到磁盘,环形缓冲区是遵循先进先出原则,Ma

使用HeartBeat实现高可用HA的配置过程详解

使用HeartBeat实现高可用HA的配置过程详解 一.写在前面 HA即(high available)高可用,又被叫做双机热备,用于关键性业务.简单理解就是,有2台机器 A 和 B,正常是 A 提供服务,B 待命闲置,当 A 宕机或服务宕掉,会切换至B机器继续提供服务.常见的实现高可用的开源软件有 heartbeat 和 keepalived. 这样,一台 web 服务器一天24小时提供web服务,难免会存在 web 服务挂掉或服务器宕机宕机的情况,那么用户就访问不了服务了,这当然不是我们期望

Nginx实现集群的负载均衡配置过程详解

Nginx实现集群的负载均衡配置过程详解 Nginx 的负载均衡功能,其实实际上和 nginx 的代理是同一个功能,只是把代理一台机器改为多台机器而已. Nginx 的负载均衡和 lvs 相比,nginx属于更高级的应用层,不牵扯到 ip 和内核的修改,它只是单纯地把用户的请求转发到后面的机器上.这就意味着,后端的 RS 不需要配置公网. 一.实验环境 Nginx 调度器 (public 172.16.254.200 privite 192.168.0.48)RS1只有内网IP (192.168

加密 解密过程详解及openssl自建CA  

            加密 解密过程详解及openssl自建CA 为了数据信息能够安全的传输要求数据要有一定的安全性那么数据的安全性包含哪些方面的特性呢?    NIST(美国信息安全署)做了如下的定义:    保密性:       1,数据的保密性 指的是数据或隐私不向非授权者泄漏                   2,隐私性  信息不被随意的收集    完整性:       1,数据的的完整性:信息或程序只能被指定或授权的方式改变不能被随意的             修改