考究Hadoop中split的计算方法

Hadoop中block块大小和split切片大小会影响到MapReduce程序在运行过程中的效率、map的个数。在本文中，以经典入门案例WordCount为例，通过debug的方式跟踪源代码，来分析hadoop中split的计算方法。

前期准备

wc.txt的准备
单词统计，以空格为分割符对单词进行切割。因此wc.txt中的单词都以空格进行分割
mr核心部分介绍
map
提取每一行，String[] arr = line.split(” “);

for(String word : arr){context.write(new Text(word),new IntWritable(1));
}

reduce
对每一个key的个数进行迭代，计算总数。

修改HDFS的文件块大小

hdfs-default.xml
首先查看默认的配置信息
文件块的大小

<property><name>dfs.blocksize</name><value>134217728</value><description>The default block size for new files, in bytes.You can use the following suffix (case insensitive):k(kilo), m(mega), g(giga), t(tera), p(peta), e(exa) to specify the size (such as 128k, 512m, 1g, etc.),Or provide complete size in bytes (such as 134217728 for 128 MB).</description>
</property>

最小文件块的大小

<property><name>dfs.namenode.fs-limits.min-block-size</name><value>1048576</value><description>Minimum block size in bytes, enforced by the Namenode at createtime. This prevents the accidental creation of files with tiny blocksizes (and thus many blocks), which can degradeperformance.</description>
</property>

hdfs-site.xml
了解默认的配置信息之后，对其进行修改

<property><name>dfs.blocksize</name><value>512</value>   //块大小需要能够整除校验和   需要符合校验和的倍数要求(切记！！必须为512的倍数)
</property>
<property><name>dfs.namenode.fs-limits.min-block-size</name><value>10</value>
</property>

通过命令行查看属性是否修改成功

$>hdfs getconf -confKey dfs.blocksize
$>hdfs getconf -confKey dfs.namenode.fs-limits.min-block-size

关键部分源码解析

对所涉及到的关键部分，进行解析
org.apache.hadoop.mapreduce.JobSubmitter类

  private int writeSplits(org.apache.hadoop.mapreduce.JobContext job,Path jobSubmitDir) throws IOException,InterruptedException, ClassNotFoundException {// 得到配置信息JobConf jConf = (JobConf)job.getConfiguration();int maps;// 判断是否是新的Mapperif (jConf.getUseNewMapper()) {// 计算新的切片信息// 作业       job：              com.sun.jdi.InvocationException occurred invoking method.// 作业提交目录 jobSubmitDir：       /tmp/hadoop-yarn/staging/zhaotao/.staging/job_1496761683234_0001maps = writeNewSplits(job, jobSubmitDir);} else {maps = writeOldSplits(jConf, jobSubmitDir);}return maps;}...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);// 得到splits切片的集合，具体的值为：//  [hdfs://s100/data/wc.txt:0+150, hdfs://s100/data/wc.txt:150+150, hdfs://s100/data/wc.txt:300+150, hdfs://s100/data/wc.txt:450+150, hdfs://s100/data/wc.txt:600+158]总共切了5个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 firstArrays.sort(array, new SplitComparator());JobSplitWriter.createSplitFiles(jobSubmitDir, conf, jobSubmitDir.getFileSystem(conf), array);return array.length;}

org.apache.hadoop.mapreduce.lib.input.FileInputFormat类

  public List<InputSplit> getSplits(JobContext job) throws IOException {StopWatch sw = new StopWatch().start();// 通过getFormatMinSplitSize()方法  取得最小切片的大小为一个固定值(可通过配置文件配置得到)为0// 通过getMinSplitSize()方法 得到最小切片值为1// minSize的值为1long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));// maxSize的值为：9223372036854775807long maxSize = getMaxSplitSize(job);// generate splits// 创建一个集合List<InputSplit> splits = new ArrayList<InputSplit>();//  得到hdfs的文件夹列表List<FileStatus> files = listStatus(job);// 得到每个文件的状态，进行for循环for (FileStatus file: files) {// 取到文件的路径Path path = file.getPath();// 取到文件的长度long length = file.getLen();if (length != 0) {// 文件块位置的数组BlockLocation[] blkLocations;if (file instanceof LocatedFileStatus) {// 通过getBlockLocations()方法得到文件块的位置// blkLocations的值：[0,512,s103,s101,s104, 512,246,s103,s104,s102]  为两块文件块blkLocations = ((LocatedFileStatus) file).getBlockLocations();} else {FileSystem fs = path.getFileSystem(job.getConfiguration());blkLocations = fs.getFileBlockLocations(file, 0, length);}// 判断文本文件是否可切割if (isSplitable(job, path)) {// 得到块大小// blockSize的值为512// 切片大小与块大小、最小切片值、最大切片值有关long blockSize = file.getBlockSize();// 计算切片大小// blockSize=512；minSize=1；maxSize=9223372036854775807// splitSize的值为512long splitSize = computeSplitSize(blockSize, minSize, maxSize);...// 最终返回的值为一个切片集合// 具体的值为：[hdfs://s100/data/wc.txt:0+150, hdfs://s100/data/wc.txt:150+150, hdfs://s100/data/wc.txt:300+150, hdfs://s100/data/wc.txt:450+150, hdfs://s100/data/wc.txt:600+158]总共切了5个return splits;...// 得到切片大小的一个下限值，返回的是一个最小切片的字节数protected long getFormatMinSplitSize() {return 1;}...public static long getMinSplitSize(JobContext job) {// SPLIT_MINSIZE = mapreduce.input.fileinputformat.split.minsize(该配置项的默认值为0)// job.getConfiguration().getLong(SPLIT_MINSIZE, 1L) = 0return job.getConfiguration().getLong(SPLIT_MINSIZE, 1L);}...public static long getMaxSplitSize(JobContext context) {// SPLIT_MAXSIZE = mapreduce.input.fileinputformat.split.maxsize// Long.MAX_VALUE的意思为取长整型的最大值9223372036854775807// context.getConfiguration().getLong(SPLIT_MAXSIZE,Long.MAX_VALUE)的值为9223372036854775807return context.getConfiguration().getLong(SPLIT_MAXSIZE, Long.MAX_VALUE);}...protected long computeSplitSize(long blockSize, long minSize,long maxSize) {// 先在maxSize与blockSize之间选择小的，再使用该值与minSize之间取得大的值// 就是取了三个值之间的中间值return Math.max(minSize, Math.min(maxSize, blockSize));}

split计算方法

通过对源代码的跟踪，将split的计算方法总结如下：
minSize = 1(默认) (属性名字：mapreduce.input.fileinputformat.split.minsize)
maxSize = Long.MAX_VALUE(默认) (属性名字：mapreduce.input.fileinputformat.split.maxsize)
blockSize = 512 (属性名字：dfs.blocksize=128mb默认值)
计算方法：max(minSize,min(blockSize,maxSize))
根据上述数据，所以该过程中的切片大小为512