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，配置大多數會生效。