Hadoop框架下MapReduce中的map个数如何控制
一个job的map阶段并行度由客户端在提交job时决定
客户端对map阶段并行度的规划基本逻辑为:
一、将待处理的文件进行逻辑切片(根据处理数据文件的大小,划分多个split),然后每一个split分配一个maptask并行处理实例
二、具体切片规划是由FileInputFormat实现类的getSplits()方法完成
切分规则如下:
1.简单地按照文件的内容长度进行切片
2.切片大小默认是datanode的切块大小128M
3.切片时不是考虑一个整体数据集,而是针对每一个文件单独切片
比如待处理数据有两个文件:
file1.txt 200M
file2.txt 50M
经过FileInputFormat的切片机制运算后,形成的切片信息如下:
file1.txt.split1– 0~128M —–maptask
file1.txt.split2– 128M~200M —–maptask
file2.txt.split1– 0~50M —–maptask
三、如何改变切片大小(参数设置)
源码是通过这个方法来规划切片大小的
protected long computeSplitSize(long blockSize, long minSize,
long maxSize) {
return Math.max(minSize, Math.min(maxSize, blockSize));
}
minsize:默认值:1;配置参数: mapreduce.input.fileinputformat.split.minsize
maxsize:默认值:Long.MAXValue; 配置参数:mapreduce.input.fileinputformat.split.maxsize
blocksize:hdfs切片大小
调整切片大小结论:
maxsize(切片最大值):
参数如果调得比blocksize小,则会让切片变小,而且就等于配置的这个参数的值
minsize (切片最小值):
参数调的比blockSize大,则可以让切片变得比blocksize还大
控制map个数的核心源码
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
//getFormatMinSplitSize 默认返回1,getMinSplitSize 为用户设置的最小分片数, 如果用户设置的大于1,则为用户设置的最小分片数
long maxSize = getMaxSplitSize(job);
//getMaxSplitSize为用户设置的最大分片数,默认最大为long 9223372036854775807L
long splitSize = computeSplitSize(blockSize, minSize,
maxSize);
protected long computeSplitSize(long blockSize, long minSize, long maxSize) {
return Math.max(minSize, Math.min(maxSize, blockSize));
}
由上述代码可以看出在
maxSize默认等于long(长整形)
blockSize默认在hadoop2.0之后为128M
minSize默认等于1
因此默认的切片大小splitSize等于128M也就是说等于块大小
一个切片对应于一个map任务,因此在默认情况下一个块对应于一个map任务。
要想人为控制map的个数可以从minSize和MaxSize入手。
想要增加map的个数,可以将maxSize调整小于blockSize;想要减小map的个数,可以调整minSize>blockSize。
具体调整可以在job配置中增加如下配置
FileInputFormat.setMinInputSplitSize(job, 301349250);//设置minSize
FileInputFormat.setMaxInputSplitSize(job, 10000);//设置maxSize
在实验中,
测试 文件大小 297M(311349250)
块大小128M
测试代码
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, 其中一个运行数据太小,浪费资源。
总结,分片过程大概为,先遍历目标文件,过滤部分不符合要求的文件, 然后添加到列表,然后按照文件名来切分分片 (大小为前面计算分片大小的公式, 最后有个文件尾可能合并,其实常写网络程序的都知道), 然后添加到分片列表,然后每个分片读取自身对应的部分给MAP处理