Yarn下Map数控制

时间：2014-07-16 00:19:43 阅读：324 评论：0 收藏：0 [点我收藏+]

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

        List splits = new ArrayList();
        List files = listStatus(job);
        for (FileStatus file : files) {
            Path path = file.getPath();
            long length = file.getLen();
            if (length != 0L) {
                FileSystem fs = path.getFileSystem(job.getConfiguration());
                BlockLocation[] blkLocations = fs.getFileBlockLocations(file,
                        0L, length);
                if (isSplitable(job, path)) {
                    long blockSize = file.getBlockSize();
                    long splitSize = computeSplitSize(blockSize, minSize,
                            maxSize);

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

                        bytesRemaining -= splitSize;
                    }

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

        job.getConfiguration().setLong(
                "mapreduce.input.fileinputformat.numinputfiles", files.size());
        LOG.debug("Total # of splits: " + splits.size());
        return splits;
    }

Yarn 下好像没了1*下的由用户设置预期的Map数

核心代码

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

getFormatMinSplitSize 默认返回1，getMinSplitSize 为用户设置的最小分片数, 如果用户设置的大于1，则为用户设置的最小分片数
long maxSize = getMaxSplitSize(job);

getMaxSplitSize为用户设置的最大分片数，默认最大为9223372036854775807L

long splitSize = computeSplitSize(blockSize, minSize,
                            maxSize);

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

测试文件大小 297M(311349250)

块大小128M

测试代码

测试1

FileInputFormat.setMinInputSplitSize(job, 301349250);
FileInputFormat.setMaxInputSplitSize(job, 10000);

测试后Map个数为1，由上面分片公式算出分片大小为301349250, 比 311349250小，理论应该为两个map, 再看分片函数

while (bytesRemaining / splitSize > 1.1D) {
                       int blkIndex = getBlockIndex(blkLocations, length
                               - bytesRemaining);
                       splits.add(makeSplit(path, length - bytesRemaining,
                               splitSize, blkLocations[blkIndex].getHosts()));

                       bytesRemaining -= splitSize;
                   }

只要剩余的文件大小不超过分片大小的1.1倍，则会分到一个分片中，避免开两个MAP，其中一个运行数据太小，浪费资源。

测试2

FileInputFormat.setMinInputSplitSize(job, 150*1024*1024);

FileInputFormat.setMaxInputSplitSize(job, 10000);

MAP 数为2

测试3

在原有的输入目录下，添加一个很小的文件,几K，测试是否会合并

FileInputFormat.setMinInputSplitSize(job, 150*1024*1024);
FileInputFormat.setMaxInputSplitSize(job, 10000);

Map数变为了3

看源代码

for (FileStatus file : files) {

}