深度分析如何在Hadoop中控制Map的数量

深度分析如何在Hadoop中控制Map的数量

[email protected]

很多文档中描述，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. for (FileStatus file: files) {
  4. Path path = file.getPath();
  5. FileSystem fs = path.getFileSystem(job);
  6. if ((length != 0) && isSplitable(fs, path)) {
  7. long blockSize = file.getBlockSize();
  8. long splitSize = computeSplitSize(goalSize, minSize, blockSize);
  9. long bytesRemaining = length;
  10. while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
  11. String[] splitHosts = getSplitHosts(blkLocations,length-bytesRemaining, splitSize, clusterMap);
  12. splits.add(new FileSplit(path, length-bytesRemaining, splitSize, splitHosts));
  13. bytesRemaining -= splitSize;
  14. }
  15. if (bytesRemaining != 0) {
  16. splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining, blkLocations[blkLocations.length-1].getHosts()));
  17. }
  18. } else if (length != 0) {
  19. String[] splitHosts = getSplitHosts(blkLocations,0,length,clusterMap);
  20. splits.add(new FileSplit(path, 0, length, splitHosts));
  21. } else {
  22. //Create empty hosts array for zero length files
  23. splits.add(new FileSplit(path, 0, length, new String[0]));
  24. }
  25. }
  26. return splits.toArray(new FileSplit[splits.size()]);
  27. protected long computeSplitSize(long goalSize, long minSize, long blockSize) {
  28. return Math.max(minSize, Math.min(goalSize, blockSize));
  29. }

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

时间： 2024-10-21 20:37:49

深度分析如何在Hadoop中控制Map的数量的相关文章

如何在hadoop中控制map的个数

hadooop提供了一个设置map个数的参数mapred.map.tasks,我们可以通过这个参数来控制map的个数.但是通过这种方式设置map 的个数,并不是每次都有效的.原因是mapred.map.tasks只是一个hadoop的参考数值,最终map的个数,还取决于其他的因素. 为了方便介绍,先来看几个名词: block_size : hdfs的文件块大小,默认为64M,可以通过参数dfs.block.size设置 total_size : 输入文件整体的大小 input_file_num

Hadoop 中关于 map，reduce 数量设置

map和reduce是hadoop的核心功能,hadoop正是通过多个map和reduce的并行运行来实现任务的分布式并行计算,从这个观点来看,如果将map和reduce的数量设置为1,那么用户的任务就没有并行执行,但是map和reduce的数量也不能过多,数量过多虽然可以提高任务并行度,但是太多的map和reduce也会导致整个hadoop框架因为过度的系统资源开销而使任务失败.所以用户在提交map/reduce作业时应该在一个合理的范围内,这样既可以增强系统负载匀衡,也可以降低任务失败的开销

Android如何在activity中控制屏幕点亮与关闭

Android中通过各种Lock锁对电源进行控制,需要注意的是加锁和解锁必须成对出现. //启用屏幕常亮功能 PowerManager pm =(PowerManager) getSystemService(POWER_SERVICE); wakeLock= pm.newWakeLock(PowerManager.FULL_WAKE_LOCK| PowerManager.ACQUIRE_CAUSES_WAKEUP, myPowerManager.TAG); wakeLock.acquire();

Hadoop中的map端连接和reduce端连接以及半连接

MapReduce表连接操作之Map端join https://blog.csdn.net/lzm1340458776/article/details/42971075 MapReduce表连接操作之Reduce端join https://blog.csdn.net/lzm1340458776/article/details/42971485 MapReduce表连接之半连接SemiJoin https://blog.csdn.net/lzm1340458776/article/details/

Hadoop MR Job 关于如何控制Map Task 数量

整理下,基本分两个方式: 一.对于大量大文件(大于block块设置的大小) 增大minSize,即增大mapred.min.split.size的值,原因:splitsize=max(minisize,min(maxsize,blocksize)),blocksize一般不会做修改. 在没有设置minisize,maxsize时,splitsize取blocksize. 二.对于大量小文件(小于block块设置的大小) 这种情况通过增大mapred.min.split.size不可行, 需要使用

Hadoop2.6.0的FileInputFormat的任务切分原理分析（即如何控制FileInputFormat的map任务数量）

前言 首先确保已经搭建好Hadoop集群环境,可以参考<Linux下Hadoop集群环境的搭建>一文的内容.我在测试mapreduce任务时,发现相比于使用Job.setNumReduceTasks(int)控制reduce任务数量而言,控制map任务数量一直是一个困扰我的问题.好在经过很多摸索与实验,终于梳理出来,希望对在工作中进行Hadoop进行性能调优的新人们有个借鉴.本文只针对FileInputFormat的任务划分进行分析,其它类型的InputFormat的划分方式又各有不同.虽然如

Hadoop中的控制脚本

1.提出问题 在上篇博文中,提到了为什么要配置ssh免密码登录,说是Hadoop控制脚本依赖SSH来执行针对整个集群的操作,那么Hadoop中控制脚本都是什么东西呢?具体是如何通过SSH来针对整个集群的操作?网上完全分布模式下Hadoop的搭建很多,可是看完后,真的了解吗?为什么要配置Hadoop下conf目录下的masters文件和slaves文件,masters文件里面主要记录的是什么东西,slaves文件中又记录的是什么东西,masters文件和slaves文件都有什么作用?好,我看到过一

Hadoop map任务数量的计算

Hadoop中决定map个数的的因素有几个,由于版本的不同,决定因素也不一样,掌握这些因素对了解hadoop分片的划分有很大帮助, 并且对优化hadoop性能也很有大的益处. 旧API中getSplits方法: 1 public InputSplit[] getSplits(JobConf job, int numSplits) 2 throws IOException { 3 FileStatus[] files = listStatus(job); 4 5 // Save the numbe

Hadoop中常用的InputFormat、OutputFormat（转）

Hadoop中的Map Reduce框架依赖InputFormat提供数据,依赖OutputFormat输出数据,每一个Map Reduce程序都离不开它们.Hadoop提供了一系列InputFormat和OutputFormat方便开发,本文介绍几种常用的: TextInputFormat 作为默认的文件输入格式,用于读取纯文本文件,文件被分为一系列以LF或者CR结束的行,key是每一行的位置偏移量,是LongWritable类型的,value是每一行的内容,为Text类型. KeyValue