简单搞定Shuffle机制运行原理

1）mapreduce中，map阶段处理的数据如何传递给reduce阶段，是mapreduce框架中最关键的一个流程，这个流程就叫shuffle；

2）shuffle: 洗牌、发牌（核心机制：数据分区、排序、缓存）；

3）具体来说：就是将maptask输出的处理结果数据，分发给reducetask，并在分发的过程中，对数据按key进行了分区和排序。

2.4.2 Shuffle 结构

Shuffle缓存流程：

shuffle是MR处理流程中的一个过程，它的每一个处理步骤是分散在各个map task和reduce task节点上完成的。

2.4.3 partition分区

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

如果1

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

例如：假设自定义分区数为5，则

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

（2）job.setNumReduceTasks(2)会报错

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

2.4.5 Shuffle 运行机制

2）流程详解

上面的流程是整个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

2.4.6 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类型要对应起来

Mapper

这里的分片不是物理分片，输入分片存储的并非数据本身，而是一个分片长度和一个记录数据的位置的数据

例如10MB文件，切分10各1MB小文件，0-1MB位置定义为第一个切片，1MB-2MB定义为第二个分片

map阶段（每个分片对应一个map task）

每个mapper任务都是一个java进程，它读取hdfs文件中自己对应的输入分片，将分片中记录按照一定规则调用map函数解析成键值对，如<word,1>,<word,1>形式，如果有100个键值对，就调用100次map方法

键：每一行的起始位置            值：本行的文本内容

map shuffle阶段、洗牌阶段，分区partition,排序sort，combine(本地reducer),合并

map方法输出的数据，进入到内存缓冲区，缓冲区满了后，启动线程写磁盘，在启动线程写磁盘之前，对数据进行key的hash分区，对每个分区进行key值排脊早神序，设置了combiner，则对排序的数据做简单的合并重复key值 *** 作，如<word,2>,写磁盘 *** 作会产生多个文件，当map写完磁盘后则对文件进行一次合并，确保一个map task最终只生成一个数据文件

reduce shuffle阶段：copy、merge、reduce

map方法输出完成后，reduce线程会启动copy线程，请求所有map task的输出结果，如果reduce端接收的数据量小，则直接存内存中，数据量超过内存，则数据数据合并后写磁盘，在写磁盘过程中会把这些文件合并成一个更大的有序文件，最后一次合并的结果没有写磁盘，直接输入给reduce函数中

对copy过来的数据先睁配放入内存缓冲区中，如果是数据量超过缓冲区大小，则对数据合并后写磁盘，如果设置combiner，combiner也可以这个时候做合并，如果map task1中的<word,1>,map task2中的<word,2>,那么combiner之后则为<word,{1,2}>

调用reduce函数

reduce阶段分组好的<word,{1,2}>,调用reduce函数进行聚合<word,3>,将结果输出到hdfs，每个reduce进程会对应一个输出文件，名称以part-开头

词频统计mr过程：

split：由于输入文件太大，mapreduce会对其进行分割，大文件会被切分成多份

map：解析樱亏出每一行中的每个单词，并在后面记上数字1，表示此单词出现过1次

shuffle：将每一份中相同的单词分组到一起，并按默认字母顺序进行升序排序

reduce：将每一组中的单词出现的次数进行累加求和

以上复杂的运行过程，用一张图来简单说明，方便理解和记忆，如下图所示：