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()的值为1getMinSplitSize(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的相关操作其实并不繁琐至少逻辑顺序是非常清晰明了的希望大家都能有所收获。