MapReduce详解

缺乏、安全感 2022-08-14 01:42 172阅读 0赞

**Hadoop -- MapReduce过程**

昨天我们对MapReduce作了大概了解，知道它如何进行数据处理。今天我们走进MapReduce，分析MapReduce源代码，看看它到底是如何实现的，我们会根据数据流动的顺序来进行分析 :-)

**1.读取数据（InputFormat）**

读取数据由InputFormat类完成。InputFormat类的功能如下：

*  验证作业的输入格式
 *  由InputSplit类将指定输入目录中的输入文件拆分成逻辑文件，即key/value序对，每个逻辑文件将被分配到一个Mapper中
 *  提供RecordReader实现类，从逻辑文件中搜集输入记录给Mapper类进行处理

**InputFormat**类 定义了两个抽象方法：getSplits和createRecordReader，前者根据作业（job）的配置，将输入文件切片并返回一个 InputSplit类型的数组；后者为InputSplit实例对象生成一个RecordReader对象，在InputSplit对象使用 前，MapReduce框架会先进行RecordReader的实例化操作。**InputFormat**类图如Figure1:

[![Hadoop -- MapReduce过程][Hadoop_wbr_--_wbr_MapReduce]][Hadoop_wbr_--_wbr_MapReduce_Hadoop_wbr_--_wbr_MapReduce]

Figure1：InputFormat类图

**InputSplit**类定义了两个抽象方法：getLength和getLocations，getLength方法获取分片的大小，因此能根据输入分片的大小进行排序；getLocations方法获取输入数据所在节点的名称。InputSplit类图如Figure2：

[![Hadoop -- MapReduce过程][Hadoop_wbr_--_wbr_MapReduce 1]][Hadoop_wbr_--_wbr_MapReduce_Hadoop_wbr_--_wbr_MapReduce 1]

Figure2：InputSplit类图

**RecordReader**类实现了**Closeable**接口，共6个抽象方法：initialize，nextKeyValue，getCurrentKey，getCurrentValue，getProgress和close。RecordReader类图如下：

[![Hadoop -- MapReduce过程][Hadoop_wbr_--_wbr_MapReduce 2]][Hadoop_wbr_--_wbr_MapReduce_Hadoop_wbr_--_wbr_MapReduce 2]

Figure3：RecordReader类图

*  initialize：调用RecordReader对象时进行实例化；
 *  nextKeyValue：读取下一个key/value对，读取成功返回true；
 *  getCurrentKey：获取当前key值，返回当前key值，如果没有key值返回null；
 *  getCurrentValue：获取当前value，即记录。
 *  getProgress：获取当前RecordReader处理数据的进度，返回数值为0.0~1.0
 *  close：关闭RecordReader

**RecordReader**将输入数据切片并转换成key/value对，该key/value对作为Mapper的输入。一般情况下，输入文件分片后是一组连续的记录，最少有N个分片（N是输入文件数）。用户需要控制好分片数，因为分片的大小和数量对作业的运作性能影响很大。

在使用**InputFormat**类时，需要指定数据的输入格式。Hadoop Map-Reduce框架提供了大量输入格式可供选择，主要区别在于文本输入格式（textual input format）和二进制输入格式（binary input format），下面列出几个较常用的格式：

*  KeyValueTextInputFormat：每行一对key/value序对；
 *  TextInputFormat：key值为行号，value为行的内容；
 *  NLineInputFormat：与KeyValueTextInputFormat类似，不同点是NLineInputFormat根据输入文件的行数分片，KeyValueTextInputFormat根据输入文件字节数分片；
 *  MultiFileInputFormat：一个抽象类，用户可以通过实现该类，将多个文件聚集到一个分片中；
 *  SequenceFileInputFormat：输入文件为Hadoop序列文件，其中包含序列化的key/value对。

KeyValueTextInputFormat和SequenceFileInputFormat格式是最常用的输入格式。可使用如下格式进行设置：

首先定义一个**JobConf**类对象（当然该对象需要初始化）：

JobConf conf；

然后通过conf调用setInputFormat设置输入数据的格式：

conf.setInputFormat(KeyValueTextInputFormat.class);

最后设置输入目录的路径：

FileInputFormat.setInputPaths(conf,MapReduceConfig.getInputDirectionary() );

由**InputFormat**类图可以看到，创建记录读取器时还需要**TaskAttemptContext**对象，获取分片时需要**JobContext**对 象，它们分别是获取任务（Task）和作业（Job）的信息。因为每一个文件的分片对应于一个任务（task），而getSplits方法返回的是 InputSplit类数组，用于完成整个作业（job），因此调用getSplits方法需要提供job的配置信息。这两个类我们后面再进行详细分析。 现在我们来总结一下读取数据需要做些什么：

1.  指定输入文件的目录
2.  指定一个类用于读取数据并将该数据转换成key/value对

到此读取数据阶段完成，接下来进入Map阶段，我们拭目以待吧 :-)

**2.Map阶段**

