Hadoop-2.4.1学习之怎样确定Mapper数量 _HaDoop

会挽雕弓如满月，西北望，射天狼。这篇文章主要讲述Hadoop-2.4.1学习之怎样确定Mapper数量相关的知识，希望能为你提供帮助。
MapReduce框架的优势是能够在集群中并行运行mapper和reducer任务，那怎样确定mapper和reducer的数量呢，或者说怎样以编程的方式控制作业启动的mapper和reducer数量呢？在《Hadoop-2.4.1学习之Mapper和Reducer》中以前提及建议reducer的数量为(0.95~1.75 ) * 节点数量 * 每一个节点上最大的容器数，并可用法Job.setNumReduceTasks(int)。mapper的数量由输入文件的大小确定。且没有相应的setNumMapTasks方法，但能够通过Configuration.set(JobContext.NUM_MAPS, int)设置，当中JobContext.NUM_MAPS的值为mapreduce.job.maps，而在Hadoop的官方站点上对该參数的描写叙述为与MapReduce框架和作业配置巧妙地交互。而且设置起来更加复杂。
从这样一句含糊不清的话无法得知到底怎样确定mapper的数量。显然仅仅能求助于源码了。

在Hadoop中MapReduce作业通过JobSubmitter类的submitJobInternal(Jobjob, Cluster cluster)方法向系统提交作业（该方法不仅设置mapper数量。还运行了一些其他操作如检查输出格式等，感兴趣的能够參考源码），在该方法中与设置mapper有关的代码例如以下：

int maps = writeSplits(job, submitJobDir); conf.setInt(MRJobConfig.NUM_MAPS, maps); LOG.info("number of splits:" + maps);

方法writeSplits返回mapper的数量，该方法的源码例如以下：

private int writeSplits(org.apache.hadoop.mapreduce.JobContext job,Path jobSubmitDir) throws IOException,InterruptedException, ClassNotFoundException { JobConf jConf = (JobConf)job.getConfiguration(); int maps; if (jConf.getUseNewMapper()) { maps = writeNewSplits(job, jobSubmitDir); } else { maps = writeOldSplits(jConf, jobSubmitDir); } return maps; }

在该方法中，依据是否使用了新版本号的JobContext而使用不同的方法计算mapper数量。实际情况是jConf.getUseNewMapper()将返回true，因此将运行writeNewSplits(job,jobSubmitDir)语句，该方法的源码例如以下：

Configuration conf = job.getConfiguration(); InputFormat< ?, ?
> input = ReflectionUtils.newInstance(job.getInputFormatClass(), conf); List< InputSplit> splits = input.getSplits(job); T[] array = (T[]) splits.toArray(new InputSplit[splits.size()]); // sort the splits into order based on size, so that the biggest // go first Arrays.sort(array, new SplitComparator()); JobSplitWriter.createSplitFiles(jobSubmitDir, conf, jobSubmitDir.getFileSystem(conf), array); return array.length;

通过上面的代码能够得知，实际的mapper数量为输入分片的数量，而分片的数量又由使用的输入格式决定，默觉得TextInputFormat，该类为FileInputFormat的子类。确定分片数量的任务交由FileInputFormat的getSplits(job)完毕，在此补充一下FileInputFormat继承自抽象类InputFormat，该类定义了MapReduce作业的输入规范，当中的抽象方法List< InputSplit> getSplits(JobContext context)定义了怎样将输入切割为InputSplit。不同的输入有不同的分隔逻辑，而分隔得到的每一个InputSplit交由不同的mapper处理，因此该方法的返回值确定了mapper的数量。以下将分为两部分学习该方法是怎样在FileInputFormat中实现的，为了将注意力集中在最重要的部分。对日志输出等信息将不做介绍，完整的实现能够參考源码。
【Hadoop-2.4.1学习之怎样确定Mapper数量】 首先是第一部分，该部分代码计算了最大InputSplit和最小InputSplit的值，例如以下：

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

当中的getMinSplitSize和getMaxSplitSize方法分别用于获取最小InputSplit和最大InputSplit的值。相应的配置參数分别为mapreduce.input.fileinputformat.split.minsize。默认值为1L和mapreduce.input.fileinputformat.split.maxsize，默认值为Long.MAX_VALUE，十六进制数值为 0x7fffffffffffffffL，相应的十进制为9223372036854775807，getFormatMinSplitSize方法返回该输入格式下InputSplit的下限。
以上数字的单位都是byte。由此得出minSize的大小为1L。maxSize的大小为Long.MAX_VALUE。
其次是生成InputSplit的第二部分。在该部分将生成包括InputSplit的List，而List的大小为InputSplit的数量，进而确定了mapper的数量。当中重要的代码为：

if (isSplitable(job, path)) { long blockSize = file.getBlockSize(); long splitSize = computeSplitSize(blockSize, minSize, maxSize); long bytesRemaining = length; while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) { int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining); splits.add(makeSplit(path, length-bytesRemaining, splitSize, blkLocations[blkIndex].getHosts())); bytesRemaining -= splitSize; } if (bytesRemaining != 0) { int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining); splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining, blkLocations[blkIndex].getHosts())); } }

blockSize的值为參数dfs.blocksize的值，默觉得128M。方法computeSplitSize(blockSize, minSize, maxSize)依据blockSize，minSize。maxSize确定InputSplit的大小，源码例如以下：

Math.max(minSize, Math.min(maxSize, blockSize))

从该代码并结合第一部分的分析能够得知，InputSplit的大小取决于dfs.blocksiz、mapreduce.input.fileinputformat.split.minsize、mapreduce.input.fileinputformat.split.maxsize和所使用的输入格式。
在输入格式为TextInputFormat的情况下，且不改动InputSplit的最大值和最小值的情况，InputSplit的终于值为dfs.blocksize的值。

变量SPLIT_SLOP的值为1.1。决定了当剩余文件大小多大时停止依照变量splitSize切割文件。
依据代码可知，当剩余文件小于等于1.1倍splitSize时，将把剩余的文件做为一个InputSplit。即最后一个InputSplit的大小最大为1.1倍splitSize。

总结本文分析了在输入格式为默认的TextInputFormat的情况，怎样确定mapper的数量。在不改动源码的情况下（改动输入格式的InputSplit下限）。程序猿能够通过设置dfs.blocksiz、mapreduce.input.fileinputformat.split.minsize、mapreduce.input.fileinputformat.split.maxsize參数的值设置InputSplit的大小来影响InputSplit的数量。进而决定mapper的数量。
当输入为其他格式时，处理逻辑又不同样了，比方当输入格式为DBInputFormat时。会依据输入表的行数（记录数）决定mapper的数量。很多其他细节能够參考源码。