正文

（转） 通过input分片的大小来设置map的个数

IP云
IP云V管理员/6 阅读
08/09
提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

摘要 通过input分片的大小来设置map的个数

map inputsplit hadoop

前言：在具体执行Hadoop程序的时候，我们要根据不同的情况来设置Map的个数。除了设置固定的每个节点上可运行的最大map个数外，我们还需要控制真正执行Map操作的任务个数。
 1.如何控制实际运行的map任务个数
 我们知道，文件在上传到Hdfs文件系统的时候，被切分成不同的Block块（默认大小为64MB）。但是每个Map处理的分块有时候并不是系统的物理Block块大小。实际处理的输入分块的大小是根据InputSplit来设定的，那么InputSplit是怎么得到的呢？

?
1
2
3
4
5
 InputSplit=Math.max(minSize, Math.min(maxSize, blockSize)
 
 其中：minSize=mapred.min.split.size
 
     maxSize=mapred.max.split.size

我们通过改变InputFormat中分片的多少来控制实际使用的Map数量，而控制InputFormat中的分片多少就需要控制每个InputSplit分片的大小
 2.如何控制每个split分片的大小
 Hadoop默认的输入格式是TextInputFormat,他里边定义了文件读取的方式和分片的方式。我们打开他的源文件（org.apache.hadoop.mapreduce.lib.input包中）：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
package org.apache.hadoop.mapreduce.lib.input;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.SplittableCompressionCodec;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
public class TextInputFormat extends FileInputFormat<LongWritable, Text> {
   @Override
   public RecordReader<LongWritable, Text> 
     createRecordReader(InputSplit split,
                       TaskAttemptContext context) {
     return new LineRecordReader();
   }
   @Override
   protected boolean isSplitable(JobContext context, Path file) {
     CompressionCodec codec = 
       new CompressionCodecFactory(context.getConfiguration()).getCodec(file);
     if (null == codec) {
       return true;
     }
     return codec instanceof SplittableCompressionCodec;
   }
}

通过源代码，我们发现TextInputFormat继承了FileInputFormat，而在TextInputFormat中，我们并没有发现具体的进行文件切分的部分，TextInputFormat应该是采用了FileInputFormat默认的InputSplit方法。因此，我们打开FileInputFormat的源代码，在其中发现：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
 public static void setMinInputSplitSize(Job job,long size) {
     job.getConfiguration().setLong("mapred.min.split.size", size);
   }
   public static long getMinSplitSize(JobContext job) {
     return job.getConfiguration().getLong("mapred.min.split.size", 1L);
   }
    
 public static void setMaxInputSplitSize(Job job,long size) {
     job.getConfiguration().setLong("mapred.max.split.size", size);
   }
   public static long getMaxSplitSize(JobContext context) {
     return context.getConfiguration().getLong("mapred.max.split.size",Long.MAX_VALUE);
   }

如上我们可以看到，Hadoop在这里实现了对mapred.min.split.size和mapred.max.split.size的定义，且默认值分别为1和Long的最大。因此，我们在程序只需重新赋值给这两个值就可以控制InputSplit分片的大小了。
3.假如我们想要设置的分片大小为10MB
 则我们可以在MapReduce程序的驱动部分添加如下代码：

?
1
2
3
TextInputFormat.setMinInputSplitSize(job,1024L);//设置最小分片大小
 
TextInputFormat.setMaxInputSplitSize(job,1024×1024×10L);//设置最大分片大小