MapReduce高级

shuffle阶段

概述

MapReduce会确保每个reducer的输入都是按键排序的。从map方法输出数据开始、到作为输入数据传给reduce方法的过程称为shuffle。在此，我们将学习shuffle是如何工作的，因为它有助于我们理解工作机制（如果需要优化MapReduce程序）。shuffle属于不断被优化和改进的代码库的一部分，因此会随着版本的不同，细节上可能会发生变量。不管怎样，从许多方面来看，shuffle是MapReduce的“心脏“，是奇迹发生的地方。

Hadoop权威指南

map端

​ map方法开始产生输出数据时，并不是简单地将它写到磁盘。这个过程非常复杂，它利用缓冲的方式写到内存并出于效率的考虑进行预排序。

​ 每个map任务都会有一个环形内存缓冲区用于存储map的输出数据。在默认情况下，缓冲区的大小为100MB,这个值可以通过mapreduce.task.io.sort.mb属性来调整。一旦缓冲区的内容达到阙值（默认是0.8，或者是80%，属性是mapreduce.map.sort.spill.percent），一个后台线程便开始把内容溢写(spill)到磁盘里，这个位置由属性mapreduce.cluster.local.dir来指定的。在将数据溢写到磁盘过程中，map的输出数据继续写到缓冲区，但如果在此期间缓冲区被填满，map会被阻塞直到写磁盘过程完成。

​ 在写磁盘之前，线程会根据分区器的逻辑把数据划分为不同的分区(partition)。然后，在每个分区中，后台线程会按键进行内存中排序（QuickSort，默认是字典顺序）。如果指定了一个combiner函数，它就在排序后的输出上运行。运行combiner函数使得map输出结果更紧凑，因此减少写到磁盘的数据和传递给reducer的数据。

​ 每次内存缓冲区达到溢出阖值，就会新建一个溢出文件(spill file),因此在map任务写完其最后一个输出记录之后，可能会有几个溢出文件。在MapTask任务完成之前，多个溢出文件被合并成一个已分区且已排序的输出文件。配置属性mapreduce.task.io.sort.factor控制着一次最多能合并多少个文件，默认值是10。

​ 如果至少存在3个溢出文件(通过mapreduce.map.combine.minspills属性设置)时，则combiner就会在输出文件写到磁盘之前再次运行。combiner可以在输入上反复运行，但并不影响最终结果。如果只有1或2个溢出文件，那么由于map输出规模减少，因而不值得调用combiner产生开销，因此不会为该map输出再次运行combiner。

​ 为了使写磁盘的速度更快，节约磁盘空间，并且减少传给reducer的数据量，在溢写到磁盘的过程中对数据进行压缩往往是个很好的主意。在默认情况下，输出是不压缩的，但只要将mapreduce.map.output, compress设置为true,就可以轻松启用此功能。使用的压缩库由mapreduce.map.output.compress.codec指定。

扩展 环形缓冲区的详解

Reduce端

​ reducer通过HTTP得到输出文件的分区。用于文件分区的工作线程的数量由任务的mapreduce. shuffle.max. threads属性控制，此设置针对的是每一个节点管理器，而不是针对每个map任务。

​ 现在转到处理过程的reduce部分。map输出文件位于运行MapTask的本地磁盘（注意，尽管map输出经常写到MapTask本地磁盘，但reduce输出并不这样）。现在，ApplicatioinMaster需要为分区文件运行reduce任务。并且，reduce任务需要集群上若干个map任务的map输出作为其特殊的分区文件。每个map任务的完成时间可能不同，因此在每个任务完成时，reduce任务就开始复制其输出。这就是reduce任务的复制阶段。reduce任务有少量复制线程，因此能够并行取得map输出。默认值是5个线程，但这个默认值可以修改设置mapreduce.reduce.shuffle. parallelcopies 属性即可。

reducer如何知道要从哪台机器取得map输出呢？

map任务成功完成后，它们会使用心跳机制通知它们的application master。因此，对于指定作业，application master知道map输出和主机位置之间的映射关系。reducer中的一个线程定期询问master以便获取map输出主机的位置，直到获得所有输出位置。

由于第一个reducer可能失败，因此主机并没有在第一个reducer检索到map输出时就立即从磁盘上删除它们。相反，主机会等待，直到application master告知它删除map输出，这是作业完成后执行的。
复制代码

