文章目录

• 1. Map阶段
• • 1.1 把输入文件(夹)划分为很多InputSplit(Split)
  • 1.2 分配并执行map作业
• 2. Shuffle阶段
• • 2.1 Partition(分区)
  • 2.2 Sort(排序)
  • 2.3 Group(分组)
  • 2.4 Combiner(规约)
  • 2.5 序列化并写入Linux磁盘内存
  • 2.6 反序列化读取数据到不同的reduce节点
  • 2.7 Reduce端数据进行合并、排序、分组
• 3. Reduce阶段
• • 3.1 执行reduce方法
  • 3.2 保存结果到HDFS

       MapReduce是一种分布式计算模型，是Google提出来的，主要用于搜索领域，解决海量数据的计算问题。我自己在学习的过程中遇到了很多疑问，例如：

• 为什么有的人博客中写：“Block默认大小为64M”，而我学的Block默认却是128M？
• Split的大小划分与Block有什么关系？
• Partition分区有什么用？为什么要分区？是如何进行分区的？
• Shuffle到底包含那些过程？是只包含从Map到Reduce的传输么？
• MapReduce到底有几个阶段？怎么每一篇博客的阶段划分都不一样？

       MapReduce的全套过程分为三个大阶段，分别是Map、Shuffle和Reduce。结合多篇资料，我最终确定划分11个小步骤来描述这个过程，在后续的内容中我也会结合一部分源码来进行剖析。

1. Map阶段

map就是对数据进行局部汇总，reduce就是对局部数据进行最终汇总。

1.1 把输入文件(夹)划分为很多InputSplit(Split)

       首先，一个Block的默认大小为128M，之所以会有博客写为64M，是因为在Hadoop2.x中修改了这个默认设置。Split的默认大小为128M，但并不是每一个Split都是128M，具体分析过程如下，请看如下源码：

SplitSize = Math.max(minSize, Math.min(maxSize, BlockSize));

       也就是说Split的默认大小取决于minSize、maxSize以及BlockSize这三个变量。
       其中minSize的相关源码为：Math.max(getFormatMinSplitSize(), getMinSplitSize(job))；这行源码中getFormatMinSplitSize()的值为1，getMinSplitSize(job)的值为0，因此minSize的值为1；
       maxSize的相关源码为：maxSize=getMaxSplitSize(job)=Long.MAX_VALUE，也就是说maxSize的值为Long.MAX_VALUE；
       BlockSize的值默认为128M。
       所以最终SplitSize=128M。

       上面说的是Split的默认大小与Block相同，都是128M，但并不是说一个Block就对应一个Split，这里仅描述大小关系。但是之所以说并不是每一个Split都是128M，因为文件总不可能都是128M的整数倍，那么多出的那一部分怎么处理？源码中会判断剩余待切分文件大小/splitsize是否大于SPLIT_SLOP（值为1.1），如果大于1.1，那么会继续切分；如果小于1.1，会将剩下的部分切到同一个Split。

举几个例子帮助理解：

1. 一个1G的文件，会产生多少个Split？
   Block块默认是128M，所以1G的文件会产生8个Block块，默认情况下Split的大小和Block块的大小一致，也就是8个Split。
2. 1000个文件，每个文件100KB，会产生多少个Split？
   一个文件，不管再小，都会占用一个Block，所以这1000个小文件会产生1000个Block，默认情况下Split的大小和Block块的大小一致，那最终会产生1000个Split。
3. 一个140M的文件，会产生多少个Split？
   这个有点特殊，140M的文件虽然会产生2个Block，但140M/128M=1.09375<1.1，所以这个文件只会产生一个Split，这个文件其实再稍微大1M就可以产生2个Split。

1.2 分配并执行map作业

       默认一个Split对应一个Map，框架调用Mapper类中的map(…)函数，map函数的输入是<k1,v1>，输出是<k2,v2>。

2. Shuffle阶段

       Shuffle是介于Map和Reduce之间的一个过程，可以分为Map端的shuffle和Reduce端的Shuffle。MapReduce中，Map阶段处理的数据如何传递给Reduce阶段，是框架中最关键的一个流程，这个流程就叫Shuffle。

2.1 Partition(分区)

       分区默认使用HashPartitioner，使用哈希方法对key进行分区，getPatition方法相关源码为：(key.hashcode()&Inyeger.MAX_VALUE)%numReduceTask，其中numReduceTask默认为1，而任何书向1取余都为0，因此默认只有一个分区，又因为一个分区对应一个Reduce任务，所以只有也一个Ruduce。若要提高并行度，增加Reduce任务数，只需要修改numReduceTask数值即可。
       但是使用这种哈希方法分区有可能会导致数据倾斜问题，就比如现在一个文件中包含100万条数据，每个数据都是一个十以内数字，其中数字5出现了900万次，现在设置numReduceTask为10，那么根据哈希方法分区，其中的900万条数据都被分到分区5对应的Reduce任务下，这无疑是严重的影响了系统的运行效率。这种问题的解法并不会在这里展开说明，后续博客中会更新相关内容。

2.2 Sort(排序)

       按照Key，采用字典顺序进行排序，Sort操作无论是否需要，在逻辑上都必须执行。

2.3 Group(分组)

       分组是根据map<key, value>中的key进行分组，目的是提高Reduce的并行度。

2.4 Combiner(规约)

       规约是可选操作，在map端输出中先做一次合并，相当于做了一个局部的reduce操作。规约操作会将map输出的<k1,v1>，<k1,v2>，<k2,v3>这样的数据转化为<k1,{v1,v2}>，<k2,{v3}>。

2.5 序列化并写入Linux磁盘内存

       序列化（Serialization）是指把结构化对象转化为字节流，当要在进程间传递对象或持久化对象的时候，就需要进行这个操作。这里进行序列化是为了把map的执行结果写入内存。

2.6 反序列化读取数据到不同的reduce节点

       反序列化（Deserialization）是序列化的逆过程，把字节流转为结构化对象，当要将接收到或从磁盘读取的字节流转换为对象，就要进行这个操作。这里进行反序列化是为了读取数据到不同的reduce节点。

2.7 Reduce端数据进行合并、排序、分组

       reduce端接收到的是多个map的输出，对多个map任务中相同分区的数据进行合并、排序、分组。虽然之前在map中已经做了排序和分组，这边也做这些操作其实并不重复，因为map端是局部的操作，而reduce端是全局的操作，之前是每个map任务内进行排序，是有序的，但是多个map任务之间就是无序的了。

3. Reduce阶段

3.1 执行reduce方法

       框架调用Reducer类中的reduce方法，reduce方法的输入是<k2,{v2}>，输出是<k3,v3>。一个<k2,{v2}>调用一次reduce函数。程序员需要覆盖reduce函数，实现具体的业务逻辑。

3.2 保存结果到HDFS

       框架会把reduce的输出结果保存到HDFS中。

       以上就是博客的全部内容，MapReduce的相关操作其实并不繁琐，至少逻辑顺序是非常清晰明了的，希望大家都能有所收获。