Hadoop任务优化建议 - 【Dr.Elephant系列文章-6】

使用Dr.Elephant来分析我们的任务，可以知道有哪些地方可以进行优化。

加速你的任务流程

对于特定的任务，最好有特定的参数配置。对于很多的应用场景来说，默认的任务配置并不能保证每个任务都有最好的性能。尽管对这些任务进行调优会花费一些时间，但是这些调优带来的性能提升是非常可观的。
有几个任务参数需要特别注意：mapper数量，reducer数量，io.*的配置，内存使用设置以及生成的文件数量。对这几个参数进行配置，让参数更适合当前的任务，可以极大的提升任务的执行性能。
Apache的官网中Hadoop Map/Reduce
Tutorial这篇文章提供很多详细且有用的调试建议，有兴趣的可以仔细阅读一下。

一些常规建议

逐步调优非常重要

对于Pig任务来说，如果使用默认参数来设置reducer的数量，这对任务的性能可能是致命的。一般来说，对每个Pig任务，都花一些时间来调优参数PARALLEL是非常值得做的。例如：
memberFeaturesGrouped = GROUP memberFeatures BY memberId PARALLEL 90;
文件数量 VS 块数量
过多的小文件对于HDFS平台的NameNode的存储来说，会造成很大的压力。NameNode使用大约70 byte来存储一个文件，使用60 byte来存储一个块。一般情况下，对于任务来说，使用一个较大的文件要比使用十个小文件的效率高一些。

Java任务内存管理

默认情况下，每个map/reduce作业会分配的最大可使用内存空间为2G。对于java任务，这2G的空间既包括1G的堆空间，又包括0.5-1G的非堆空间。对于有些任务来说，默认的空间分配可能是不满足使用的。下面列举了一些能够减少内存使用的技巧：

UseCompressedOops

32位系统中的JVM使用32bit的无符号整形来定位内存区域，最大可表示的堆空间为2^32-1=4GB。64位的JVM使用64bit的无符号long整形来表示内存某个位置，最大可表示的内存堆大小为2^64=16艾字节。虽然可表示的内存空间大大增加了，但是使用long代替int，保存这些long整形会比保存int占用的内存空间增大。大约需要使用原来空间的1.5倍。使用64位的JVM不再限制我们可使用的堆空间为1G，这个提升给程序设计带来了很大的便利。我们能适当的缩减64位系统在保存内存空间信息时的内存使用吗？答案是肯定的。最新的JVM支持在使用时添加选项CompressedOops，它的作用是在一些情况下使用32bit的空间代替64bit空间来保存内存定位信息。这在一定程度上减少了内存的使用，我们可以在应用程序中按照如下方式选择使用CompressedOops选项：
hadoop-inject.mapreduce.(map|reduce).java.opts=-Xmx1G -XX:+UseCompressedOops
azkaban默认会使用自定义的属性覆盖掉默认配置属性，而不是将自定义的部分添加到mapred-site.xml默认属性中。在设置了使用CompressedOops选项之后，我们需要确认CompressedOops选项和其他默认的配置都是有效的。需要确认的是：”-Xmx1G”是配置文件mapred-site.xml中的，而其他的配置文件是我们自定义的。

UseCompressedStrings

这个选项会将String类型的变量转化为byte[]类型来保存。如果一个任务中使用了大量的String类型变量，那么这个选项将会极大的节约内存使用。在参数mapreduce.(map|reduce).java.opts配置中添加-XX:+UseCompressedString就会激活这个选项。每个作业分配的虚拟内存空间是需要的物理内存空间的2.1倍。如果我们程序抛出以下错误：
Container [pid=PID,containerID=container_ID]
 is running beyond virtual memory limits. Current usage: 365.1 MB of 1
GB physical memory used; 3.2 GB of 2.1 GB virtual memory used. Killing
container
我们就可以使用这个选项来对程序进行优化。

重要的调优参数

Mapper

mapreduce.input.fileinputformat.split.minsize
这个参数表示输入到map中的每个文件块的大小的最小值。通过增加这个值的大小，可以增加每个map中输入文件块的大小，从而减少了map的数量。例如，设置mapreduce.input.fileinputformat.split.minsize的大小为4倍的HDFS块大小（dfs.blocksize）时，那么输入到每个map的数量就是4倍的dfs.blocksize，这样就减少了map的数量。如果把这个值设置为256，那么输入的文件大小就是268435456bit。
mapreduce.input.fileinputformat.split.maxsize
这个参数表示当使用CombinedFileInputFormat和MultiFileInputFormat参数时，输入到map的每个文件大小的最大值。当设置这个值小于HDFS的块大小（dfs.blocksize）时，可以增加mapper的数量。例如，我们设置mapreduce.input.fileinputformat.split.maxsize为dfs.blocksize的1/4，这样就将输入到每个map的文件大小限制为dfs.blocksize的1/4，就增加了map的数量。当使用CombineFileInputFormat选项时，任务就只会使用1个mapper来处理。

Reducer