​ 如果map输出相当小，会被复制到reduce任务JVM的内存（缓冲区大小由mapreduce.reduce.shuffle.input. buffer.percent 属性控制，指定用于此用途的堆空间的百分比），否则，map输出被复制到磁盘。一旦内存缓冲区达到阈值大小（由 mapreduce.reduce.shuffle.merge.percent 决定）或达到 map 输出阈值（由 mapreduce. reduce. merge. inmem .threshold 控制），则合并后溢出写到磁盘中。如果指定combiner,则在合并期间运行它以降低写入硬盘的数据量。

​ 随着磁盘上的溢写文件数量增多，后台线程会将它们合并为更大的、排好序的文件。这会为后面的合并节省一些时间。注意，为了合并，压缩的map输出（通过map任务）都必须在内存中被解压缩。

​ 复制完所有map输出后，reduce任务进入排序阶段（更恰当的说法是合并阶段，因为排序是在map端进行的），这个阶段将合并map输岀，维持其顺序排序。这是循环进行的。比如，如果有50个map输出，而合并因子是10（10为默认设置，由mapreduce.task. io.sort.factor,与 的合并类似），合并将进行 5 趟 ，每趟将10个文件合并成一个文件，因此最后有5个中间文件。

​ 在最后阶段，即reduce阶段，直接把数据输入reduce函数，从而省略了一次磁盘往返行程，并没有将这5个文件合并成一个已排序的文件作为最后一趟。最后的合并可以来自内存和磁盘片段。

1587656109344

shuffle流程总结

1. 从map函数输出到reduce函数接受输入数据，这个过程称之为shuffle.

2. map函数的输出，存储环形缓冲区（默认大小100M,阈值80M）

   环形缓冲区：其实是一个字节数组kvbuffer. 有一个sequator标记，kv原始数据从左向右填充(顺时针)，
   kvmeta是对kvbuffer的一个封装，封装成了int数组，用于存储kv原始数据的对应的元数据valstart，
   keystart，partition，vallen信息，从右向左(逆时针)。参考(环形缓冲区的详解一张)

3. 当达到阈值时，准备溢写到本地磁盘(因为是中间数据，因此没有必要存储在HDFS上)。在溢写前要进行对元数据分区(partition)整理，然后进行排序(quick sort,通过元数据找到出key，同一分区的所有key进行排序，排序完，元数据就已经有序了，在溢写时，按照元数据的顺序寻找原始数据进行溢写)

4. 如果有必要，可以在排序后，溢写前调用combiner函数进行运算，来达到减少数据的目的

5. 溢写文件有可能产生多个，然后对这多个溢写文件进行再次合并(也要进行分区和排序)。当溢写个数>=3时，可以再次调用combiner函数来减少数据。如果溢写个数<3时，默认不会调用combiner函数。

6. 合并的最终溢写文件可以使用压缩技术来达到节省磁盘空间和减少向reduce阶段传输数据的目的。（存储在本地磁盘中）

7. Reduce阶段通过HTTP写抓取属于自己的分区的所有map的输出数据(默认线程数是5，因此可以并发抓取)。

8. 抓取到的数据存在内存中，如果数据量大，当达到本地内存的阈值时会进行溢写操作，在溢写前会进行合并和排序(排序阶段)，然后写到磁盘中，

9. 溢写文件可能会产生多个，因此在进入reduce之前会再次合并(合并因子是10),最后一次合并要满足10这个因子，同时输入给reduce函数，而不是产生合并文件。reduce函数输出数据会直接存储在HDFS上。
复制代码

shuffle整体流程图

combiner函数

集群的可用带宽本来就很稀缺，因此在不影响结果数据的前提下，尽可能的减少磁盘IO和网络传输，是非常合适的。Hadoop允许用户针对map任务的输出指定一个combiner函数（其实是一个运行在map端的reduce函数），用于优化MR的执行效率。

特点总结：

1. Combiner是MR程序中Mapper和Reduce之外的一种组件

2. Combiner组件的父类就是Reducer

3. Combiner和Reducer之间的区别在于运行的位置

4. Reduce阶段的Reducer是每一个接收全局的Map Task 所输出的结果

5. Combiner是在合并排序后运行的。因此map端和reduce端都可以调用此函数。

6. Combiner的存在就是提高当前网络IO传输的性能，是MapReduce的一种优化手段。

7. Combiner在驱动类中的设置：

   job.setCombinerClass(MyCombiner.class);
复制代码

注意：combiner不適合做求平均值这类需求，很可能就影响了结果。

MapReduce参数优化

资源相关参数

以下参数是在用户自己的mr应用程序中配置在mapred-site.xml就可以生效

