MapReduce性能调优记录

MapReduce原理

要知道怎么对MapReduce做业进行调优前提条件是须要对Map-Reduce的过程了然于胸。
Map-Reduce运行原理图：

Map Side

1.从磁盘读取数据并分片

默认每一个block对应一个分片，一个map task

2.进行map处理

运行自定义的map业务过程

3.输出数据到缓冲区中

map输出的数据并非直接写入磁盘的，而是会先存储在一个预约义的buffer中

四、分区、排序分组的过程

对map输出的数据进行分区，按照key进行排序和分组

五、归约（可选）

至关于本地端的reduce过程

六、合并写入磁盘

对map的最终数据进行merge以后输出到磁盘中等待shuffle过程

Reduce side

1.从map端复制数据

2.对数据进行合并

以上两个步骤即为shuffle过程

3.对数据进行排序

4.进行reduce操做

5.输出到磁盘

详细的过程将会在调优技巧中体现出来

最简单的调优方式

设置Combiner

Combiner在Map端提早进行了一次Reduce处理。
可减小Map Task中间输出的结果，从而减小各个Reduce Task的远程拷贝数据量，最终表现为Map Task和Reduce Task执行时间缩短。

选择合理的Writable类型

为应用程序处理的数据选择合适的Writable类型可大大提高性能。
好比处理整数类型数据时，直接采用IntWritable比先以Text类型读入在转换为整数类型要高效。
若是输出整数的大部分可用一个或两个字节保存，那么直接采用VIntWritable或者VLongWritable，它们采用了变长整型的编码方式，能够大大减小输出数据量。

做业级别调优

增长输入文件的副本数

假设集群有1个Namenode+8个Datanode节点，HDFS默认的副本数为3
那么map端读取数据的时候，在启动map task的机器上读取本地的数据为3/8，一部分数据是经过网络从其余节点拿到的
那么若是副本数设置为8会是什么状况？
至关于每一个子节点上都会有一份完整的数据，map读取的时候直接从本地拿，不须要经过网络这一层了

可是在实际状况中设置副本数为8是不可行的，由于数据自己很是庞大，副本数超过5对集群的磁盘就很是有压力了，因此这项设置须要酌情处理

该配置在hdfs-side.xml的dfs.replication项中设置

Map side tuning

InputFormat

这是map阶段的第一步，从磁盘读取数据并切片，每一个分片由一个map task处理

当输入的是海量的小文件的时候，会启动大量的map task，效率及其之慢，有效的解决方式是使用CombineInputFormat自定义分片策略对小文件进行合并处理
从而减小map task的数量，减小map过程使用的时间
详情请看：自定义分片策略解决大量小文件问题

另外，map task的启动数量也和下面这几个参数有关系：

• mapred.min.split.size：Input Split的最小值 默认值1
• mapred.max.split.size：Input Split的最大值
• dfs.block.size：HDFS 中一个block大小，默认值128MB

当mapred.min.split.size小于dfs.block.size的时候，一个block会被分为多个分片，也就是对应多个map task
当mapred.min.split.size大于dfs.block.size的时候，一个分片可能对应多个block，也就是一个map task读取多个block数据

集群的网络、IO等性能很好的时候，建议调高dfs.block.size
根据数据源的特性，主要调整mapred.min.split.size来控制map task的数量

Buffer

该阶段是map side中将结果输出到磁盘以前的一个处理方式，经过对其进行设置的话能够减小map任务的IO开销，从而提升性能

因为map任务运行时中间结果首先存储在buffer中,默认当缓存的使用量达到80%的时候就开始写入磁盘,这个过程叫作spill(溢出)
这个buffer默认的大小是100M能够经过设定io.sort.mb的值来进行调整

当map产生的数据很是大时，若是默认的buffer大小不够看，那么势必会进行很是屡次的spill，进行spill就意味着要写磁盘，产生IO开销
这时候就能够把io.sort.mb调大，那么map在整个计算过程当中spill的次数就势必会下降，map task对磁盘的操做就会变少
若是map tasks的瓶颈在磁盘上，这样调整就会大大提升map的计算性能

可是若是将io.sort.mb调的很是大的时候，对机器的配置要求就很是高，由于占用内存过大，因此须要根据状况进行配置

map并非要等到buffer所有写满时才进行spill，由于若是所有写满了再去写spill，势必会形成map的计算部分等待buffer释放空间的状况。
因此，map实际上是当buffer被写满到必定程度（好比80%）时，才开始进行spill
能够经过设置io.sort.spill.percent的值来调整这个阈值
这个参数一样也是影响spill频繁程度，进而影响map task运行周期对磁盘的读写频率

可是一般状况下只须要对io.sort.mb进行调整便可

