本篇分两部分，第一部分分析使用 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，配置大多数会生效。