Mapreduce shuffle和排序

Mapreduce为了确保每个reducer的输入都按键排序。系统执行排序的过程-----将map的输出作为输入传给reducer 称为shuffle。学习shuffle是如何工作的有助于我们理解mapreduce工作机制。shuffle属于hadoop不断被优化和改进的代码库的一部分。从许多方面看,shuffle是mapreduce的“心脏”,是奇迹出现的地方。

下面这张图介绍了mapreduce里shuffle的工作原理:

从图可以看出shuffle发生在map端和reduce端之间,将map端的输出与reduce端的输入对应。
map 端
map函数开始产生输出时,并不是简单地将它输出到磁盘。这个过程更复杂,利用缓冲的方式写到内存,并出于效率的考虑进行预排序。shuffle原理图就看出来。
每个map任务都有一个环形内存缓冲区,用于存储任务的输出。默认情况是100MB,可以通过io.sort.mb属性调整。一旦缓冲内容达到阀值(io.sort.spill.percent,默认0.80,或者80%),一个后台线程开始把内容写到磁盘中。在写磁盘过程中,map输出继续被写到缓冲区,但如果在此期间缓冲区被填满,map会阻塞直到写磁盘过程完成。在写磁盘之前,线程首先根据数据最终要传送到reducer把数据划分成相应的分区,在每个分区中,后台线程按键进行内排序,如果有一个combiner,它会在排序后的输出上运行。
reducer通过HTTP方式得到输出文件的分区。用于文件分区的工作线程的数量由任务的tracker.http.threads属性控制,此设置针对每个tasktracker,而不是针对每个map任务槽。默认值是40,在运行大型作业的大型集群上,此值可以根据需要调整。

reducer端

map端输出文件位于运行map任务的tasktracker的本地磁盘,现在,tasktracker需要为分区文件运行reduce任务。更进一步,reduce任务需要集群上若干个map任务完成,reduce任务就开始复制其输出。这就是reduce任务的复制阶段。reduce任务有少量复制线程,所以能并行取得map输出。默认值是5个线程,可以通过设置mapred.reduce.parallel.copies属性改变。

在这个过程中我们由于要提到一个问题,reducer如何知道要从那个tasktracker取得map输出呢?

map任务成功完成之后,它们通知其父tasktracker状态已更新,然后tasktracker通知jobtracker。这些通知都是通过心跳机制传输的。因此,对于指定作业,jobtracker知道map输出和tasktracker之间的映射关系。reduce中的一个线程定期询问jobtracker以便获得map输出的位置,直到它获得所有输出位置。
由于reducer可能失败,因此tasktracker并没有在第一个reducer检索到map输出时就立即从磁盘上删除它们。相反,tasktracker会等待,直到jobtracker告知它可以删除map输出,这是作业完成后执行的。

如果map输出相当小,则会被复制到reduce tasktracker的内存(缓冲区大小由mapred.job.shuffle.input.buffer.percent属性控制),否则,map输出被复制到磁盘。一旦内存缓冲区达到阀值大小(由mapred.job.shuffle.merge.percent决定)或达到map输出阀值(mapred.inmem.merge.threshold控制),则合并后溢出写到磁盘中。

随着磁盘上副本的增多,后台线程会将它们合并为更大的、排好序的文件。这会为后面的合并节省一些时间。注意,为了合并,压缩的map输出都必须在内存中被解压缩。

复制完所有map输出被复制期间,reduce任务进入排序阶段(sort phase 更恰当的说法是合并阶段,因为排序是在map端进行的),这个阶段将合并map输出,维持其顺序排序。这是循环进行的。比如,如果有50个map输出,而合并因子是10 (10默认值设置,由io.sort.factor属性设置,与map的合并类似),合并将进行5趟。每趟将10个文件合并成一个文件,因此最后有5个中间文件。
在最后阶段,即reduce阶段,直接把数据输入reduce函数,从而省略了一次磁盘往返行程,并没有将5个文件合并成一个已排序的文件作为最后一趟。最后的合并既可来自内存和磁盘片段。

在reduce阶段,对已排序输出中的每个键都要调用reduce函数。此阶段的输出直接写到输出文件系统中。

时间: 2024-10-03 15:01:56

Mapreduce shuffle和排序的相关文章

MapReduce shuffle过程剖析及调优