**   **完成读取数据操作后，现在我们可以将分片后的数据作为**Mapper**的输入，进入Map阶段。先来看看**Mapper**类中包含什么东西，如图Figure4：

[![Hadoop -- MapReduce过程][Hadoop_wbr_--_wbr_MapReduce 3]][Hadoop_wbr_--_wbr_MapReduce_Hadoop_wbr_--_wbr_MapReduce 3]

Figure4：Mapper类图

Maps是用于将输入记录转换成中间记录的一个个独立的任务。这些中间记录不需要与输入记录保持相同数据类型。一个给定的输入<key,value>对可能被映射（map）到0个或多个输出序对。

Hadoop Map-Reduce框架为每个由**InputFormat**中的**InputSplit**生成一个map任务。而Mapper的实现类可以通过**JobContext**类的getConfiguration方法访问job的Configuration。

Hadoop Map-Reduce框架首先调用setup(org.apache.hadoop.mapreduce.Mapper.Context)方法，接着为InputSplit中的每个key/value对调用map(Object, Object, Context)方法。最后调用cleanup(Context)方法。

现在我们来看看**Mapper**类中的这几个方法：

*  setup：任务开始时调用，不做任何事；
 *  map：接受一个分片中的key/value对输入就调用一次，生成另一种类型的key/value对作为输出，可重写；
 *  cleanup：任务结束时调用，不做任何事；
 *  run：按以此调用setup, map, cleanup方法，可通过重写来控制Mapper过程。

执行完Map过程后，所有具有输出key值相同的中间value随后会被Hadoop Map-Reduce框架进行组合，并被传输到**Reducer**类。用户可以通过**RewComparator**类来控制排序和组合过程。

**Mapper**操作的输出记录将会被拆分给每一个Reducer类，用户可自定义**Partitioner**来指派哪个keys（同时还有记录）被分配到哪个**Reducer**中。

用户可通过设置Job中方法setCombinerClass(Class)来执行本地中间输出的聚集操作。这可以减少数据由**Mapper**经**Reducer**转换的规模。

如果作业中没有reduces过程，**Mapper**的输出记录将直接被写到**OutputFormat**中而不需要根据keys进行排序。

通过上面的分析，大概知道Map阶段是怎么操作的，输入数据经过Map阶段后，转换成另一组的key/value序对。这些序对等待**Partitioner**类进行处理。

**3.Reduce阶段**

没错，Reduce阶段是可选的。当需要进行reduce操作时，Hadoop框架会为每一个具有相同key值的记录集合调用一次reduce方法。我们先来看看**Reducer**类，Figure5为**Reducer**类图：

[![Hadoop -- MapReduce过程][Hadoop_wbr_--_wbr_MapReduce 4]][Hadoop_wbr_--_wbr_MapReduce_Hadoop_wbr_--_wbr_MapReduce 4]

Figure5：Reducer类图

Reducer类也很简单，定义的方法名与**Mapper**一样。Reducer有3个主要阶段：

*  Shuffle：通过网络（HTTP协议）访问并复制Mapper的输出记录；
 *  Sort：对Reducer的所有输入根据其key值进行归并排序（不同的**Mapper**可能输出相同key值）；
 *  Secondary Sort：对具有相同key值的记录再进行一次排序，此阶段需要扩展key值并定义一个聚合比较器（grouping comparator），比较器聚合记录并决定哪些记录被分配到同一个reduce方法中。

可通过Job\#setGroupingComparatorClass（class）设置比较器，排序的顺序由Job\#setSortComparatorClass（class）控制。

具 体过程：输入数据经过Map阶段分片后，需要经过一次整理（更准确应该是归并）。Map阶段的输入记录根据key值被归并在一起：混洗（shuffle） 与排序（sort）。混洗（shuffle）操作将对前面map操作的输出记录进行再一次分片并排序，对于经过混洗（shuffle）操作的每一分片，排 序（sort）操作对其进行归并排序，使得具有相同key值的记录被归并在一起。所有Map阶段的输出记录都需要经过这个阶段。经过归并排序的记录将由 Reduce任务来读取。在这个过程中会使用到一个类**Partitioner**，Figure5为**Partitioner**的类图：

[![Hadoop -- MapReduce过程][Hadoop_wbr_--_wbr_MapReduce 5]][Hadoop_wbr_--_wbr_MapReduce_Hadoop_wbr_--_wbr_MapReduce 5]

Figure6：Partitioner类图

**Partitioner**类很简单，只定义了一个getPartition方法，用于划分键值空间（key space）。该方法将Map阶段的所有输出记录根据key值产生另一组分片，并将具有相同key的记录分配到同一分区。该函数返回分区号，也可以说是分区的ID。

**Partitioner**的规则如下：

<key1, value1> ->int

根据key1生成一个分区的ID，具有相同ID的记录被分配到同一个分区，被同一个**Reducer**进行处理。

大家看了**Mapper**和**Reducer**两个类后，可能觉得很奇怪，为什么两个类这么简单，而且定义的几个方法功能也很少，却完成这么复杂的功能呢？:-)

