mapreduce中控制mapper的数量

简介:

很多文档中描述,Mapper的数量在默认情况下不可直接控制干预,因为Mapper的数量由输入的大小和个数决定。在默认情况下,最终input占据了多少block,就应该启动多少个Mapper。如果输入的文件数量巨大,但是每个文件的size都小于HDFS的blockSize,那么会造成启动的Mapper等于文件的数量(即每个文件都占据了一个block),那么很可能造成启动的Mapper数量超出限制而导致崩溃。这些逻辑确实是正确的,但都是在默认情况下的逻辑。其实如果进行一些客户化的设置,就可以控制了。

        在Hadoop中,设置Map task的数量不像设置Reduce task数量那样直接,即:不能够通过API直接精确的告诉Hadoop应该启动多少个Map task。

       你也许奇怪了,在API中不是提供了接口org.apache.hadoop.mapred.JobConf.setNumMapTasks(int n)吗?这个值难道不可以设置Map task的数量吗?这个API的确没错,在文档上解释”Note: This is only a hint to the framework.“,即这个值对Hadoop的框架来说仅仅是个提示,不起决定性的作用。也就是说,即便你设置了,也不一定得到你想要的效果。

1. InputFormat介绍

在具体设置Map task数量之前,非常有必要了解一下与Map-Reduce输入相关的基础知识。

这个接口(org.apache.hadoop.mapred.InputFormat)描述了Map-Reduce job的输入规格说明(input-specification),它将所有的输入文件分割成逻辑上的InputSplit,每一个InputSplit将会分给一个单独的mapper;它还提供RecordReader的具体实现,这个Reader从逻辑的InputSplit上获取input records并传给Mapper处理。

InputFormat有多种具体实现,诸如FileInputFormat(处理基于文件的输入的基础抽象类), DBInputFormat(处理基于数据库的输入,数据来自于一个能用SQL查询的表),KeyValueTextInputFormat(特殊的FineInputFormat,处理Plain Text File,文件由回车或者回车换行符分割成行,每一行由key.value.separator.in.input.line分割成Key和Value),CompositeInputFormat,DelegatingInputFormat等。在绝大多数应用场景中都会使用FileInputFormat及其子类型。

通过以上的简单介绍,我们知道InputFormat决定着InputSplit,每个InputSplit会分配给一个单独的Mapper,因此InputFormat决定了具体的Map task数量

2. FileInputFormat中影响Map数量的因素

在日常使用中,FileInputFormat是最常用的InputFormat,它有很多具体的实现。以下分析的影响Map数量的因素仅对FileInputFormat及其子类有效,其他非FileInputFormat可以去查看相应的 getSplits(JobConf job, int numSplits) 具体实现即可。

请看如下代码段(摘抄自org.apache.hadoop.mapred.FileInputFormat.getSplits,hadoop-0.20.205.0源代码):

 

[java]  view plain copy
 
  1. long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);  
  2. long minSize = Math.max(job.getLong("mapred.min.split.size", 1), minSplitSize);  
  3.   
  4. for (FileStatus file: files) {  
  5.   Path path = file.getPath();  
  6.   FileSystem fs = path.getFileSystem(job);  
  7.   if ((length != 0) && isSplitable(fs, path)) {   
  8.     long blockSize = file.getBlockSize();  
  9.     long splitSize = computeSplitSize(goalSize, minSize, blockSize);  
  10.       
  11.     long bytesRemaining = length;  
  12.     while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {  
  13.       String[] splitHosts = getSplitHosts(blkLocations,length-bytesRemaining, splitSize, clusterMap);  
  14.       splits.add(new FileSplit(path, length-bytesRemaining, splitSize, splitHosts));  
  15.       bytesRemaining -= splitSize;  
  16.     }  
  17.   
  18.     if (bytesRemaining != 0) {  
  19.       splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining, blkLocations[blkLocations.length-1].getHosts()));  
  20.     }  
  21.   } else if (length != 0) {  
  22.     String[] splitHosts = getSplitHosts(blkLocations,0,length,clusterMap);  
  23.     splits.add(new FileSplit(path, 0, length, splitHosts));  
  24.   } else {   
  25.     //Create empty hosts array for zero length files  
  26.     splits.add(new FileSplit(path, 0, length, new String[0]));  
  27.   }  
  28. }  
  29.   
  30. return splits.toArray(new FileSplit[splits.size()]);  
  31.   
  32. protected long computeSplitSize(long goalSize, long minSize, long blockSize) {  
  33.     return Math.max(minSize, Math.min(goalSize, blockSize));  
  34. }  

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相比取较大的

3. 如何调整Map的数量

有了2的分析,下面调整Map的数量就很容易了。

3.1 减小Map-Reduce job 启动时创建的Mapper数量

当处理大批量的大数据时,一种常见的情况是job启动的mapper数量太多而超出了系统限制,导致Hadoop抛出异常终止执行。解决这种异常的思路是减少mapper的数量。具体如下:

