5.6 在Hadoop中控制Map的数量

2016-03-09 23:36:50 4,879 0

在WordCount案例中,我们通过job.setNumReduceTasks(nums);方法可以手工的指定reducer任务的数量。

不过悲剧的是,在默认情况下,Mapper任务的数量我们不能人工干预。

在默认情况下,最终input占据了多少block,就应该启动多少个Mapper。如果输入的文件数量巨大,但是每个文件的size都小于HDFS的blockSize,那么会造成启动的Mapper等于文件的数量(即每个文件都占据了一个block),那么很可能造成启动的Mapper数量超出限制而导致崩溃。

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

1. InputFormat介绍

这个接口(org.apache.hadoop.mapred.InputFormat)描述了Map-Reduce job的输入规格说明(input-specification)

其中有一点为:

Split-up the input file(s) into logical InputSplits, each of which is then assigned to an individual Mapper.


翻译成中文就是:

将所有的输入文件分割成逻辑上的InputSplit,每一个InputSplit将会分给一个单独的mapper。

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

2. FileInputFormat中影响Map数量的因素

在日常使用中,FileInputFormat是最常用的InputFormat,它有很多具体的实现。以下分析的影响Map数量的因素仅对 FileInputFormat及其子类有效,其他非FileInputFormat可以去查看相应的 getSplits 具体实现即可。

请看如下代码段(摘抄自org.apache.hadoop.mapreduce.lib.input.FileInputFormat.getSplits(JobContext),hadoop-2.6.0源代码):

public List<InputSplit> getSplits(JobContext job) throws IOException {
    Stopwatch sw = new Stopwatch().start();
    long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
    long maxSize = getMaxSplitSize(job);

    // generate splits
    List<InputSplit> splits = new ArrayList<InputSplit>();
    List<FileStatus> files = listStatus(job);
    for (FileStatus file: files) {
      Path path = file.getPath();
      long length = file.getLen();
      if (length != 0) {
        BlockLocation[] blkLocations;
        if (file instanceof LocatedFileStatus) {
          blkLocations = ((LocatedFileStatus) file).getBlockLocations();
        } else {
          FileSystem fs = path.getFileSystem(job.getConfiguration());
          blkLocations = fs.getFileBlockLocations(file, 0, length);
        }
        if (isSplitable(job, path)) {
          long blockSize = file.getBlockSize();
          long splitSize = computeSplitSize(blockSize, minSize, maxSize);

          long bytesRemaining = length;
          while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
            int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
            splits.add(makeSplit(path, length-bytesRemaining, splitSize,
                        blkLocations[blkIndex].getHosts(),
                        blkLocations[blkIndex].getCachedHosts()));
            bytesRemaining -= splitSize;
          }

          if (bytesRemaining != 0) {
            int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
            splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
                       blkLocations[blkIndex].getHosts(),
                       blkLocations[blkIndex].getCachedHosts()));
          }
        } else { // not splitable
          splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
                      blkLocations[0].getCachedHosts()));
        }
      } else { 
        //Create empty hosts array for zero length files
        splits.add(makeSplit(path, 0, length, new String[0]));
      }
    }
    // Save the number of input files for metrics/loadgen
    job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
    sw.stop();
    if (LOG.isDebugEnabled()) {
      LOG.debug("Total # of splits generated by getSplits: " + splits.size()
          + ", TimeTaken: " + sw.elapsedMillis());
    }
    return splits;
  }



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被hadoop namenode -format之后,blockSize就已经确定了(由格式化时dfs.block.size决定),如果要更改blockSize,需要重新格式化HDFS,这样当然会丢失已有的数据。所以通常情况下只能通过增大minSize,即增大mapred.min.split.size的值。

3.1.2 输入文件数量巨大,且都是小文件

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

3.2 增加Map-Reduce job 启动时创建的Mapper数量

增加mapper的数量,可以通过减小每个mapper的输入做到,即减小blockSize或者减小mapred.min.split.size的值。


参考:

http://yaseminavcular.blogspot.com/2011/06/how-to-set-number-of-maps-with-hadoop.html

http://svn.apache.org/repos/asf/hadoop/common/tags/release-2.6.0

http://blog.csdn.net/strongerbit/article/details/7440111