Hadoop之MapReduce工作原理

约定不等于承诺〃 2022-04-16 02:21 298阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0OTkzNjMx_size_16_color_FFFFFF_t_70][]

# Map阶段 #

①**输入****分片（****inputsplit****）**，这个时候也就是输入数据的时候，这时会进行会通过内部计算对数据进行**逻辑上的分片**。默认情况下这里的分片与HDFS中文件的分块是一致的。每一个逻辑上的分片也就对应着一个mapper任务。

②**Mapper**将切片的数据输入到map函数中进行处理。

③**Buffer**函数将结果输出首先放入buffer（缓冲区）中从而为后面的操作（也就是写入硬盘）做准备。这里着重介绍一下buffer这个东西。Buffer从逻辑上来说是一个环形的缓冲区默认大小为100M，当第一次输入的数据超过它容量的80%这个阈值的时候（80%是它的溢写比）就会自动的写入硬盘。就在它写入硬盘的时候程序会继续往剩余的20%的空间中写数据。又由于80%的那些输入也会输入到磁盘中所以这样可以动态的写入更多的数据。但是如果一旦将整个缓冲区写满的话，写入程序将会挂起一直等到缓冲环中的数据都写完了写入程序将会再次启动开始向缓冲环中写数据。

⑤**溢写（spill）**，在达到缓冲环容量的阈值时，缓冲环中的内容会写到硬盘上。这个过程叫做溢写。这时会伴随着一些列的操作，分区、排序、合并的操作。又，每一个溢写操作会对应生成一个文件。也就是一个mapperTask会对应着多次溢写也就会处理多个文件。下面依次解释名词。

**分区（partition）**：默认的分区是hash分区这样会保证所有相同的key就会在存储在一起。当然我们可以基于各种意义进行分区。这样的话具有相同意义的kv对就会存储在一起。同时指出一个分区就会对应一个Reducer（不管key是否相同他们只要在同一个分区内那么就会被一个Reducer所获取）。

**排序（Sort）**：按照一定的规则**按照键排序**这样key相同的数据就会排在一起这时候partition内部会整体有序。按照字典顺序或者是数字大小顺序或者是我们自定义的顺序（这时必须实现Comparable接口实现compaleTo方法）。**由于这个原则我们在写MR程序的时候如果一个自定义的类充当了key值那么它一定要实现WritableComparable接口如果不实现的话就会因为找不到排序规则而出现如下异常。**

<table> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;width:426.1pt;"> <p style="margin-left:0cm;">java.lang.ClassCastException: class statisticScore.FlowInfo</p> <p style="margin-left:0cm;">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at java.lang.Class.asSubclass(Unknown Source)</p> <p style="margin-left:0cm;">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at org.apache.hadoop.mapred.JobConf.getOutputKeyComparator(JobConf.java:887)</p> <p style="margin-left:0cm;">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.init(MapTask.java:1004)</p> <p style="margin-left:0cm;">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at org.apache.hadoop.mapred.MapTask.createSortingCollector(MapTask.java:402)</p> <p style="margin-left:0cm;">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at org.apache.hadoop.mapred.MapTask.access$100(MapTask.java:81)</p> </td> 
  </tr> 
 </tbody> 
</table>

**合并（Combiner）：**这时程序在后台启动一个线程Combiner，提前将key值相同的数据合并起来可以减少往硬盘上写数据的量节省了空间，另一方面这些合并的数据会通过网络传输到Reducer相关的节点上这样的话就会节省网络的开销。Combiner的本质是一个Reducer它的合并操作也是一个Reduce操作。**但是****这样做的先决条件就是我们的任务在合并之后最终的结果不会改变。如果提前做****Reducer****优化而影响到结果计算的话那么就不能做提前优化。**

**⑥****归并**

将Splii写入磁盘形成的小文件合并成为一个大的文件。由于归并最后的文件的数据呈现形式不是前面的小文件简单的相加。所以这时又会将分将前面的分区，排序与合并。当然这里的排序与前面溢写时候的排序有所不同，**而是将前面的几个局部有序的文件合并成一个全局有序的大文件。**此时Combiner后台线程任然可能在做提前的优化合并。（读到这里有点费解到底是大文件好还是小文件好？）

# Reduce阶段 #

**⑦****获取数据（fetch）**reduce会主动向map阶段产生的数据中**获取属于自己的数据，也就是每个Reducer对应的那些特定分区（partitioner）的数据**。

**⑧****归并（merge）**将这些拿到零散的数据再次归并，与前面的归并不同的是这里的归并产生的并不是一个大文件，顺便这里产生的数据在内存和硬盘上都有一份。

**⑨分组（****group****）**，分组和分区是不同的**。分区的目的是根据****key****值来决定****Mapper****输出的记录会被哪一个****Reducer****所获取。分组常常是在分区内部进行的，在一个分区内相同的****key****可以分为一组。**

**⑩Reducer**将这些数据输入reduce函数中进行处理计算最后输出结果。有几个Reducer也就会对应的输出到几个文件中。

# Shuffle阶段 #

上述过程的③到⑨

# 数量关系的总结 #

*  一个Splie对应一个MapperTask。

*  一个MapperTask会有多次溢写，一次溢写对应着一个文件，所以一个MapperTask对应着多个文件。
 *  一个分区对应着一个Reducer，又一个Reducer对应着一个写出文件所以一个分区对应着一个写出文件。
 *  一个分区中可能存在多个key，一个key又可以划分为一个分组所以一个分区可以对应多个分组。
 *  在Reducer阶段的merge形成的数据在内存中和磁盘中都有所以有两份。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0OTkzNjMx_size_16_color_FFFFFF_t_70]: /images/20220416/6a169651314b42029e05c1c0d21da18b.png