Merge

该阶段是map产生spill以后，对spill进行处理的过程，经过对其进行配置也能够达到优化IO开销的目的

map产生spill以后必须将些spill进行合并,这个过程叫作merge
merge过程是并行处理spill的,每次并行多少个spill是由参数io.sort.factor指定的,默认为10个

若是产生的spill很是多，merge的时候每次只能处理10个spill，那么仍是会形成频繁的IO处理
适当的调大每次并行处理的spill数有利于减小merge数所以能够影响map的性能

可是若是调整的数值过大，并行处理spill的进程过多会对机器形成很大压力

Combine

咱们知道若是map side设置了Combiner，那么会根据设定的函数对map输出的数据进行一次类reduce的预处理
可是和分组、排序分组不同的是，combine发生的阶段多是在merge以前，也多是在merge以后

这个时机能够由一个参数控制：min.num.spill.for.combine，默认值为3
当job中设定了combiner，而且spill数最少有3个的时候，那么combiner函数就会在merge产生结果文件以前运行

例如，产生的spill很是多，虽然咱们能够经过merge阶段的io.sort.factor进行优化配置，可是在此以前咱们还能够经过先执行combine对结果进行处理以后再对数据进行merge
这样一来，到merge阶段的数据量将会进一步减小，IO开销也会被降到最低

输出中间数据到磁盘

这个阶段是map side的最后一个步骤，在这个步骤中也能够经过压缩选项的配置来获得任务的优化

其实不管是spill的时候，仍是最后merge产生的结果文件，都是能够压缩的
压缩的好处在于，经过压缩减小写入读出磁盘的数据量。对中间结果很是大，磁盘速度成为map执行瓶颈的job，尤为有用

控制输出是否使用压缩的参数是mapred.compress.map.output，值为true或者false
启用压缩以后，会牺牲CPU的一些计算资源，可是能够节省IO开销，很是适合IO密集型的做业（若是是CPU密集型的做业不建议设置）

设置压缩的时候，咱们能够选择不一样的压缩算法
Hadoop默认提供了GzipCodec，LzoCodec，BZip2Codec，LzmaCodec等压缩格式

一般来讲，想要达到比较平衡的cpu和磁盘压缩比，LzoCodec比较合适，但也要取决于job的具体状况
若是想要自行选择中间结果的压缩算法，能够设置配置参数：

mapred.map.output.compression.codec=org.apache.hadoop.io.compress.DefaultCodec
//或者其余用户自行选择的压缩方式

Map side tuning总结

从上面提到的几点能够看到，map端的性能瓶颈都是频繁的IO操做形成的，全部的优化也都是针对IO进行的，而优化的瓶颈又很大程度上被机器的配置等外部因素所限制

map端调优的相关参数：

选项	类型	默认值	描述
mapred.min.split.size	int	1	Input Split的最小值
mapred.max.split.size	int	.	Input Split的最大值
io.sort.mb	int	100	map缓冲区大小
io.sort.spill.percent	float	0.8	缓冲区阈值
io.sort.factor	int	10	并行处理spill的个数
min.num.spill.for.combine	int	3	最少有多少个spill的时候combine在merge以前进行
mapred.compress.map.output	boolean	false	map中间数据是否采用压缩
mapred.map.output.compression.codec	String	.	压缩算法

Reduce side tuning

Shuffle

1.Copy

因为job的每个map都会根据reduce(n)数将数据分红map 输出结果分红n个partition，因此map的中间结果中是有可能包含每个reduce须要处理的部分数据的
为了优化reduce的执行时间，hadoop中等第一个map结束后，全部的reduce就开始尝试从完成的map中下载该reduce对应的partition部分数据

在这个shuffle过程当中，因为map的数量一般是不少个的，而每一个map中又都有可能包含每一个reduce所须要的数据
因此对于每一个reduce来讲，去各个map中拿数据也是并行的，能够经过mapred.reduce.parallel.copies这个参数来调整，默认为5
当map数量不少的时候，就能够适当调大这个值，减小shuffle过程使用的时间

还有一种状况是：reduce从map中拿数据的时候，有可能由于中间结果丢失、网络等其余缘由致使map任务失败
而reduce不会由于map失败就永无止境的等待下去，它会尝试去别的地方得到本身的数据（这段时间失败的map可能会被重跑）
因此设置reduce获取数据的超时时间能够避免一些由于网络很差致使没法得到数据的状况
mapred.reduce.copy.backoff，默认300s
通常状况下不用调整这个值，由于生产环境的网络都是很流畅的

2.Merge

因为reduce是并行将map结果下载到本地，因此也是须要进行merge的，因此io.sort.factor的配置选项一样会影响reduce进行merge时的行为

