首页 > 代码库 > MapReduce获取分片数目
MapReduce获取分片数目
问题
MapReduce Application中mapper的数目和分片的数目是一样的,但是分片数目和什么有关呢?
- 默认情况下,分片和输入文件的分块数是相等的。也不完全相等,如果block size大小事128M,文件大小为128.1M,文件的block数目为2,但是application运行过程中,你会发现分片数目是1,而不是2,其中的机理,后面会分析
- 有的程序会设置map的数目,那么map数目是怎样影响分片的数目的呢?
- 如果文件大小为0,是否会作为一个分片传给map任务?
流程
FileInputFormat.getSplits返回文件的分片数目,这部分将介绍其运行流程,后面将粘贴其源码并给出注释
- 通过listStatus()获取输入文件列表files,其中会遍历输入目录的子目录,并过滤掉部分文件,如文件_SUCCESS
- 获取所有的文件大小totalSIze
- goalSIze=totalSize/numMaps。numMaps是用户指定的map数目
- files中取出一个文件file
- 计算splitSize。splitSize=max(minSplitSize,min(file.blockSize,goalSize)),其中minSplitSize是允许的最小分片大小,默认为1B
- 后面根据splitSize大小将file分片。在分片的时候,如果剩余的大小不大于splitSize*1.1,且大于0B的时候,会将该区域整个作为一个分片。这样做是为了防止一个mapper处理的数据太小
- 将file的分片加入到splits中
- 返回4,直到将files遍历完
- 结束,返回splits
源码
public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException { //获取输入文件列表files,其中会遍历输入目录的子目录,并过滤掉部分文件,如文件_SUCCESS 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(); } /* * numSplits为设置的map数目 * 期待的分片大小 */ long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits); /* * FileInputFormat.SPLIT_MINSIZE为参数值:mapreduce.input.fileinputformat.split.minsize,默认为0 * minSplitSize默认为1 */ 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(); /* * 计算分片的大小,每一个文件都要计算一次 *computeSplitSize的计算公式为 Math.max(minSize, Math.min(goalSize, blockSize)); */ long splitSize = computeSplitSize(goalSize, minSize, blockSize); long bytesRemaining = length; while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) { String[] splitHosts = getSplitHosts(blkLocations, length-bytesRemaining, splitSize, clusterMap); splits.add(makeSplit(path, length-bytesRemaining, splitSize, splitHosts)); bytesRemaining -= splitSize; } if (bytesRemaining != 0) { String[] splitHosts = getSplitHosts(blkLocations, length - bytesRemaining, bytesRemaining, clusterMap); splits.add(makeSplit(path, length - bytesRemaining, bytesRemaining, splitHosts)); } } else { String[] splitHosts = getSplitHosts(blkLocations,0,length,clusterMap); splits.add(makeSplit(path, 0, length, splitHosts)); } } else { //Create empty hosts array for zero length files splits.add(makeSplit(path, 0, length, new String[0])); } } LOG.debug("Total # of splits: " + splits.size()); return splits.toArray(new FileSplit[splits.size()]); }
总结
看源码还是很有用的。很多时候,博客或者书介绍的不是很中肯,或者会有错误,看源码就不会出现这些问题。
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。