Hadoop Mapreduce shuffle 过程详解
转载自:https://my.oschina.net/u/2293326/blog/607540 欢迎阅读原创。
image.png
Map 端导读 :
1. map buffer
当map task开始运算，并产生中间数据时，其产生的中间结果并非直接就简单的写入磁盘。
这中间的过程比较复杂，并且利用到了内存buffer来进行已经产生的部分结果的缓存，并在内存buffer中进行一些预排序来优化整个map的性能。
如上图所示，每一个map都会对应存在一个内存buffer（MapOutputBuffer，即上图的buffer in memory），map会将已经产生的部分结果先写入到该buffer中，这个buffer默认是100MB大小，但是这个大小是可以根据job提交时的参数设定来调整的.
该参数即为：mapreduce.task.io.sort.mb
当map的产生数据非常大时，并且把mapreduce.task.io.sort.mb调大，那么map在整个计算过程中spill的次数就势必会降低，map task对磁盘的操作就会变少.
2. map spill size
map在运行过程中，不停的向该buffer中写入已有的计算结果，但是该buffer并不一定能将全部的map输出缓存下来，当map输出超出一定阈值（比如100M），那么map就必须将该buffer中的数据写入到磁盘中去，这个过程在mapreduce中叫做spill。
map并不是要等到将该buffer全部写满时才进行spill，因为如果全部写满了再去写spill，势必会造成map的计算部分等待buffer释放空间的情况。所以，map其实是当buffer被写满到一定程度（比如80%）时，就开始进行spill。
这个阈值也是由一个job的配置参数来控制，即mapreduce.map.sort.spill.percent，默认为0.80或80%
这个参数同样也是影响spill频繁程度，进而影响map task运行周期对磁盘的读写频率的。但非特殊情况下，通常不需要人为的调整。调整mapreduce.task.io.sort.mb对用户来说更加方便。
3. map spill file merge
当 map task 的计算部分全部完成后，如果map有输出，就会生成一个或者多个spill文件，这些文件就是map的输出结果。
map在正常退出之前(cleanup)，需要将这些spill合并（merge）成一个，所以map在结束之前还有一个merge的过程。
merge的过程中，有一个参数可以调整这个过程的行为，该参数为：mapreduce.task.io.sort.factor。该参数默认为10。
它表示当merge spill文件时，最多能有多少并行的stream向merge文件中写入。比如如果map产生的数据非常的大，产生的spill文件大于10，而mapreduce.task.io.sort.factor使用的是默认的10，那么当map计算完成做merge时，就没有办法一次将所有的spill文件merge成一个，而是会分多次，每次最多10个stream。
这也就是说，当map的中间结果非常大，调大mapreduce.task.io.sort.factor，有利于减少merge次数，进而减少map对磁盘的读写频率，有可能达到优化作业的目的。
4. map combiner
当job指定了combiner的时候，我们都知道map结束后会在map端根据combiner定义的函数将map结果进行合并。
运行combiner函数的时机有可能会是merge完成之前，或者之后，这个时机可以由一个参数控制，即mapreduce.map.combine.minspills（default 3）。
当job中设定了combiner，并且spill数大于等于3的时候，那么combiner函数就会在merge产生结果文件之前运行。通过这样的方式，就可以在spill非常多需要merge，并且很多数据需要做conbine的时候，减少写入到磁盘文件的数据数量，同样是为了减少对磁盘的读写频率，有可能达到优化作业的目的。
5. map output compress
减少中间结果读写进出磁盘的方法不止这些，还有就是压缩。
也就是说map的中间，无论是spill的时候，还是最后merge产生的结果文件，都是可以压缩的。压缩的好处在于，通过压缩减少写入读出磁盘的数据量。
对中间结果非常大，磁盘速度成为map执行瓶颈的job，尤其有用。控制map中间结果是否使用压缩的参数为：mapreduce.map.output.compress(true/false)。
将这个参数设置为true时，那么map在写中间结果时，就会将数据压缩后再写入磁盘，读结果时也会采用先解压后读取数据。
这样做的后果就是：写入磁盘的 中间结果数据量会变少，但是cpu会消耗一些用来压缩和解压。
所以这种方式通常适合job中间结果非常大，瓶颈不在cpu，而是在磁盘的读写的情况。说的 直白一些就是用cpu换IO。
根据观察，通常大部分的作业cpu都不是瓶颈，除非运算逻辑异常复杂。所以对中间结果采用压缩通常来说是有收益的。以下是一 个wordcount中间结果采用压缩和不采用压缩产生的map中间结果本地磁盘读写的数据量对比：
map中间结果不压缩：
image.png
map中间结果压缩：
image.png
可以看出，同样的job，同样的数据，在采用压缩的情况下，map中间结果能缩小将近10倍，如果map的瓶颈在磁盘，那么job的性能提升将会非常可观。
当采用map中间结果压缩的情况下，用户还可以选择压缩时采用哪种压缩格式进行压缩，现在hadoop支持的压缩格式有：GzipCodec，LzoCodec，BZip2Codec，LzmaCodec等压缩格式。
通常来说，想要达到比较平衡的cpu和磁盘压缩比，LzoCodec比较适合。但也要取决于job的具体情况。
用户若想要自行选择中间结果的压缩算法，可以设置配置参数：mapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.DefaultCodec或者其他用户自行选择的压缩方式。
6.Map side相关参数调优
| 选项 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| mapreduce.task.io.sort.mb | int | 100 | 缓存map中间结果的buffer大小(in MB) |
| io.sort.record.percent | float | 0.05 | io.sort.mb中用来保存map output记录边界的百分比，其他缓存用来保存数据 |
| io.sort.spill.percent mapreduce.map.sort.spill.percent | float | 0.80 | map开始做spill操作的阈值 |
| io.sort.factor mapreduce.task.io.sort.factor | int | 10 | 做merge操作时同时操作的stream数上限。 |
| min.num.spill.for.combine | int | 3 | combiner函数运行的最小spill数 |
| mapred.compress.map.output | boolean | false | map中间结果是否采用压缩 |
| mapred.map.output.compression.codec | class name | org.apache.hadoop.io.compress.DefaultCodec | map中间结果的压缩格式 |
Reduce 端导读 :
reduce的运行是分成三个阶段的。分别为 copy->sort->reduce。
由于job的每一个map都会根据reduce(n)数将数据分成map 输出结果分成n个partition，所以map的中间结果中是有可能包含每一个reduce需要处理的部分数据的。
所以，为了优化reduce的执行时间，hadoop中是等job的第一个map结束后，所有的reduce就开始尝试从完成的map中下载该reduce对应的partition部分数据。
这个过程就是通常所说的shuffle，也就是copy过程。
1. reduce shuffle parallelcopies
Reduce task在做shuffle时，实际上就是从不同的已经完成的map上去下载属于自己这个reduce的部分数据.
由于map通常有许多个，所以对一个reduce来说，下载也可以是并行的从多个map下载这个并行度是可以调整的，调整参数为：mapreduce.reduce.shuffle.parallelcopies（default 5）。
默认情况下，每个只会有5个并行的下载线程在从map下数据，如果一个时间段内job完成的map有100个或者更多，那么reduce也最多只能同时下载 5个map的数据，所以这个参数比较适合map很多并且完成的比较快的job的情况下调大，有利于reduce更快的获取属于自己部分的数据。
2. reduce merge
Reduce将map结果下载到本地时，同样也是需要进行merge的，所以mapreduce.task.io.sort.factor的配置选项同样会影响reduce进行merge时的行为，该参数的详细介绍上文已经提到，当发现reduce在shuffle阶段iowait非常的高的时候，就有可能通过调大这个参数来加大一次merge时的并发吞吐，优化reduce效率。
3. reduce spill
Reduce在shuffle阶段对下载来的map数据，并不是立刻就写入磁盘的，而是会先缓存在内存中，然后当使用内存达到一定量的时候才刷入磁盘。这个内存大小的控制就不像map一样可以通过 mapreduce.task.io.sort.mb 来设定了，而是通过另外一个参数来设置：mapreduce.reduce.shuffle.input.buffer.percent（default 0.7).
这个参数其实是一个百分比，意思是说，shuffile在reduce内存中的数据最多使用内存量为：0.7 × maxHeap of reduce task。
也就是说，如果该reduce task的最大heap使用量（通常通过mapreduce.reduce.java.opts来设置，比如设置为-Xmx1024m）的一定比例用来缓存数据。
默认情况下，reduce会使用其heapsize的70%来在内存中缓存数据。
如果reduce的heap由于业务原因调整的比较大，相应的缓存大小也会变大，这也是为什么reduce用来做缓存的参数是一个百分比，而不是一个固定的值了。
假设 mapreduce.reduce.shuffle.input.buffer.percent 为 0.7，reduce task的max heapsize为1G，那么用来做下载数据缓存的内存就为大概700MB左右，这700M的内存，跟map端一样，也不是要等到全部写满才会往磁盘刷的，而是当这700M中被使用到了一定的限度（通常是一个百分比），就会开始往磁盘刷。
这个限度阈值也是可以通过job参数来设定的，设定参数为：mapreduce.reduce.shuffle.merge.percent（default 0.66）。如果下载速度很快，很容易就把内存缓存撑大，那么调整一下这个参数有可能会对reduce的性能有所帮助。
当reduce将所有的map上对应自己partition的数据下载完成后，就会开始真正的reduce计算阶段（中间有个sort阶段通常时间非常短，几秒钟就完成了，因为整个下载阶段就已经是边下载边sort，然后边merge的）。
当reduce task真正进入reduce函数的计算阶段的时候，有一个参数也是可以调整reduce的计算行为。也就是：mapreduce.reduce.input.buffer.percent（default 0.0）。
由于reduce计算时肯定也是需要消耗内存的，而在读取reduce需要的数据时，同样是需要内存作为buffer，这个参数是控制，需要多少的内存百 分比来作为reduce读已经sort好的数据的buffer百分比。默认情况下为0，也就是说，默认情况下，reduce是全部从磁盘开始读处理数据。 如果这个参数大于0，那么就会有一定量的数据被缓存在内存并输送给reduce，当reduce计算逻辑消耗内存很小时，可以分一部分内存用来缓存数据， 反正reduce的内存闲着也是闲着。
4. reduce 相关参数调优
| 选项 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| mapred.reduce.parallel.copies | int | 5 | 每个reduce并行下载map结果的最大线程数 |
| io.sort.factor | int | 10 | 同上 |
| mapred.job.shuffle.input.buffer.percent | float | 0.7 | 用来缓存shuffle数据的reduce task heap百分比 |
| mapred.job.shuffle.merge.percent | float | 0.66 | 缓存的内存中多少百分比后开始做merge操作 |
| mapred.job.reduce.input.buffer.percent | float | 0.0 | sort完成后reduce计算阶段用来缓存数据的百分比 |