1. mapreduce.map.memory.mb: 一个Map Task可使用的资源上限（单位:MB），默认为1024。如果Map Task实际使用的资源量超过该值，则会被强制杀死。	
2. mapreduce.reduce.memory.mb: 一个Reduce Task可使用的资源上限（单位:MB），默认为1024。如果Reduce Task实际使用的资源量超过该值，则会被强制杀死。	
3. mapreduce.map.cpu.vcores: 每个Map task可使用的最多cpu core数目, 默认值: 1
4. mapreduce.reduce.cpu.vcores: 每个Reduce task可使用的最多cpu core数目, 默认值: 1
5. mapreduce.map.java.opts: Map Task的JVM参数，你可以在此配置默认的java heap size等参数.
   比如：
   -Xmx1024m -verbose:gc -Xloggc:/tmp/@taskid@.gc” （@taskid@会被Hadoop框架自动换为相应的taskid）, 
     默认值: ""	
6. mapreduce.reduce.java.opts: Reduce Task的JVM参数,可以在此配置默认的java heap size等参数. 
   比如:
   “-Xmx1024m -verbose:gc -Xloggc:/tmp/@taskid@.gc”, 默认值: “”
复制代码

下面的配置，应该在yarn启动之前就配置在服务器的yarn-site.xml配置文件中才能生效

7. yarn.scheduler.minimum-allocation-mb	  1024   给应用程序container分配的最小内存
8. yarn.scheduler.maximum-allocation-mb	  8192	给应用程序container分配的最大内存
9. yarn.scheduler.minimum-allocation-vcores	1	
10. yarn.scheduler.maximum-allocation-vcores	32
11. yarn.nodemanager.resource.memory-mb   8192   每台NodeManager最大可用内存
12. yarn.nodemanager.resource.cpu-vcores    8    每台NodeManager最大可用cpu核数
复制代码

shuffle性能优化的关键参数，应在yarn启动之前就配置好

13. mapreduce.task.io.sort.mb   100         //shuffle的环形缓冲区大小，默认100m
14. mapreduce.map.sort.spill.percent   0.8    //环形缓冲区溢出的阈值，默认80%
复制代码

容错相关参数

1. mapreduce.map.maxattempts: 每个Map Task最大重试次数，一旦重试参数超过该值，则认为Map Task运行失败，默认值：4。
2. mapreduce.reduce.maxattempts: 每个Reduce Task最大重试次数，一旦重试参数超过该值，则认为Map Task运行失败，默认值：4。
3. mapreduce.map.failures.maxpercent: 当失败的Map Task失败比例超过该值时，整个作业则失败，默认值
为0. 如果你的应用程序允许丢弃部分输入数据，则该该值设为一个大于0的值，比如5，表示如果有低于5%的Map 
Task失败（如果一个MapTask重试次数超过mapreduce.map.maxattempts，则认为这个Map Task失败，其对应的输入数据将不会产生任何结果），整个作业仍认为成功。
    
4. mapreduce.reduce.failures.maxpercent: 当失败的Reduce Task失败比例超过该值时，整个作业则失败，默认值为0.
5. mapreduce.task.timeout: Task超时时间，经常需要设置的一个参数，该参数表达的意思为：如果一个task
在一定时间内没有任何进入，即不会读取新的数据，也没有输出数据，则认为该task处于block状态，可能是卡住
了，也许永远会卡住，为了防止因为用户程序永远block住不退出，则强制设置了一个该超时时间（单位毫秒），默
认是300000。如果你的程序对每条输入数据的处理时间过长（比如会访问数据库，通过网络拉取数据等），建议将该
参数调大，该参数过小常出现的错误提示  
是“AttemptID:attempt_14267829456721_123456_m_000224_0 Timed out after 300 sec  
        sContainer killed by the ApplicationMaster.”。
复制代码

本地运行MapReduce作业

设置以下几个参数:

mapreduce.framework.name=local

fs.defaultFS=local(file:///)
复制代码

效率和稳定性相关参数

1) mapreduce.map.speculative: 是否为Map Task打开推测执行机制，默认为false

2) mapreduce.reduce.speculative: 是否为Reduce Task打开推测执行机制，默认为false

3) mapreduce.job.user.classpath.first & MapReduce.task.classpath.user.precedence：当同一个class同时出现在用户jar包和hadoop jar中时，优先使用哪个jar包中的class，默认为false，表示优先使用hadoop jar中的class

4) mapreduce.input.fileinputformat.split.minsize: FileInputFormat做切片时的最小切片大小

5) mapreduce.input.fileinputformat.split.maxsize:  FileInputFormat做切片时的最大切片大小(切片的默认大小就等于blocksize，即 134217728)