更新记录 2017-07-18 初稿 MapReduce简介 在Hadoop MapReduce中,框架会确保reduce收到的输入数据是根据key排序过的.数据从Mapper输出到Reducer接收,是一个很复杂的过程,框架处理了所有问题,并提供了很多配置项及扩展点.一个MapReduce的大致数据流如下图: 更详细的MapReduce介绍参考Hadoop MapReduce原理与实例. Mapper的输出排序.然后传送到Reducer的过程,称为shuffle.本文详细地解析shuffle过

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

(转)MapReduce二次排序

一.概述 MapReduce框架对处理结果的输出会根据key值进行默认的排序,这个默认排序可以满足一部分需求,但是也是十分有限的.在我们实际的需求当中,往往有要对reduce输出结果进行二次排序的需求.对于二次排序的实现,网络上已经有很多人分享过了,但是对二次排序的实现的原理以及整个MapReduce框架的处理流程的分析还是有非常大的出入,而且部分分析是没有经过验证的.本文将通过一个实际的MapReduce二次排序例子,讲述二次排序的实现和其MapReduce的整个处理流程,并且通过结果和map

Map、Reduce任务中Shuffle和排序的过程

Map.Reduce任务中Shuffle和排序的过程 流程分析: Map端: 1.每个输入分片会让一个map任务来处理,默认情况下,以HDFS的一个块的大小(默认为64M)为一个分片,当然我们也可以设置块的大小.map输出 的结果会暂且放在一个环形内存缓冲区中(该缓冲区的大小默认为100M,由io.sort.mb属性控制),当该缓冲区快要溢出时(默认为缓冲区大小的 80%,由io.sort.spill.percent属性控制),会在本地文件系统中创建一个溢出文件,将该缓冲区中的数据写入这个文件.

mapreduce.shuffle set in yarn.nodemanager.aux-services is invalid

15/07/01 20:14:41 FATAL containermanager.AuxServices: Failed to initialize mapreduce.shuffle java.lang.IllegalArgumentException: The ServiceName: mapreduce.shuffle set in yarn.nodemanager.aux-services is invalid.The valid service name should only con

Hadoop学习笔记—11.MapReduce中的排序和分组

一.写在之前的 1.1 回顾Map阶段四大步凑 首先,我们回顾一下在MapReduce中,排序和分组在哪里被执行: 从上图中可以清楚地看出,在Step1.4也就是第四步中,需要对不同分区中的数据进行排序和分组,默认情况下,是按照key进行排序和分组. 1.2 实验场景数据文件 在一些特定的数据文件中,不一定都是类似于WordCount单次统计这种规范的数据,比如下面这类数据,它虽然只有两列,但是却有一定的实践意义. 3 3 3 2 3 1 2 2 2 1 1 1 (1)如果按照第一列升序排列,当

mapreduce 实现数子排序

设计思路: 使用mapreduce的默认排序,按照key值进行排序的,如果key为封装int的IntWritable类型,那么MapReduce按照数字大小对key排序,如果key为封装为String的Text类型,那么MapReduce按照字典顺序对字符串排序. 首先map阶段将输入的数字作为key,  并记录相同key出现的次数,在reduce阶段将输入的key作为输出的value,如果相同值存在多个,循环便利输出. 源数据:file1 2 32 654 32 15 756 65223 fi

MapReduce中的排序

hadoop的计算模型就是map/reduce,每一个计算任务会被分割成很多互不依赖的map/reduce计算单元,将所有的计算单元执行完毕后整个计算任务就完成了.因为计算单元之间互不依赖所以计算单元可以分配到不同的计算机上执行,这样就可以将计算压力平摊到多个机器上面.当然性能线性提高是有条件的,前提是计算任务所采用的算法必须能够适应map/reduce模式.例如对于海量数据排序任务来说,绝大多数的排序算法都是不适应map/reduce模式的,如堆排序,插入排序,冒泡排序都是不适用于map/re

Hadoop on Mac with IntelliJ IDEA - 10 陆喜恒. Hadoop实战(第2版)6.4.1(Shuffle和排序)Map端 内容整理

下午对着源码看陆喜恒. Hadoop实战(第2版)6.4.1  (Shuffle和排序)Map端,发现与Hadoop 1.2.1的源码有些出入.下面作个简单的记录,方便起见,引用自书本的语句都用斜体表示. 依书本,从MapTask.java开始.这个类有多个内部类: 从书的描述可知,collect()并不在MapTask类,而在MapOutputBuffer类,其函数功能是 1.定义输出内存缓冲区为环形结构2.定义输出内存缓冲区内容到磁盘的操作 在collect函数中将缓冲区的内容写出时会调用s