大家好，又见面了，我是你们的朋友全栈君。
1，在介绍hadoop写文件的时候我们经常会说首先分割文件为多个块；那么是怎么分割的呢？
这里其实不要有过的纠结，这里的块是block，是hdfs中切块的大小，属于物理划分，默认64M，在hadoop-default.xml配置中有体现：
<property>
<name>dfs.block.size</name>
<value>67108864</value>
<description>The default block size for new files.</description>
</property> 当然如果文件没有64M也不会占据整块空间。
将文件分割成多个块后，形成一个数据队列，然后依次写入datanode列表。
再者，如果写入的是个文件夹，而且每个文件的都不大，这样在hdfs中是默认每个文件一个块的，即使没有64m，当然也可做优化处理，不过hbase更便利于处理把小文件合并到一个块中，这个我会在其他博文中介绍。
2，下面我们说说split，并与block的关系
首先，split是mapreduce中的概念，而block是hdfs中切块的大小。
如下：
//设置要处理的文本数据所存放的路径
FileInputFormat.setInputPaths(wordCountJob, "hdfs://ubuntu:9000/input/aa.txt");
FileOutputFormat.setOutputPath(wordCountJob, new Path("hdfs://ubuntu:9000/output/")); 我们设置要处理的文件路径时都会用到fileInputFormat类， 不过我们更多看到的是inputFormat，其实fileInputFormat这个类的也是实现inputFomat接口，
下面我们接着看源码，说明为什么需要分片？
以hadoop自带的wordCount的源码为例：
for (int i = 0; i < otherArgs.length - 1; ++i) {
FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
}
FileOutputFormat.setOutputPath(job,
new Path(otherArgs[otherArgs.length - 1]));
System.exit(job.waitForCompletion(true) ? 0 : 1); 我们看到使用的InputFormat是FileOutputFormat，任务执行调用了Job的waitForCompletion方法。waitForCompletion方法中真正提交job的代码如下：
public boolean waitForCompletion(boolean verbose
) throws IOException, InterruptedException,
ClassNotFoundException {
if (state == JobState.DEFINE) {
submit();
}
// 省略本文不关心的代码
return isSuccessful();
} 这里的submit方法的实现如下：
public void submit()
throws IOException, InterruptedException, ClassNotFoundException {
// 省略本文不关心的代码</span>
final JobSubmitter submitter =
getJobSubmitter(cluster.getFileSystem(), cluster.getClient());
status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
public JobStatus run() throws IOException, InterruptedException,
ClassNotFoundException {
return submitter.submitJobInternal(Job.this, cluster);
}
});
state = JobState.RUNNING;
LOG.info("The url to track the job: " + getTrackingURL());
} submit方法首先创建了JobSubmitter实例，然后异步调用了JobSubmitter的submitJobInternal方法。JobSubmitter的submitJobInternal方法有关划分任务的代码如下：
// Create the splits for the job
LOG.debug("Creating splits at " + jtFs.makeQualified(submitJobDir));
int maps = writeSplits(job, submitJobDir);
conf.setInt(MRJobConfig.NUM_MAPS, maps);
LOG.info("number of splits:" + maps); writeSplits方法的实现如下：
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;
} 由于WordCount使用的是新的mapreduce API，所以最终会调用writeNewSplits方法。writeNewSplits的实现如下：
private <T extends InputSplit>
int writeNewSplits(JobContext job, Path jobSubmitDir) throws IOException,
InterruptedException, ClassNotFoundException {
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;
} writeNewSplits方法中，划分任务数量最关键的代码即为InputFormat的getSplits方法（提示：大家可以直接通过此处的调用，查看不同InputFormat的划分任务实现）。根据前面的分析我们知道此时的InputFormat即为FileOutputFormat，其getSplits方法的实现如下：
public List<InputSplit> getSplits(JobContext job) throws IOException {
Stopwatch sw = new Stopwatch().start();
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
long maxSize = getMaxSplitSize(job);
// generate splits
List<InputSplit> splits = new ArrayList<InputSplit>();
List<FileStatus> files = listStatus(job);
for (FileStatus file: files) {
Path path = file.getPath();
long length = file.getLen();
if (length != 0) {
BlockLocation[] blkLocations;
if (file instanceof LocatedFileStatus) {
blkLocations = ((LocatedFileStatus) file).getBlockLocations();
} else {
FileSystem fs = path.getFileSystem(job.getConfiguration());
blkLocations = fs.getFileBlockLocations(file, 0, length);
}
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(),
blkLocations[blkIndex].getCachedHosts()));
bytesRemaining -= splitSize;
}
if (bytesRemaining != 0) {
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
blkLocations[blkIndex].getHosts(),
blkLocations[blkIndex].getCachedHosts()));
}
} else { // not splitable
splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
blkLocations[0].getCachedHosts()));
}
} else {
//Create empty hosts array for zero length files
splits.add(makeSplit(path, 0, length, new String[0]));
}
}
// Save the number of input files for metrics/loadgen
job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
sw.stop();
if (LOG.isDebugEnabled()) {
LOG.debug("Total # of splits generated by getSplits: " + splits.size()
+ ", TimeTaken: " + sw.elapsedMillis());
}
return splits;
} totalSize：是整个Map-Reduce job所有输入的总大小。
numSplits：来自job.getNumMapTasks()，即在job启动时用org.apache.hadoop.mapred.JobConf.setNumMapTasks(int n)设置的值，给M-R框架的Map数量的提示。
goalSize：是输入总大小与提示Map task数量的比值，即期望每个Mapper处理多少的数据，仅仅是期望，具体处理的数据数由下面的computeSplitSize决定。
minSplitSize：默认为1，可由子类复写函数protected void setMinSplitSize(long minSplitSize) 重新设置。一般情况下，都为1，特殊情况除外。
minSize：取的1和mapred.min.split.size中较大的一个。
blockSize：HDFS的块大小，默认为64M，一般大的HDFS都设置成128M。
splitSize：就是最终每个Split的大小，那么Map的数量基本上就是totalSize/splitSize。
接下来看看computeSplitSize的逻辑：首先在goalSize（期望每个Mapper处理的数据量）和HDFS的block size中取较小的，然后与mapred.min.split.size相比取较大的。
一个片为一个splits，即一个map，只要搞清楚片的大小，就能计算出运行时的map数。而一个split的大小是由goalSize, minSize, blockSize这三个值决定的。computeSplitSize的逻辑是，先从goalSize和blockSize两个值中选出最小的那个（比如一般不设置map数，这时blockSize为当前文件的块size，而goalSize是文件大小除以用户设置的map数得到的，如果没设置的话，默认是1），在默认的大多数情况下，blockSize比较小。然后再取blockSize和minSize中最大的那个。而minSize如果不通过”mapred.min.split.size”设置的话（”mapred.min.split.size”默认为0），minSize为1，可理解为一个block块，这样得出的一个splits的size就是blockSize，即一个块一个map，有多少块就有多少map。
split的大小时默认和hdfs的block块大小一致，但是可以通过配置文件自己设置：
其中有俩个配置文件（如下）：
--minsize 默认大小为1
mapreduce.input.fileinputformat.split.minsize
--maxsize 默认大小为Long.MAXValue
mapreduce.input.fileinputformat.split.maxsize举例：
比如说我问写入一个文件夹，里面有10个只有几k的文件，所以List<FileStatus> files = listStatus(job);方法返回的files列表的大小为10。在遍历files列表的过程中，会获取每个文件的blockSize，最终调用computeSplitSize方法计算每个输入文件应当划分的任务数。computeSplitSize方法的实现如下：
protected long computeSplitSize(long blockSize, long minSize,
long maxSize) {
return Math.max(minSize, Math.min(maxSize, blockSize));
} 然后根据
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job)); 很明显取split默认值，也就是一个块，那么10个就要分为10块，这也说明为什么处理小文件时，block的大小小于split的 大小。同时我们看到了一个split分配一个map任务。
这里我们可以总结下split大小与block的关系：
（1）block块的小于split分片的最小值，那split的值就是split分片的大小
（2）block块的小大介于split分片配置的最小值和最大值之间，block的大小就是split的大小。
（3）block块的大小大于split分片的最大值，split的大小就是split配置的最大值。但会增加map执行的并发度，但是会造成在节点之间拉取数据
也有公式可以计算split也就是map任务数，这里就不做讨论了。
一个map对应一个split分片吗？
经过上面的讨论，答案是显而易见的：
map个数：由任务切片spilt决定的，默认情况下一个split的大小就是block
由参与任务的文件个数决定的
发布者：全栈程序员-站长，转载请注明出处：https://javaforall.net/106130.html原文链接：https://javaforall.net