3.1.1 输入文件size巨大,但不是小文件

这种情况可以通过增大每个mapper的input size,即增大minSize或者增大blockSize来减少所需的mapper的数量。增大blockSize通常不可行,因为当HDFS被hadoop namenode -format之后,blockSize就已经确定了(由格式化时dfs.block.size决定),如果要更改blockSize,需要重新格式化HDFS,这样当然会丢失已有的数据。所以通常情况下只能通过增大minSize,即增大mapred.min.split.size的值。

3.1.2 输入文件数量巨大,且都是小文件

所谓小文件,就是单个文件的size小于blockSize。这种情况通过增大mapred.min.split.size不可行,需要使用FileInputFormat衍生的CombineFileInputFormat将多个input path合并成一个InputSplit送给mapper处理,从而减少mapper的数量。具体细节稍后会更新并展开。

3.2 增加Map-Reduce job 启动时创建的Mapper数量

增加mapper的数量,可以通过减小每个mapper的输入做到,即减小blockSize或者减小mapred.min.split.size的值。

参考资料

http://yaseminavcular.blogspot.com/2011/06/how-to-set-number-of-maps-with-hadoop.html

http://svn.apache.org/repos/asf/hadoop/common/tags/release-0.20.205.0

本文转自博客园知识天地的博客,原文链接:mapreduce中控制mapper的数量,如需转载请自行联系原博主。


相关文章
|
1月前
|
SQL 分布式计算 Java
Hadoop-11-MapReduce JOIN 操作的Java实现 Driver Mapper Reducer具体实现逻辑 模拟SQL进行联表操作
Hadoop-11-MapReduce JOIN 操作的Java实现 Driver Mapper Reducer具体实现逻辑 模拟SQL进行联表操作
31 3
|
1月前
|
分布式计算 资源调度 Hadoop
Hadoop-10-HDFS集群 Java实现MapReduce WordCount计算 Hadoop序列化 编写Mapper和Reducer和Driver 附带POM 详细代码 图文等内容
Hadoop-10-HDFS集群 Java实现MapReduce WordCount计算 Hadoop序列化 编写Mapper和Reducer和Driver 附带POM 详细代码 图文等内容
95 3
|
2月前
|
分布式计算 数据库
Mapreduce中的Mapper&reducer
【9月更文挑战第19天】在 MapReduce 框架中,Mapper 和 Reducer 是处理大规模数据集的关键组件。Mapper 负责将输入数据分割成键值对,而 Reducer 则对这些键值对进行汇总处理,生成最终结果。两者通过并行处理和分布式计算协同工作,Mapper 将数据转换为键值对,Reducer 对相同键的值进行聚合。开发人员需实现相应接口并编写定制逻辑,以充分利用框架优势,处理大规模数据集并获得有价值的结果。
|
分布式计算 Hadoop
MapReduce框架Mapper和Reducer类源码分析
一:Mapper类 在Hadoop的mapper类中,有4个主要的函数,分别是:setup,cleanup,map,run。代码如下: protected void setup(Context context) throws IOException, InterruptedEx...
943 0
|
6月前
|
分布式计算 Hadoop
Hadoop系列 mapreduce 原理分析
Hadoop系列 mapreduce 原理分析
79 1
|
6月前
|
存储 分布式计算 负载均衡
【大数据技术Hadoop+Spark】MapReduce概要、思想、编程模型组件、工作原理详解(超详细)
【大数据技术Hadoop+Spark】MapReduce概要、思想、编程模型组件、工作原理详解(超详细)
220 0
|
5月前
|
分布式计算 Hadoop Java
Hadoop MapReduce编程
该教程指导编写Hadoop MapReduce程序处理天气数据。任务包括计算每个城市ID的最高、最低气温、气温出现次数和平均气温。在读取数据时需忽略表头,且数据应为整数。教程中提供了环境变量设置、Java编译、jar包创建及MapReduce执行的步骤说明,但假设读者已具备基础操作技能。此外,还提到一个扩展练习,通过分区功能将具有相同尾数的数字分组到不同文件。
63 1
|
5月前
|
数据采集 SQL 分布式计算
|
6月前
|
分布式计算 Hadoop Java
Hadoop MapReduce 调优参数
对于 Hadoop v3.1.3,针对三台4核4G服务器的MapReduce调优参数包括:`mapreduce.reduce.shuffle.parallelcopies`设为10以加速Shuffle,`mapreduce.reduce.shuffle.input.buffer.percent`和`mapreduce.reduce.shuffle.merge.percent`分别设为0.8以减少磁盘IO。
70 1
|
6月前
|
分布式计算 并行计算 搜索推荐
Hadoop MapReduce计算框架
【5月更文挑战第10天】HadoopMapReduce计算框架
52 3

热门文章

最新文章