如果大家看一下源码就会很容易理解了，这里就以**Reducer**类为例进行说明：

**Reducer**类中定义了一个内部类**Context**，它继承了**ReduceContext**类，下面是**Context**类：

public class Context

extends ReduceContext<KEYIN,VALUEIN,KEYOUT,VALUEOUT> \{

public Context(Configuration conf, TaskAttemptID taskid,

RawKeyValueIterator input,

Counter inputKeyCounter,

Counter inputValueCounter,

RecordWriter<KEYOUT,VALUEOUT> output,

OutputCommitter committer,

StatusReporter reporter,

RawComparator<KEYIN> comparator,

Class<KEYIN> keyClass,

Class<VALUEIN> valueClass

) throws IOException, InterruptedException \{

super(conf, taskid, input, inputKeyCounter, inputValueCounter,

output, committer, reporter,

comparator, keyClass, valueClass);

这个**Context**类帮我们做了很多事情，为了能使这个内部类起作用，我们至少需要向HadoopMap-Reduce框架提供以下信息：

*  reduce的任务数（the number of reduce tasks）；如果为零，不执行Reduce阶段；
 *  包含reduce方法的类；
 *  reduce任务中输入的key/value类型，默认与reduce操作的输出类型相同；
 *  reduce任务的输出类型；
 *  reduce任务输出记录所存储的文件类型；

reduce的任务数决定了一个作业（job）中输出的文件数，合适的数字可以提高整个job的运行性能，另外，reduce的任务数是所能进行的最大reduce任务并行运算数量；排序所用时间由key的数量决定。可以通过以下方式设置reduce的任务数：

JobConf.setNumReduceTasks(num);

包含reduce方法的类设置如下：

JobConf.setReduceClass(Reducer.class);

Key/value的输入输出类型以及输出文件类型我们在后面的输出阶段在说。设置好后，我们就结束Reduce阶段了。

由**Context**类的构造函数可知，用户还可以进行其他的自定义设置，这在后面我们会谈到。

同样，**Mapper**类中也定义了一个名为**Context**的内部类，它继承了**MapContext**类。我们以后有时间再来看看**ReduceContext**和**MapContext**两个类。

**4.输出阶段**

输出阶段用户至少需要做2件事：

*  指定记录输出的目录；
 *  通知Hadoop Map-Reduce框架输出记录key/value对的类型；

e.g.

FileOutputFormat.setOutputPath(JobConf, getOutputDirectory());

JobConf.setOutputKeyClass(Text.class);

JobConf.setOutputValueClass(Text.class);

注意，这里假定Map阶段的输出记录的类型都是Text。如果需要修改作业（job）输出记录的类型，除了修改setOutputKeyClass中的参数外，还需要修改map方法的输出类型，使得两者的类型相同。通知上面设置后，数据会以指定的类型输出到指定的目录中。

到此mapreduce基本完成。

[Hadoop_wbr_--_wbr_MapReduce]: http://s1.sinaimg.cn/middle/61ef4925ga896052dc530&690
[Hadoop_wbr_--_wbr_MapReduce_Hadoop_wbr_--_wbr_MapReduce]: /images/20220810/fb458b089ecb40e194c80ccdcf9911ba.png
[Hadoop_wbr_--_wbr_MapReduce 1]: http://s16.sinaimg.cn/middle/61ef4925ga8960788f03f&690
[Hadoop_wbr_--_wbr_MapReduce_Hadoop_wbr_--_wbr_MapReduce 1]: /images/20220810/0be1850ce5fc443a9f156f3f17ed1dd7.png
[Hadoop_wbr_--_wbr_MapReduce 2]: http://s15.sinaimg.cn/middle/61ef4925ga8960a1f738e&690
[Hadoop_wbr_--_wbr_MapReduce_Hadoop_wbr_--_wbr_MapReduce 2]: /images/20220810/7f04415b6dda4e8a8e04e84a599f1541.png
[Hadoop_wbr_--_wbr_MapReduce 3]: http://s12.sinaimg.cn/middle/61ef4925ga8960de84cfb&690
[Hadoop_wbr_--_wbr_MapReduce_Hadoop_wbr_--_wbr_MapReduce 3]: /images/20220810/f34a7e6988a845618165f17d7bac6be1.png
[Hadoop_wbr_--_wbr_MapReduce 4]: http://s1.sinaimg.cn/middle/61ef4925ga89610539b50&690
[Hadoop_wbr_--_wbr_MapReduce_Hadoop_wbr_--_wbr_MapReduce 4]: /images/20220810/140d0d17447d4d9499d9385cf78f3821.png
[Hadoop_wbr_--_wbr_MapReduce 5]: http://s5.sinaimg.cn/middle/61ef4925g7742350c7f54&690
[Hadoop_wbr_--_wbr_MapReduce_Hadoop_wbr_--_wbr_MapReduce 5]: /images/20220810/97ddd991c64f45ccbcb410992c61fd17.png