hadoop中map和reduce的数量设置

封云亭

　　hadoop中map和reduce的数量设置，有以下几种方式来设置
　　一、mapred-default.xml
　　

　　这个文件包含主要的你的站点定制的Hadoop。尽管文件名以mapred开头，通过它可以控制用户maps和 reduces的默认的设置。
　　
下面是一些有用变量：
　　


名字　　


含义　　

dfs.block.size　　


分布式文件系统中每个数据块的大小 （bytes）io.sort.factor　　


合并排序时每层输入的文件数　　

io.sort.mb　　


排序输入的reduce时缓存大小　　


io.file.buffer.size
用于缓存输入输出文件的字节数　　


mapred.reduce.parallel.copies
从map到reduce输出的线程数　　

dfs.replication　　


每个DFS块的备份数　　


mapred.child.java.opts
传给子任务jvm的选项　　


mapred.min.split.size
在一个map输入分裂的最小字节数　　


mapred.output.compress
Reduce的输出是否被压缩　　
二、不再使用模式设置，通过代码控制：
　　

　　
Mapper的数量在默认情况下不可直接控制干预，因为Mapper的数量由输入的大小和个数决定。在默认情况下，最终input占据了多少block，就应该启动多少个Mapper。如果输入的文件数量巨大，但是每个文件的size都小于HDFS的blockSize，那么会造成启动的Mapper等于文件的数量（即每个文件都占据了一个block），那么很可能造成启动的Mapper数量超出限制而导致崩溃。这些逻辑确实是正确的，但都是在默认情况下的逻辑。其实如果进行一些客户化的设置，就可以控制了。
　　
在Hadoop中，设置Map task的数量不像设置Reduce task数量那样直接，即：不能够通过API直接精确的告诉Hadoop应该启动多少个Map task。
　　
你也许奇怪了，在API中不是提供了接口org.apache.hadoop.mapred.JobConf.setNumMapTasks(int
n)吗？这个值难道不可以设置Map task的数量吗？这个API的确没错，在文档上解释”Note: This is only a hint tothe
framework.“，即这个值对Hadoop的框架来说仅仅是个提示，不起决定性的作用。也就是说，即便你设置了，也不一定得到你想要的效果。
　　
1. InputFormat介绍
　　

　　
在具体设置Map task数量之前，非常有必要了解一下与Map-Reduce输入相关的基础知识。
　　

　　
这个接口（org.apache.hadoop.mapred.InputFormat）描述了Map-Reduce
job的输入规格说明（input-specification），它将所有的输入文件分割成逻辑上的InputSplit，每一个InputSplit将会分给一个单独的mapper；它还提供RecordReader的具体实现，这个Reader从逻辑的InputSplit上获取inputrecords并传给Mapper处理。
　　

　　
InputFormat有多种具体实现，诸如FileInputFormat（处理基于文件的输入的基础抽象类）,
DBInputFormat（处理基于数据库的输入，数据来自于一个能用SQL查询的表）,KeyValueTextInputFormat（特殊的FineInputFormat，处理PlainText
File，文件由回车或者回车换行符分割成行，每一行由key.value.separator.in.input.line分割成Key和Value），CompositeInputFormat，DelegatingInputFormat等。在绝大多数应用场景中都会使用FileInputFormat及其子类型。
　　

　　
通过以上的简单介绍，我们知道InputFormat决定着InputSplit，每个InputSplit会分配给一个单独的Mapper，因此InputFormat决定了具体的Map task数量。
　　

　　
2. FileInputFormat中影响Map数量的因素
　　
在日常使用中，FileInputFormat是最常用的InputFormat，它有很多具体的实现。以下分析的影响Map数量的因素仅对FileInputFormat及其子类有效，其他非FileInputFormat可以去查看相应的getSplits(JobConf job, int numSplits) 具体实现即可。
　　请看如下代码段（摘抄自org.apache.hadoop.mapred.FileInputFormat.getSplits）
　　

    long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);long minSize = Math.max(job.getLong("mapred.min.split.size", 1), minSplitSize);　　

