你好,游客 登录 注册 搜索
背景:
阅读新闻

Hadoop旧mapreduce的map任务切分原理

[日期:2016-06-21] 来源:CSDN  作者:泰山不老生 [字体: ]

前言

最近在工作过程中接触一些Hive数据仓库中的表,这些表实际是从关系型数据库通过Sqoop抽到Hive的。在开发过程中对map任务的划分进行性能调优,发现mapreduce中关于FileInputFormat的参数调整都不起作用,最后发现这些老任务都是用旧版的mapreduce开发的,于是顺便研究下旧版mapreduce的任务划分策略。有关新版mapreduce的任务划分策略,大家可以参考我之前的博文《 Hadoop2.6.0的FileInputFormat的任务切分原理分析(即如何控制FileInputFormat的map任务数量) 》。

Hadoop

源码分析

根据《 Hadoop2.6.0的FileInputFormat的任务切分原理分析(即如何控制FileInputFormat的map任务数量) 》一文的内容,我们知道map任务的划分关键在于FileInputFormat 的getSplits方法的实现策略,现在我们来看看其源码:

public InputSplit[] getSplits(JobConf job, int numSplits)
    throws IOException {
    Stopwatch sw = new Stopwatch().start();
    FileStatus[] files = listStatus(job);
    
    // Save the number of input files for metrics/loadgen
    job.setLong(NUM_INPUT_FILES, files.length);
    long totalSize = 0;                           // compute total size
    for (FileStatus file: files) {                // check we have valid files
      if (file.isDirectory()) {
        throw new IOException("Not a file: "+ file.getPath());
      }
      totalSize += file.getLen();
    }

    long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
    long minSize = Math.max(job.getLong(org.apache.hadoop.mapreduce.lib.input.
      FileInputFormat.SPLIT_MINSIZE, 1), minSplitSize);

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

          long bytesRemaining = length;
          while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
            String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,
                length-bytesRemaining, splitSize, clusterMap);
            splits.add(makeSplit(path, length-bytesRemaining, splitSize,
                splitHosts[0], splitHosts[1]));
            bytesRemaining -= splitSize;
          }

          if (bytesRemaining != 0) {
            String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations, length
                - bytesRemaining, bytesRemaining, clusterMap);
            splits.add(makeSplit(path, length - bytesRemaining, bytesRemaining,
                splitHosts[0], splitHosts[1]));
          }
        } else {
          String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,0,length,clusterMap);
          splits.add(makeSplit(path, 0, length, splitHosts[0], splitHosts[1]));
        }
      } else { 
        //Create empty hosts array for zero length files
        splits.add(makeSplit(path, 0, length, new String[0]));
      }
    }
    sw.stop();
    if (LOG.isDebugEnabled()) {
      LOG.debug("Total # of splits generated by getSplits: " + splits.size()
          + ", TimeTaken: " + sw.elapsedMillis());
    }
    return splits.toArray(new FileSplit[splits.size()]);
  }

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

这里对以上代码的划分策略进行整理:

  1. 遍历当前作业的所有输入文件,然后将累积这些文件的字节数并保存到变量totalSize中;
  2. 如果用户指定了mapreduce.job.maps参数,那么这个参数会被保存在入参numSplits中;
  3. 用户想要通过 numSplits控制map任务的数量,那么需求对 totalSize进行平分,以便确定每个map任务划分的输入大小。这个计算很简单,即使用 totalSize除以numSplits,最后得到的目标划分大小存储在变量goalSize中;
  4. 常量SPLIT_MINSIZE实际是由参数mapreduce.input.fileinputformat.split.minsize来控制的,如果没有配置则默认是1。minSplitSize默认是1,切旧版FileIntputFormat没有设置此变量的地方。最后取 SPLIT_MINSIZE和 minSplitSize的最大值,并保存在变量minSize中;
  5. 遍历当前作业的每个输入文件,计算每个输入文件,将被划分的任务数量,最后将每个文件划分的任务数量合并起来就是整个作业划分的任务数量。
以上只是总体分析了作业的任务划分,有关每个输入文件的任务数量划分步骤如下:
  1. 判断文件的大小,只有文件字节数大于0才是有意义的;
  2. 判断文件是否是可以切分的,只有能够切分的文件才会继续进行任务数量划分;
  3. 调用文件的getBlockSize方法,获取文件的块大小并存储在变量blockSize中;
  4. 调用 computeSplitSize方法计算最后划分给每个任务的输入大小,并保存在splitSize中。计算公式为: splitSize =  max(minSize, min(goalSize, blockSize));
  5. 将文件按照 splitSize的大小进行划分,不足 splitSize大小的也算作一个任务划分数。

总结

根据以上分析发现旧版mapreduce和新版mapreduce的 FileIntputFormat关于map任务数量划分的实现逻辑不同,在对它们进行开发和性能优化时要特别注意。




收藏 推荐 打印 | 录入:elainebo | 阅读:
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数
点评:
       
评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款