Hadoop旧mapreduce的map任务切分原理
时间:2022-04-28
本文章向大家介绍Hadoop旧mapreduce的map任务切分原理,主要内容包括其使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。
前言
最近在工作过程中接触一些Hive数据仓库中的表,这些表实际是从关系型数据库通过Sqoop抽到Hive的。在开发过程中对map任务的划分进行性能调优,发现mapreduce中关于FileInputFormat的参数调整都不起作用,最后发现这些老任务都是用旧版的mapreduce开发的,于是顺便研究下旧版mapreduce的任务划分策略。有关新版mapreduce的任务划分策略,大家可以参考我之前的博文《Hadoop2.6.0的FileInputFormat的任务切分原理分析(即如何控制FileInputFormat的map任务数量)》。
源码分析
根据《Hadoop2.6.0的FileInputFormat的任务切分原理分析(即如何控制FileInputFormat的map任务数量)》一文的内容,我们知道map任务的划分关键在于FileInputFormat的getSplits方法的实现策略,现在我们来看看其源码:
public InputSplit[] getSplits(JobConf job, int numSplits)
throws IOException {
Stopwatch sw = new Stopwatch().start();
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();
}
long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
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();
long splitSize = computeSplitSize(goalSize, minSize, blockSize);
long bytesRemaining = length;
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,
length-bytesRemaining, splitSize, clusterMap);
splits.add(makeSplit(path, length-bytesRemaining, splitSize,
splitHosts[0], splitHosts[1]));
bytesRemaining -= splitSize;
}
if (bytesRemaining != 0) {
String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations, length
- bytesRemaining, bytesRemaining, clusterMap);
splits.add(makeSplit(path, length - bytesRemaining, bytesRemaining,
splitHosts[0], splitHosts[1]));
}
} else {
String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,0,length,clusterMap);
splits.add(makeSplit(path, 0, length, splitHosts[0], splitHosts[1]));
}
} else {
//Create empty hosts array for zero length files
splits.add(makeSplit(path, 0, length, new String[0]));
}
}
sw.stop();
if (LOG.isDebugEnabled()) {
LOG.debug("Total # of splits generated by getSplits: " + splits.size()
+ ", TimeTaken: " + sw.elapsedMillis());
}
return splits.toArray(new FileSplit[splits.size()]);
}
protected long computeSplitSize(long goalSize, long minSize,
long blockSize) {
return Math.max(minSize, Math.min(goalSize, blockSize));
}
这里对以上代码的划分策略进行整理:
- 遍历当前作业的所有输入文件,然后将累积这些文件的字节数并保存到变量totalSize中;
- 如果用户指定了mapreduce.job.maps参数,那么这个参数会被保存在入参numSplits中;
- 用户想要通过numSplits控制map任务的数量,那么需求对totalSize进行平分,以便确定每个map任务划分的输入大小。这个计算很简单,即使用totalSize除以numSplits,最后得到的目标划分大小存储在变量goalSize中;
- 常量SPLIT_MINSIZE实际是由参数mapreduce.input.fileinputformat.split.minsize来控制的,如果没有配置则默认是1。minSplitSize默认是1,切旧版FileIntputFormat没有设置此变量的地方。最后取SPLIT_MINSIZE和minSplitSize的最大值,并保存在变量minSize中;
- 遍历当前作业的每个输入文件,计算每个输入文件,将被划分的任务数量,最后将每个文件划分的任务数量合并起来就是整个作业划分的任务数量。
以上只是总体分析了作业的任务划分,有关每个输入文件的任务数量划分步骤如下:
- 判断文件的大小,只有文件字节数大于0才是有意义的;
- 判断文件是否是可以切分的,只有能够切分的文件才会继续进行任务数量划分;
- 调用文件的getBlockSize方法,获取文件的块大小并存储在变量blockSize中;
- 调用computeSplitSize方法计算最后划分给每个任务的输入大小,并保存在splitSize中。计算公式为:splitSize = max(minSize, min(goalSize, blockSize));
- 将文件按照splitSize的大小进行划分,不足splitSize大小的也算作一个任务划分数。
总结
根据以上分析发现旧版mapreduce和新版mapreduce的FileIntputFormat关于map任务数量划分的实现逻辑不同,在对它们进行开发和性能优化时要特别注意。
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- springBoot快速入门
- springBoot配置文件
- springBoot按条件装配:Condition
- springBoot @Enable*注解的工作原理
- 使用dom4j修改XML格式的字符串
- springBoot @EnableAutoConfiguration深入分析
- SpringBoot事件监听
- SpringBoot Web(SpringMVC)
- SpringBoot使用servletAPI与异常处理
- Redis数据迁移至Codis集群方案
- Oracle分析函数
- springBoot定制内嵌的Tomcat
- SpringBoot JDBC/AOP
- SpringBoot日志
- DockerFile就这么简单