Hadoop中的MapReduce的shuffle过程及调优

待我称王封你为后i 2022-12-05 05:15 136阅读 0赞

## 概述 ##

Shuffle，即混洗、洗牌，顾名思义就是对数据打乱重新分配。Shuffle发生在Map输出至Reduce的输入过程之间。主要分为两部分

1.  Map任务输出的数据进行分组、合并、排序，并写入本地磁盘
2.  Reduce任务拉取数据进行合并、排序

## Shuffle过程 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lGX0xpMTIz_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lGX0xpMTIz_size_16_color_FFFFFF_t_70 1][]

### Map端 ###

*  Map端输出时，会先将数据写入内存的环形缓冲区，默认大小100M，可通过参数设置
 *  当缓冲区的内容大小达到阈值（默认0.8，可通过参数设置），便有一个后台线程将写入缓冲区的数据溢写到磁盘。溢写的过程中Map任务仍然可以写数据到缓冲区，一旦缓冲区写满，Map任务将会被阻塞，知道后台线程写磁盘结束
 *  数据溢写到磁盘时会计算输出key的分区，同一个分区的key分在一组并按照key排序，最后写入磁盘文件。如果设置了combiner，会在写磁盘前调用combiner函数进行聚合，目的是减少网络IO，且不会影响Reduce计算结果
 *  每一次溢写都会产生一个文件，Map结束后会产生多个文件，但最终会被合并成一个分区且有序的文件。合并文件时可通过参数设置合并的文件个数
 *  输出到磁盘的过程可以设置为压缩，默认不压缩，可通过参数设置

### 参数调优 ###

<table> 
 <tbody> 
 <tr> 
 <td>参数名</td> 
 <td>默认值</td> 
 <td>说明</td> 
 </tr> 
 <tr> 
 <td>mapreduce.task.io.sort.mb</td> 
 <td> 100MB </td> 
 <td> Map输出时所使用的内存缓冲区大小 </td> 
 </tr> 
 <tr> 
 <td>mapreduce.map.sort.spill.percent</td> 
 <td> 0.8 </td> 
 <td> Map输出溢写到磁盘的内存阈值 </td> 
 </tr> 
 <tr> 
 <td>mapreduce.task.io.sort.factor</td> 
 <td> 10 </td> 
 <td>排序文件时一次可以合并的流数</td> 
 </tr> 
 <tr> 
 <td>mapreduce.map.output.compress</td> 
 <td> false </td> 
 <td> Map输出是否压缩 </td> 
 </tr> 
 <tr> 
 <td>mapredcue.map.output.compress.codec</td> 
 <td> Org.apache.hadoop.io.compress.DefaultCodec </td> 
 <td> Map输出压缩的编解码器 </td> 
 </tr> 
 </tbody> 
</table>

### Reduce端 ###

*  Reduce端主要涉及复制和排序。一个reduce任务需要从多个map端拉取对应分区的数据，只要map任务完成，reduce即可开始复制数据，可通过参数设置复制的并行度，默认为5
 *  map任务完成后通过心跳统计欧诺个只Application Master，reduce端会有一个线程定期查询Application Master，以获取完成的map任务的位置，从而到对应位置的机器复制数据
 *  reduce复制数据将数据写入内存，可通过参数设置可用的内存比
 *  如果数据大小达到一定阈值（可通过参数设置），或者复制的文件数达到阈值（可通过参数设置）则将内存的数据合并并溢写到磁盘。如果设置了combiner，则会在写入磁盘之前调用combiner函数进行聚合以减少写入磁盘的数据量
 *  复制结束后将会对数据进行排序，如果溢写产生的多个文件，将会进行文件合并，每次合并的文件数可通过参数设置。经过多次合并后最后一次合并排序将会直接作为reduce任务的输入

### 参数调优 ###

<table> 
 <tbody> 
 <tr> 
 <td>参数</td> 
 <td>默认值</td> 
 <td>说明</td> 
 </tr> 
 <tr> 
 <td>mapreduce.reduce.shuffle.parallelcopies</td> 
 <td> 5 </td> 
 <td>并发复制的线程数</td> 
 </tr> 
 <tr> 
 <td>mapreduce.task.io.sort.factor</td> 
 <td> 10 </td> 
 <td> 排序文件时一次可以合并的流数 </td> 
 </tr> 
 <tr> 
 <td>mapreduce.reduce.shuffle.input.buffer.percent</td> 
 <td> 0.7 </td> 
 <td>shuffle的复制阶段，用来存放map输出缓冲区占reduce堆内存的百分比</td> 
 </tr> 
 <tr> 
 <td>mapreduce.reduce.shuffle.merge.percent</td> 
 <td> 0.66 </td> 
 <td>Reduce输出缓冲区的阈值，超过该比例将进行合并和溢写磁盘</td> 
 </tr> 
 <tr> 
 <td>mapreduce.reduce.merge.inmem.threshold</td> 
 <td> 1000 </td> 
 <td>阈值，当累计的map输出文件超过该值，进行合并和溢写磁盘，0或者复制意味着该参数无效，合并和溢写只由mapreduce.reduce.shuffle.merge.percent控制</td> 
 </tr> 
 <tr> 
 <td>mapreduce.reduce.input.buffer.percent</td> 
 <td> 0.0 </td> 
 <td>在reduce过程（开始reduce函数时），内存中保存map输出的空间占整个堆空间的比例。默认情况下，reduce任务开始前所有的map输出合并到磁盘。以便为reducer提供更多的内存，如果reducer需要的内存较少，可以增加此值以最小化磁盘访问次数</td> 
 </tr> 
 </tbody> 
</table>

参考：

\[1\] [https://www.cnblogs.com/duma/p/10704544.html][https_www.cnblogs.com_duma_p_10704544.html]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lGX0xpMTIz_size_16_color_FFFFFF_t_70]: /images/20221123/40bf67b3fef74afea39a1573842754f8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lGX0xpMTIz_size_16_color_FFFFFF_t_70 1]: /images/20221123/2d2b596b61e34b9caa0ebe3ebe27edfe.png
[https_www.cnblogs.com_duma_p_10704544.html]: https://www.cnblogs.com/duma/p/10704544.html