当前位置: 代码迷 >> 综合 >> Mapreduce Shuffle运行机制
  详细解决方案

Mapreduce Shuffle运行机制

热度:38   发布时间:2023-12-13 03:12:02.0

概念:

mapreduce中 map阶段处理的数据传递给reduce阶段是mp框架中最关键的一个流程 叫做shuffle

shuffle :洗牌  核心机制 :数据分区 排序 缓存

其实就是将maptask 输出的处理的结果数据分发给reducetask 分发的过程中进行了分区排序

shuffle 是MR处理中额一个流程  每一个处理步骤是分散在各个maptask 和reducetask上完成的

partition分区

如果reduceTask的数量> getPartition的结果数,则会多产生几个空的输出文件part-r-000xx

如果1<reduceTask的数量<getPartition的结果数,则有一部分分区数据无处安放,会Exception

如果reduceTask的数量=1,则不管mapTask端输出多少个分区文件,最终结果都交给这一个reduceTask,最终也就只会产生一个结果文件 part-r-00000

例如:假设自定义分区数为5,则

1job.setNumReduceTasks(1);会正常运行,只不过会产生一个输出文件

2job.setNumReduceTasks(2);会报错

3job.setNumReduceTasks(6);大于5,程序会正常运行,会产生空文件

shuffle运行机制

具体过程详解

上面的流程是整个mapreduce最全工作流程,但是shuffle过程只是从第7步开始到第16步结束,具体shuffle过程详解

1)maptask收集我们的map()方法输出的kv对,放到内存缓冲区中

2)从内存缓冲区不断溢出本地磁盘文件,可能会溢出多个文件

3)多个溢出文件会被合并成大的溢出文件

4)在溢出过程中,及合并的过程中,都要调用partitoner进行分组和针对key进行排序

5)reducetask根据自己的分区号,去各个maptask机器上取相应的结果分区数据

6)reducetask会取到同一个分区的来自不同maptask的结果文件,reducetask会将这些文件再进行合并(归并排序)

7)合并成大文件后,shuffle的过程也就结束了,后面进入reducetask的逻辑运算过程(从文件中取出一个一个的键值对group,调用用户自定义的reduce()方法)

3)注意

Shuffle中的缓冲区大小会影响到mapreduce程序的执行效率,原则上说,缓冲区越大,磁盘io的次数越少,执行速度就越快。

缓冲区的大小可以通过参数调整,参数:io.sort.mb  默认100M

 

输入分片:

我们输入N个文件,例如文件大小分别为3MB、65MB、127MB,而hdfs的块的自定义大小为64MB,那么这三个文件,就会被划分为1、2、2一共五个输入分片。每个分片(input split)对应一个map。而且每个map执行的数据大小不均,这个也是mapreduce优化计算的一个关键点。

map阶段:就是程序员编写好的map函数了,因此map函数效率相对好控制,而且一般map操作都是本地化操作也就是在数据存储节点上进行;

combiner阶段:合并

1)combiner是MR程序中Mapper和Reducer之外的一种组件

2)combiner组件的父类就是Reducer

3)combiner和reducer的区别在于运行的位置:

Combiner是在每一个maptask所在的节点运行

Reducer是接收全局所有Mapper的输出结果;

4)combiner的意义就是对每一个maptask的输出进行局部汇总,以减小网络传输量

6)combiner能够应用的前提是不能影响最终的业务逻辑,而且,combiner的输出kv应该跟reducer的输入kv类型要对应起来

7)combiner阶段不一定有,是可选的,combiner其实也是一种reduce操作,因此我们看见WordCount类里是用reduce进行加载的。Combiner是一个本地化的reduce操作,它是map运算的后续操作,主要是在map计算出中间文件前做一个简单的合并重复key值的操作,例如我们对文件里的单词频率做统计,map计算时候如果碰到一个hadoop的单词就会记录为1,但是这篇文章里hadoop可能会出现n多次,那么map输出文件冗余就会很多,因此在reduce计算前对相同的key做一个合并操作,那么文件会变小,这样就提高了宽带的传输效率,毕竟hadoop计算力宽带资源往往是计算的瓶颈也是最为宝贵的资源,但是combiner操作是有风险的,使用它的原则是combiner的输入不会影响到reduce计算的最终输入,例如:如果计算只是求总数,最大值,最小值可以使用combiner,但是做平均值计算使用combiner的话,最终的reduce计算结果就会出错

 shuffle阶段:

将map的输出后进行shuffle,这个是mapreduce优化的重点地方。Shuffle一开始就是map阶段做输出操作,一般mapreduce计算的都是海量数据,map输出时候不可能把所有文件都放到内存操作,因此map写入磁盘的过程十分的复杂,更何况map输出时候要对结果进行排序,内存开销是很大的,map在做输出时候会在内存里开启一个环形内存缓冲区,这个缓冲区专门用来输出的,默认大小是100mb,并且在配置文件里为这个缓冲区设定了一个阀值,默认是80%,同时map还会为输出操作启动一个守护线程,如果缓冲区的内存达到了阀值的80%时候,这个守护线程就会把内容写到磁盘上,这个过程叫spill,另外的20%内存可以继续写入要写进磁盘的数据,写入磁盘和写入内存操作是互不干扰的,如果缓存区被撑满了,那么map就会阻塞写入内存的操作,让写入磁盘操作完成后再继续执行写入内存操作,前面我讲到写入磁盘前会有个排序操作,这个是在写入磁盘操作时候进行,不是在写入内存时候进行的,如果我们定义了combiner函数,那么排序前还会执行combiner操作。

每次split操作也就是写入磁盘操作时候就会写一个溢出文件,也就是说在做map输出有几次spill就会产生多少个溢出文件,等map输出全部做完后,map会合并这些输出文件。这个过程里还会有一个Partitioner操作,对于这个操作很多人都很迷糊,其实Partitioner操作和map阶段的输入分片(Input split)很像,一个Partitioner对应一个reduce作业,如果我们mapreduce操作只有一个reduce操作,那么Partitioner就只有一个,如果我们有多个reduce操作,那么Partitioner对应的就会有多个,Partitioner因此就是reduce的输入分片,这个程序员可以编程控制,主要是根据实际key和value的值,根据实际业务类型或者为了更好的reduce负载均衡要求进行,这是提高reduce效率的一个关键所在。到了reduce阶段就是合并map输出文件了,Partitioner会找到对应的map输出文件,然后进行复制操作,复制操作时reduce会开启几个复制线程,这些线程默认个数是5个,程序员也可以在配置文件更改复制线程的个数,这个复制过程和map写入磁盘过程类似,也有阀值和内存大小,阀值一样可以在配置文件里配置,而内存大小是直接使用reduce的tasktracker的内存大小,复制时候reduce还会进行排序操作和合并文件操作,这些操作完了就会进行reduce计算了。

reduce阶段:

和map函数一样也是程序员编写的,最终结果是存储在hdfs上的。

 

 

 

  相关解决方案