for (FileStatus file: files) {　　Path path
= file.getPath();　　FileSystem fs
= path.getFileSystem(job);if ((length != 0) && isSplitable(fs, path)) {long blockSize = file.getBlockSize();long splitSize = computeSplitSize(goalSize, minSize, blockSize);long bytesRemaining = length;while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {　　String[] splitHosts
= getSplitHosts(blkLocations,length-bytesRemaining, splitSize, clusterMap);　　splits.add(
new FileSplit(path, length-bytesRemaining, splitSize, splitHosts));　　bytesRemaining
-= splitSize;　　}
　　

if (bytesRemaining != 0) {　　splits.add(
new FileSplit(path, length-bytesRemaining, bytesRemaining, blkLocations[blkLocations.length-1].getHosts()));　　}
　　}
else if (length != 0) {　　String[] splitHosts
= getSplitHosts(blkLocations,0,length,clusterMap);　　splits.add(
new FileSplit(path, 0, length, splitHosts));　　}
else {//Create empty hosts array for zero length files　　splits.add(new FileSplit(path, 0, length, new String[0]));
　　}
　　}
　　

　　return splits.toArray(new FileSplit[splits.size()]);
　　

　　protected long computeSplitSize(long goalSize, long minSize, long blockSize) {
　　return Math.max(minSize, Math.min(goalSize, blockSize));
　　}
　　


　　totalSize：是整个Map-Reduce job所有输入的总大小。numSplits：来自job.getNumMapTasks()，即在job启动时用org.apache.hadoop.mapred.JobConf.setNumMapTasks(int n)设置的值，给M-R框架的Map数量的提示。goalSize：是输入总大小与提示Map task数量的比值，即期望每个Mapper处理多少的数据，仅仅是期望，具体处理的数据数由下面的computeSplitSize决定。minSplitSize：默认为1，可由子类复写函数protected void setMinSplitSize(long minSplitSize) 重新设置。一般情况下，都为1，特殊情况除外。minSize：取的1和mapred.min.split.size中较大的一个。blockSize：HDFS的块大小，默认为64M，一般大的HDFS都设置成128M。splitSize：就是最终每个Split的大小，那么Map的数量基本上就是totalSize/splitSize。接下来看看computeSplitSize的逻辑：首先在goalSize（期望每个Mapper处理的数据量）和HDFS的block>　　

　　
3. 如何调整Map的数量有了2的分析，下面调整Map的数量就很容易了。
　　

　　
3.1 减小Map-Reduce job 启动时创建的Mapper数量当处理大批量的大数据时，一种常见的情况是job启动的mapper数量太多而超出了系统限制，导致Hadoop抛出异常终止执行。解决这种异常的思路是减少mapper的数量。具体如下：
　　
3.1.1 输入文件size巨大，但不是小文件这种情况可以通过增大每个mapper的input
size，即增大minSize或者增大blockSize来减少所需的mapper的数量。增大blockSize通常不可行，因为当HDFS被hadoopnamenode
-format之后，blockSize就已经确定了（由格式化时dfs.block.size决定），如果要更改blockSize，需要重新格式化HDFS，这样当然会丢失已有的数据。所以通常情况下只能通过增大minSize，即增大mapred.min.split.size的值。
　　

　　
3.1.2
输入文件数量巨大，且都是小文件所谓小文件，就是单个文件的size小于blockSize。这种情况通过增大mapred.min.split.size不可行，需要使用FileInputFormat衍生的CombineFileInputFormat将多个inputpath合并成一个InputSplit送给mapper处理，从而减少mapper的数量。具体细节稍后会更新并展开。
　　

　　
3.2 增加Map-Reduce job 启动时创建的Mapper数量增加mapper的数量，可以通过减小每个mapper的输入做到，即减小blockSize或者减小mapred.min.split.size的值。