和map同样，reduce下载过来的数据也是存入一个buffer中而不是立刻写入磁盘的，因此咱们一样能够控制这个值来减小IO开销
控制该值的参数为：
mapred.job.shuffle.input.buffer.percent，默认0.7，这是一个百分比，意思是reduce的可用内存中拿出70%做为buffer存放数据

reduce的可用内存经过mapred.child.java.opts来设置，好比置为-Xmx1024m，该参数是同时设定map和reduce task的可用内存，通常为map buffer大小的两倍左右

设置了reduce端的buffer大小，咱们一样能够经过一个参数来控制buffer中的数据达到一个阈值的时候开始往磁盘写数据：mapred.job.shuffle.merge.percent，默认为0.66

Sort

sort的过程通常很是短，由于是边copy边merge边sort的，后面就直接进入真正的reduce计算阶段了

Reduce

以前咱们说过reduc端的buffer，默认状况下，数据达到一个阈值的时候，buffer中的数据就会写入磁盘，而后reduce会从磁盘中得到全部的数据
也就是说，buffer和reduce是没有直接关联的，中间多个一个写磁盘->读磁盘的过程，既然有这个弊端，那么就能够经过参数来配置
使得buffer中的一部分数据能够直接输送到reduce，从而减小IO开销：mapred.job.reduce.input.buffer.percent，默认为0.0

当值大于0的时候，会保留指定比例的内存读buffer中的数据直接拿给reduce使用
这样一来，设置buffer须要内存，读取数据须要内存，reduce计算也要内存，因此要根据做业的运行状况进行调整

Reduce side tuning总结

和map阶段差很少，reduce节点的调优也是主要集中在加大内存使用量，减小IO，增大并行数

reduce调优主要参数：

选项	类型	默认值	描述
mapred.reduce.parallel.copies	int	5	每一个reduce去map中拿数据的并行数
mapred.reduce.copy.backoff	int	300	获取map数据最大超时时间
mapred.job.shuffle.input.buffer.percent	float	0.7	buffer大小占reduce可用内存的比例
mapred.child.java.opts	String	.	-Xmx1024m设置reduce可用内存为1g
mapred.job.shuffle.merge.percent	float	0.66	buffer中的数据达到多少比例开始写入磁盘
mapred.job.reduce.input.buffer.percent	float	0.0	指定多少比例的内存用来存放buffer中的数据

MapReduce tuning总结

Map Task和Reduce Task调优的一个原则就是
减小数据的传输量
尽可能使用内存
减小磁盘IO的次数
增大任务并行数
除此以外还有根据本身集群及网络的实际状况来调优

Map task和Reduce task的启动数

在集群部署完毕以后，根据机器的配置状况，咱们就能够经过必定的公式知道每一个节点上container的大小和数量

1.mapper数量

每一个做业启动的mapper由输入的分片数决定，每一个节点启动的mapper数应该是在10-100之间，且最好每一个map的执行时间至少一分钟
若是输入的文件巨大，会产生无数个mapper的状况，应该使用mapred.tasktracker.map.tasks.maximum参数肯定每一个tasktracker可以启动的最大mapper数，默认只有2
以避免同时启动过多的mapper

2.reducer数量

reducer的启动数量官方建议是0.95或者1.75*节点数*每一个节点的container数
使用0.95的时候reduce只须要一轮就能够完成
使用1.75的时候完成较快的reducer会进行第二轮计算，并进行负载均衡
增长reducer的数量会增长集群的负担，可是会获得较好的负载均衡结果和减低失败成本

一些详细的参数：

选项	类型	默认值	描述
mapred.reduce.tasks	int	1	reduce task数量
mapred.tasktracker.map.tasks.maximum	int	2	每一个节点上可以启动map task的最大数量
mapred.tasktracker.reduce.tasks.maximum	int	2	每一个节点上可以启动reduce task的最大数量
mapred.reduce.slowstart.completed.maps	float	0.05	map阶段完成5%的时候开始进行reduce计算

map和reduce task是同时启动的，很长一段时间是并存的
共存的时间取决于mapred.reduce.slowstart.completed.maps的设置
若是设置为0.6.那么reduce将在map完成60%后进入运行态

若是设置的map和reduce参数都很大，势必形成map和reduce争抢资源，形成有些进程饥饿，超时出错，最大的可能就是socket.timeout的出错

reduce是在33%的时候完成shuffle过程，因此确保reduce进行到33%的时候map任务所有完成，能够经过观察任务界面的完成度进行调整
当reduce到达33%的时候，map刚好达到100%设置最佳的比例，可让map先完成，可是不要让reduce等待计算资源

做者：@小黑