MapReduce ：基于 FileInputFormat 的 mapper 数量控制

本篇分两部分，第一部分分析使用 java 提交 mapreduce 任务时对 mapper 数量的控制，第二部分分析使用 streaming 形式提交 mapreduce 任务时对 mapper 数量的控制。

 

环境：hadoop-3.0.2

前言：

熟悉 hadoop mapreduce 的人可能已经知道，即使在程序里对 conf 显式地设置了 mapred.map.tasks 或 mapreduce.job.maps，程序也并没有运行期望数量的 mapper。

这是因为，mapper 的数量由输入的大小、HDFS 当前设置的 BlockSize、以及当前配置中的 split min size 和 split max size 等参数共同确定，并不会受到简单的人工设置 mapper num 的影响。

因此，对于 mapper num 的控制，需要我们理解 hadoop 中对于 FileInputFormat 类中 getSplit() 方法的实现，针对性地配置 BlockSize、split min size、split max size 等参数，才能达到目的。

重点：

值得一提并且容易忽略的是，要区分 org.apache.hadoop.mapred.FileInputFormat类和 org.apache.hadoop.mapreduce.lib.input.FileInputFormat类，两者虽然相似，但在getSplit()上的实现是有区别的。

重要区别是，hadoop streaming 中使用的 InputFormat 类，使用的是 org.apache.hadoop.mapred.FileInputFormat，仅仅需要指定 mapreduce.job.maps ，就能够设置 mapper num了（具体源码分析在第二部分）。而使用JAVA设计的 mapreduce 任务中使用的 InputFormat 类，使用的是 org.apache.hadoop.mapreduce.lib.input.FileInputFormat，则需要通过配置BlockSize、split min size、split max size 等参数来间接性地控制 mapper num。

 

一、Java 本地提交 mapreduce 任务， org.apache.hadoop.mapreduce.lib.input.FileInputFormat 的 mapper num 控制

1. 在java本地编辑 mapreduce 任务，（默认）使用 FileInputFormat 类的子类 TextInputFormat

job.setInputFormatClass(TextInputFormat.class);

2. mapper 的切分逻辑在 FileInputFormat 类中的 getSplits(）实现：

public List<InputSplit> getSplits(JobContext job) throws IOException {
        StopWatch sw = (new StopWatch()).start();
        long minSize = Math.max(this.getFormatMinSplitSize(), getMinSplitSize(job));
        long maxSize = getMaxSplitSize(job);
        List<InputSplit> splits = new ArrayList();
        List<FileStatus> files = this.listStatus(job);
        Iterator var9 = files.iterator();

        while(true) {
            while(true) {
                while(var9.hasNext()) {
                    FileStatus file = (FileStatus)var9.next();
                    Path path = file.getPath();
                    long length = file.getLen();
                    if (length != 0L) {
                        BlockLocation[] blkLocations;
                        if (file instanceof LocatedFileStatus) {
                            blkLocations = ((LocatedFileStatus)file).getBlockLocations();
                        } else {
                            FileSystem fs = path.getFileSystem(job.getConfiguration());
                            blkLocations = fs.getFileBlockLocations(file, 0L, length);
                        }

                        if (this.isSplitable(job, path)) {
                            long blockSize = file.getBlockSize();
                            long splitSize = this.computeSplitSize(blockSize, minSize, maxSize);

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

                            if (bytesRemaining != 0L) {
                                blkIndex = this.getBlockIndex(blkLocations, length - bytesRemaining);
                                splits.add(this.makeSplit(path, length - bytesRemaining, bytesRemaining, blkLocations[blkIndex].getHosts(), blkLocations[blkIndex].getCachedHosts()));
                            }
                        } else {
                            if (LOG.isDebugEnabled() && length > Math.min(file.getBlockSize(), minSize)) {
                                LOG.debug("File is not splittable so no parallelization is possible: " + file.getPath());
                            }

                            splits.add(this.makeSplit(path, 0L, length, blkLocations[0].getHosts(), blkLocations[0].getCachedHosts()));
                        }
                    } else {
                        splits.add(this.makeSplit(path, 0L, length, new String[0]));
                    }
                }

                job.getConfiguration().setLong("mapreduce.input.fileinputformat.numinputfiles", (long)files.size());
                sw.stop();
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Total # of splits generated by getSplits: " + splits.size() + ", TimeTaken: " + sw.now(TimeUnit.MILLISECONDS));
                }

                return splits;
            }
        }
    }

