Hadoop-MapReduce-详细解析MapTask和ReduceTask工作机制-连载中

冷不防 2023-03-01 09:27 53阅读 0赞

# 详细解析MapTask和ReduceTask工作机制 #

### 文章目录 ###

*  详细解析MapTask和ReduceTask工作机制
 *   *  MapTask工作机制
     *  ReduceTask工作机制

## MapTask工作机制 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70]  
（1）**Read阶段**：MapTask通过`InputFormat`获得的`RecordReader`，从输入InputSplit中解析出一个个**key/value**。

（2）**Map阶段**：该节点主要是将解析出的**key/value**交给用户编写map()函数处理，并产生一系列新的**key/value**。

（3）**Collect收集阶段**：在用户编写map()函数中，当数据处理完成后，一般会调用OutputCollector.collect()输出结果。在该函数内部，它会将生成的key/value分区（调用Partitioner），并写入一个环形内存缓冲区中。

（4）**Spill阶段**：即"**溢写**"，当环形缓冲区满后，MapReduce会将数据写到本地磁盘上，生成一个临时文件。需要注意的是，将数据写入本地磁盘之前，先要对数据进行一次快速排序，并在必要时对数据进行合并、压缩等操作。

溢写阶段详情：

*  **步骤1**：利用快速排序算法对缓存区内的数据进行排序，排序方式是，先按照分区编号`Partition`进行排序，然后按照`key`进行排序。这样，经过排序后，数据以分区为单位聚集在一起，且同一分区内所有数据按照 key 有序。
 *  **步骤2**：按照分区编号由小到大，依次将每个分区中的数据，写入任务工作目录下的临时文`output/spillN.out（N表示当前溢写次数）`中。如果用户设置了 Combiner，则写入文件之前，对每个分区中的数据进行一次聚集操作。
 *  **步骤3**：将分区数据的元信息，写到内存索引数据结构`SpillRecord`中，其中每个分区的元信息包括，在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过**1MB**，则将内存索引写到文件`output/spillN.out.index`中。

（5）**Merge阶段**：当所有数据处理完成后，MapTask对所有临时文件进行一次合并，以确保最终只会生成一个数据文件。

当所有数据处理完后，MapTask会将所有临时文件合并成一个大文件，并保存到文件output/file.out中，同时生成相应的索引文件output/file.out.index。

在进行文件合并过程中，MapTask以分区为单位进行合并。对于某个分区，它将采用多轮递归合并的方式。每轮合并`mapreduce.task.io.sort.factor`（默认10）个文件，并将产生的文件重新加入待合并列表中，对文件排序后，重复以上过程，直到最终得到一个大文件。

让每个MapTask最终只生成一个数据文件，可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。

## ReduceTask工作机制 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 1]  
（1）**Copy阶段**：ReduceTask 从各个 MapTask 上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。

（2）**Merge阶段**：在远程拷贝数据的同时，ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。

（3）**Sort阶段**：按照MapReduce语义，用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。

（4）Reduce阶段：reduce()函数将计算结果写到HDFS上。

**1）设置ReduceTask并行度（个数）**

ReduceTask的并行度同样影响整个Job的执行并发度和执行效率，但与MapTask的并发数由切片数决定不同，ReduceTask数量的决定是可以直接手动设置：

// 默认值是1，手动设置为4
    job.setNumReduceTasks(4);

**2）实验：测试ReduceTask多少合适**

（1）实验环境：1个Master节点，16个Slave节点：CPU:8GHZ，内存: 2G

（2）实验结论：

表 改变ReduceTask （数据量为1GB）

<table> 
 <thead> 
  <tr> 
   <th>MapTask=16</th> 
   <th></th> 
   <th></th> 
   <th></th> 
   <th></th> 
   <th></th> 
   <th></th> 
   <th></th> 
   <th></th> 
   <th></th> 
   <th></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ReduceTask</td> 
   <td>1</td> 
   <td>5</td> 
   <td>10</td> 
   <td>15</td> 
   <td>16</td> 
   <td>20</td> 
   <td>25</td> 
   <td>30</td> 
   <td>45</td> 
   <td>60</td> 
  </tr> 
  <tr> 
   <td>总时间</td> 
   <td>892</td> 
   <td>146</td> 
   <td>110</td> 
   <td>92</td> 
   <td>88</td> 
   <td>100</td> 
   <td>128</td> 
   <td>101</td> 
   <td>145</td> 
   <td>104</td> 
  </tr> 
 </tbody> 
</table>

**3）注意事项**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70]: /images/20230209/868dccebd2f24ea7a9de9a1afae1ab58.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 1]: /images/20230209/1a7c6fea7c6a4dbbbcc34d768e6f0f80.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 2]: /images/20230209/8c3e72fe3fa54a76bb362bff5dc77aff.png