mapreduce.job.reduce
reducer的数量是影响任务性能的最重要的因素之一。使用过少的reducer会造成每个作业的执行时间过长，使用过多的reducer同样会造成性能问题。针对每个特定的任务因地制宜的调整reducer的数量是一项艺术。下面列举出一些方法来帮助我们选择合适的reducer数量：
更多的reducer意味着namenode上有更多的文件。过多的文件有可能造成namenode死掉。因此，如果reduce的输出不打（小于512MB），那么请使用较少的reducer。
更多的reducer意味着每个reducer在处理数据的时间更短。如果使用的reducer数量过少，每个reducer作业消耗的时间会显著增加。reducer运行速度变快，就能处理更多的任务。
在任务中，洗牌（shuffling）操作的代价是很昂贵的。我们通过HDFS文件系统（FileSystem）的各个计数器（Counter）可以看到多少数据需要在各个节点之间进行转移。我们做了一个实验，当reducers的数量是20时，Counter如下：
FileSystemCouter:
FILE_BYTES_READ | 2950482442768
HDFS_BYTES_READ | 1223524334581
FILE_BYTES_WRITTEN | 5697256875163
可以看到，map输出的数据数量大约有5TB，洗牌的时间如下：
洗牌结束：
17-Aug-2010 | 13：32：05 | （1hrs，29mins，56sec）
排序结束：
17-Aug-2010 | 14：18：35 | （46mins，29sec）
通过这些数字可以看到，大约有5TB的数据花费了1个半小时的时间来进行洗牌，然后又花费了46分钟来进行排序。这个时间成本是巨大的！我们想让这些工作在5~15分钟左右完成。我们进行一个简单的计算：使用20个reducers会花费360分钟，那么使用200个reducers就只需要36分钟，使用400个reducers就只需要18分钟。数字计算会带来量化了的提升。将reducers的数量提高到500，可以看到结果为：
洗牌结束：
17-Aug-2010 | 16：32：32 | （12min，46sec）
排序结束：
17-Aug-2010 | 16：32：37 | （4sec）
效果很明显，通过提升reducer的数量可以带来时间消耗的减少。物极必反，如果洗牌的时间变得非常短，而且CPU的使用也非常少，那么说明reducer的数量就过多了。通过实验来决定合理的reducer数量是非常必要的。
mapreducer.job.reduce.slowstart.completedmaps
这个参数决定了在reducer开始执行之前，至少有多少比例的mapper必须执行结束。默认值是80%。对于很多的特定的任务，调整这个数字的大小可能会带来性能提升。决定这个数字的因素是：
每个reducer接收到多少数据
剩下的map每一个map作业需要花费的时间
如果map的输出数据量比较大，一般会建议让reducer提前开始执行去处理这些数据。如果map任务产生的数量不是很大，一般建议让reducer的执行时间开始的晚一些。对于洗牌过程的一个粗略的时间估计是：当所有的mapper结束后开始，到第一个reducer开始执行的时间为止。这是reducer获得所有的输入所需要的时间。一般认为，reducers开始执行时间是：最后一个map结束时间+洗牌时间。

压缩

mapreduce.map.output.compress
将这个参数置为True可以将map输出的数据进行压缩。这可以减少在多个节点之间传输的数据量，然后，我们必须确保压缩和解压的时间是要小于数据在节点之间的传输时间的，否则会有反效果。如果map输出的数据量很大，或者属于比较容易压缩的类型，那么将这个参数置为True是有必要的，这将减少洗牌的时间。如果map输出的数据量比较小，那么将这个参数置为False可以减少CPU压缩和解压的工作量。有一点需要注意，这个参数和mapreduce.output.fileoutputformat.compress不同，那个参数决定了在将任务的输出写回到HDFS时是否需要压缩。

内存

mapreduce.(map|reduce).memory.mb
新版的Hadoop中添加了对内存的限制参数。这让系统能更好的管理一个繁忙情况下的资源分配。默认情况下，Java的任务中的每个作业会使用1GB的堆大小，以及0.5-1G的非堆空间。因此，默认的mapreduce.(map|reduce).momory.mb设置为2GB。在一些情况下，这个内存大小是不够的。如果只是设置Java的参数-Xmx，任务会被杀死，需要同时设置参数mapreduce.(map|reduce).momory.mb才能有效的提升或者限制内存使用。

进阶

控制io.sort.record.percent的值
参数io.sort.record.percent决定使用多少的间接缓存空间来保存每条和每条记录的元信息。一般来说，很多现象可以表明这个参数的设置是不合理的。
假设在map作业中使用日志的xml的配置文件：

property	value
bufstart	45633950
bufend	68450908
kvstart	503315
kvend	838860
length	838860
io.sort.mb	256
io.sort.record.percent	.05

使用上面的数字：

property	value
io.sort.spill.percent (length-kvstart+kvend) / length)	.8
Size of meta data is (bufend-bufstat)	22816958 (MB)
Records in memory (length-(kvstart+kvend))	671087
Average record size (size/records)	34 Bytes
Record+Metadata	50 Bytes
Records per io.sort.mb (io.sort.mb/(record+metadata))	5.12 million
Metadata % in io.sort.mb ((records per io.sort.mb)*metadata/io.sort.mb)	.32

在256MB的空间中，我们可以保存很多的记录。我们应该设置io.sort.record.percent为0.32，而不是0.5。当使用0.5时，记录的元信息的缓存会比记录的缓存更大。
改变这个参数会让map运行的更快，而且会有更少的磁盘溢出发生。甚至减少了55%的文件溢出到磁盘的情况，减少了大约30%的CPU时间和30分钟的运行时间。
mapreduce.(map|reduce).speculative
这个参数决定了是否会有同样的map或reduce并发的执行。当数据倾斜发生时，有的mapper或reducer的运行时间会显著的增长。这时，可能不需要一些推测行为来阻止倾斜到部分map和reduce中。

作者简介：屈世超，对高并发系统设计开发感兴趣，现专注于大数据开发工作。曾任职小米科技公司服务端后台开发工程师，现担任EverString数据平台组高级开发工程师。