3. 最后确定 mapper 数量在这里：

                        if (this.isSplitable(job, path)) {
                            long blockSize = file.getBlockSize();
                            long splitSize = this.computeSplitSize(blockSize, minSize, maxSize);

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

                            if (bytesRemaining != 0L) {
                                blkIndex = this.getBlockIndex(blkLocations, length - bytesRemaining);
                                splits.add(this.makeSplit(path, length - bytesRemaining, bytesRemaining, blkLocations[blkIndex].getHosts(), blkLocations[blkIndex].getCachedHosts()));
                            }

含义：

a. 当 this.isSplitable 开启时，只要当前未分配的大小 bytesRemaining 大于 splitSize 的 1.1 倍，就添加一个 inputSplit， 即一个mapper 被生成。 

b. 最后，不足 1.1 倍splitSize 的残余，补充为一个 mapper。因此，经常发现实际分配的 mapper 数比自己定义的会多 1 个。

c. 为什么设置1.1倍？避免将不足 0.1 倍 splitSize 的量分配为一个 mapper， 避免浪费。

4.  重要的两个量：BlockSize 和 splitSize

long blockSize = file.getBlockSize();
long splitSize = this.computeSplitSize(blockSize, minSize, maxSize);

其中，blockSize 是 hdfs 设置的，一般是 64MB 或 128MB，我的 hdfs 中为 128 MB = 132217728L。这个量可认为静态，我们不宜修改。

观察 splitSize 的获得：

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

在 getSplits()中找到 minSize， maxSize， blockSize 的赋值：

long minSize = Math.max(this.getFormatMinSplitSize(), getMinSplitSize(job));
long maxSize = getMaxSplitSize(job);

找到这些量的赋值、默认值：

maxSize 的 setter/getter:    除非用户重新设置，否则 maxSize 的默认值为 Long 的最大值 

public static void setMaxInputSplitSize(Job job, long size) {
        job.getConfiguration().setLong("mapreduce.input.fileinputformat.split.maxsize", size);
    }

public static long getMaxSplitSize(JobContext context) {
        return context.getConfiguration().getLong("mapreduce.input.fileinputformat.split.maxsize", 9223372036854775807L);
    }

minSize 的 setter/getter：  除非用户重新设置，否则 minSize 的默认值为 1L

protected long getFormatMinSplitSize() {
        return 1L;
    }

public static void setMinInputSplitSize(Job job, long size) {
        job.getConfiguration().setLong("mapreduce.input.fileinputformat.split.minsize", size);
    }

public static long getMinSplitSize(JobContext job) {
        return job.getConfiguration().getLong("mapreduce.input.fileinputformat.split.minsize", 1L);
    }

因此容易算出，默认情况下，

long splitSize = this.computeSplitSize(blockSize, minSize, maxSize) = Math.max(Math.max(1L,1L), Math.min(9223372036854775807L, 128M=132217728L)) = 132217728L = 128M

5. 控制 mapper 数量

知道了上面的计算过程，我们要控制 mapper，在 BlockSize 不能动的情况下，就必须控制 minSize 和 maxSize 了。这里主要控制 maxSize。 

TextInputFormat.setMinInputSplitSize(job, 1L);//设置minSize
TextInputFormat.setMaxInputSplitSize(job, 10 * 1024 * 1024);//设置maxSize

测试输入文件大小为 40MB， 很小， 在默认情况下， 被分配为 1 个或 2 个 mapper 执行成功。

现在希望分配 4 个mapper：那么设置 maxSize 为10M ，那么 splitSize 计算为 10M。对于 40MB 的输入文件，理应分配 4 个mapper。

实际运行，运行了 5 个mapper，认为成功摆脱了默认启动 2 个mapper 的限制，额外多出的 1 个 mapper 则猜测是上文提到的，对残余量的补充 mapper。

6. 至此，对Java 本地提交 mapreduce 任务， org.apache.hadoop.mapreduce.lib.input.FileInputFormat 的 mapper num 控制方法如上。接下来讨论 streaming 使用的 org.apache.hadoop.mapred.FileInputFormat 的 mapper 控制。

二、streaming 提交 mapreduce 任务， org.apache.hadoop.mapred.FileInputFormat 的 mapper num 控制

1. 可通过 mapreduce.job.maps 直接控制，即使不是绝对精确。原因在下面的源码分析中可以看到。

hadoop dfs -rm -r -f /output && 

hadoop jar /opt/hadoop-3.0.2/share/hadoop/tools/lib/hadoop-streaming-3.0.2.jar 
-D mapreduce.reduce.tasks=0 
-D mapreduce.job.maps=7 
-input /input 
-output /output 
-mapper "cat" 
-inputformat TextInputFormat

2. 将 org.apache.hadoop.mapreduce.lib.input.FileInputFormat 中的 maxSize，尝试通过 streaming 的 -D 设置，是无效的。因为 streaming 使用的是 org.apache.hadoop.mapred.FileInputFormat，在下面的源码分析中可以看到。

3. 查看 FileInputFormat 的 getSplits 源码

    public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
        StopWatch sw = (new StopWatch()).start();
        FileStatus[] files = this.listStatus(job);
        job.setLong("mapreduce.input.fileinputformat.numinputfiles", (long)files.length);
        long totalSize = 0L;
        FileStatus[] var7 = files;
        int var8 = files.length;

        for(int var9 = 0; var9 < var8; ++var9) {
            FileStatus file = var7[var9];
            if (file.isDirectory()) {
                throw new IOException("Not a file: " + file.getPath());
            }

            totalSize += file.getLen();
        }

        long goalSize = totalSize / (long)(numSplits == 0 ? 1 : numSplits);
        long minSize = Math.max(job.getLong("mapreduce.input.fileinputformat.split.minsize", 1L), this.minSplitSize);
        ArrayList<FileSplit> splits = new ArrayList(numSplits);
        NetworkTopology clusterMap = new NetworkTopology();
        FileStatus[] var13 = files;
        int var14 = files.length;

        for(int var15 = 0; var15 < var14; ++var15) {
            FileStatus file = var13[var15];
            Path path = file.getPath();
            long length = file.getLen();
            if (length == 0L) {
                splits.add(this.makeSplit(path, 0L, length, new String[0]));
            } else {
                FileSystem fs = path.getFileSystem(job);
                BlockLocation[] blkLocations;
                if (file instanceof LocatedFileStatus) {
                    blkLocations = ((LocatedFileStatus)file).getBlockLocations();
                } else {
                    blkLocations = fs.getFileBlockLocations(file, 0L, length);
                }

                if (!this.isSplitable(fs, path)) {
                    if (LOG.isDebugEnabled() && length > Math.min(file.getBlockSize(), minSize)) {
                        LOG.debug("File is not splittable so no parallelization is possible: " + file.getPath());
                    }

                    String[][] splitHosts = this.getSplitHostsAndCachedHosts(blkLocations, 0L, length, clusterMap);
                    splits.add(this.makeSplit(path, 0L, length, splitHosts[0], splitHosts[1]));
                } else {
                    long blockSize = file.getBlockSize();
                    long splitSize = this.computeSplitSize(goalSize, minSize, blockSize);

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

                    if (bytesRemaining != 0L) {
                        splitHosts = this.getSplitHostsAndCachedHosts(blkLocations, length - bytesRemaining, bytesRemaining, clusterMap);
                        splits.add(this.makeSplit(path, length - bytesRemaining, bytesRemaining, splitHosts[0], splitHosts[1]));
                    }
                }
            }
        }

        sw.stop();
        if (LOG.isDebugEnabled()) {
            LOG.debug("Total # of splits generated by getSplits: " + splits.size() + ", TimeTaken: " + sw.now(TimeUnit.MILLISECONDS));
        }

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

与 org.apache.hadoop.mapreduce.lib.input.FileInputFormat 相似，但不同之处还是很重要的。主要在

long blockSize = file.getBlockSize();
long splitSize = this.computeSplitSize(goalSize, minSize, blockSize);

赋值：

long goalSize = totalSize / (long)(numSplits == 0 ? 1 : numSplits);
long minSize = Math.max(job.getLong("mapreduce.input.fileinputformat.split.minsize", 1L), this.minSplitSize);

4. 追溯这些量

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

private long minSplitSize = 1L;

5. 分析

minSize 默认为1L，blocksize是当前目标文件的块大小，而 splitSize 就是 BlockSize 和 goalSize 的小值。

goalSize 的计算，就是输入文件总大小与 numSplits 的比值。而 numSplits 就是我们在streaming 中设置的 -D mapreduce.job.maps

因此，在streaming中才可以简单地直接设置 mapper 的数量了。

但是，只有当 goalsize 小于 blocksize 时，mapreduce.job.maps 才会生效！

当 goalsize < blocksize，splitsize = goalsize，此时你设置的 mapreduce.job.maps 数量一般大于输入块的数量，因此配置生效。

当 goalsize > blocksize，splitsize = blocksize，此时你设置的 mapreduce.job.maps 不足，一般少于输入块的数量，因此配置不生效。

换句话说，如果输入只有一个文件，那么只要 -D mapreduce.job.maps > 1，配置大多数会生效。