在上图中，我们假设此次mapreduce有多个mapper和2个reducer，p0 p1分别代表该数据应该分配到哪个reducer端。我将mapper的过程大致分为5个过程。

 

1.prepare Input。

Mapreduce程序都需要指定输入文件，输入的格式有很多种，最常见的是保存在hdfs 上的文本文件。在用户提交job到jobtrack(ResourceManager)前的job就会根据用户的输入文件计算出需要多少mapper，多 少reducer，mapper的输入InputSplit有多大，block块名称等。mapper在prepare input阶段只需要根据inputFormat类型创建对应的RecordReader打开对应的inputSplit分片即可。如果job配置了 combiner还需初始化combiner。代码见MapTask类run方法

 

2.mapper process

这里的mapper指用户使用或自己继承的mapper类，这也是所有初学mapreduce的同学首先看到的类。

可以看到mapper默认的map方法就是取出key，value并放到context对象中。context对象包装了一个内存中的buf，下面会介绍。

1. <span style=publicvoidthrows while   }</span>

run方法就是mapper实际运行的过程：不停的从context的inputSplit对象中取出keyvalue对，通过map方法处理再保存到context包装的内存buf中。

 

3.buffer in memery
key value在写入context中后实际是写入MapOutputBuffer类中。在第一个阶段的初始化过程中，MapOutputBuffer类会根据配置文件初始化内存buffer，我们来看下都有哪些参数：

1. <span style=
2. finalfloat
   float0.8
   finalint);
3. iffloat1.0float0.0
   thrownew

   if) != sortmb) {

4. thrownew

   ,

5. classclass), job);</span>

partition：mapper的数据需要分配到reduce端的个数，由用户的job指定，默认为1.

spillper：内存buf使用到此比例就会触发spill，将内存中的数据flush成一个文件。默认为0.8

sortmb：内存buf的大小，默认100MB

indexCacheMemoryLimit：内存index的大小。默认为1024*1024

sorter：对mapper输出的key的排序，默认是快排

 

内存buffer比较复杂，贴一张图介绍一下这块内存buf的结构：

当一对keyvalue写入时首先会从wrap
buf的右侧开始往左写，同时，会把一条keyvalue的meta信息（partition，keystart，valuestart）写入到最左边的
index区域。当wrap
buf大小达到spill的触发比例后会block写入，挖出一部分数据开始spill，直到spill完成后才能继续写，不过写入位置不会置零，而是类
似循环buf那样，在spill掉数据后可以重复利用内存中的buf区域。

 

这里单独讲一下partition：

1. <span style=
2. publicvoidthrows

   }</span>

1. <span style=
2. publicint
   int
   return
     }</span>

 

4.Partition Sort Store

在第四步中，partition是和sort一起做的，负责Spill的线程在拿到一段内存buf后会调用QuickSort的sort方法进行内存中的快排。

1. <span style=this, mstart, mend, reporter);</span>

排序的算法是先按keyvalue记录的partition排序后按key的compare方法：

1. <span style=publicintfinalintfinalint
   finalint
   finalint
   finalint
   finalint
2. if
   return
3. return

   }</span>

因此，mapper输出的keyvalue首先是按partition聚合。而我们如果指定key的compare方法会在这里生效并进行排序。最后，一次spill的输出文件类似下图。

在对内存中的buf排序后开始写文件。

如果job没有定义combiner则直接写文件，如果有combiner则在这里进行combine。
在生成spill文件后还会将此次spillRecord的记录写在一个index文件中。

 

5.merge

当mapper执行完毕后，就进入merge阶段。首先看下相关的配置参数：

1. <span style=int);</span>

mergeFactor：同时merge的文件数。

 

merge阶段的目的是将多个spill生成的中间文件合并为一个输出文件，这里的合并不同
于combiner，无论有没有配置combiner这里的merge都会执行。merge阶段的输出是一个数据文件
MapFinalOutputFile和一个index文件。看下相关代码：

说下merge的算法。每个spill生成的文件中keyvalue都是有序的，但不同的文
件却是乱序的，类似多个有序文件的多路归并算法。Merger分别取出需要merge的spillfile的最小的keyvalue，放入一个内存堆中，
每次从堆中取出一个最小的值，并把此值保存到merge的输出文件中。这里和hbase中scan的算法非常相似，在分布式系统中多路归并排序真是当红小
生啊！

这里merge时不同的partition的key是不会比较的，只有相同的partition的keyvalue才会进行排序和合并。最后的输出文件类似下图。

如果用户定义了combiner，在merge的过程中也会进行combine，因为虽然第
四步中combine过但那只是部分输入的combine，在merge时仍然需要combine。这里有人问了，既然这里有combiner，为啥在
spill输出时还要combine纳，我认为是因为每次combine都会大大减少输出文件的大小，spill时就combine能减少一定的IO操
作。

 

在merge完后会把不同partition的信息保存进一个index文件以便之后reducer来拉自己部分的数据。

1. <span style=
2. spillRec.putIndex(rec, parts);</span>

最后，我们再对mapper过程中的要点总结一下：

1.对map输出<key,value>的分区（partition）是在写入内存buf前就做好的了，方法是对key的hash。我们可以通过继承Partitioner类自己实现分区，将自己想要的数据分到同一个reducer中。

2.写入内存buf速度是非常快的，但spill过程会block写入。因此，对内存buf相关参数的调优是mapreduce调优的重点之一。

3.对数据的排序是基于MapOutKey排序的，因此，我们可以重载对应的方法实现customize的排序顺序

4.combine在spill和merge中都是进行。多次的combine会减少mapreduce中的IO操作，如果使用得当会很好的提高性能。但需要注意的是要深刻理解combine的意义，比如平均值就不适合用combine。