Hadoop TextInputFormat源码分析

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Hadoop TextInputFormat源码分析相关的知识,希望对你有一定的参考价值。

InputFormat主要用于描述输入数据的格式(我们只分析新API,即org.apache.hadoop.mapreduce.lib.input.InputFormat),提供以下两个功能:

(1).数据切分:按照某个策略将输入数据切分成若干个split,以便确定MapTask个数以及对应的split。

(2).为Mapper提供输入数据:读取给定的split的数据,解析成一个个的key/value对,共Mapper使用。

InputFormat抽象类中只有两个方法,分别对应上面两个功能,源码如下:

 

[java] view plain copy
 
  1. public abstract class InputFormat<K, V> {  
  2.   
  3.   
  4.   public abstract   
  5.     List<InputSplit> getSplits(JobContext context  
  6.                                ) throws IOException, InterruptedException;  
  7.     
  8.   
  9.   public abstract   
  10.     RecordReader<K,V> createRecordReader(InputSplit split,  
  11.                                          TaskAttemptContext context  
  12.                                         ) throws IOException,   
  13.                                                  InterruptedException;  
  14.   
  15. }  


getSplits()方法返回的是InputSplit类型的集合,InputSplit分片由两个特点:

 

(1):逻辑分片即只是在逻辑上对数据进行分片,并不进行物理切分,这点和block是不同的,它只记录一些元信息,比如起始位置、长度以及所在的节点列表等;

(2):必须可序列化,分片信息要上传到HDFS文件系统,还会被JobTracker读取,序列化可以方便进程通信以及永久存储。

createRecordReader()方法返回的是RecordReader对象,该对象可以将输入数据,即InputSplit对应的数据解析成众多的key/value对,会作为MapTask的map方法的输入。

 

我们本节就一最常用的TextInputFormat为例讲解分片和读取分片数据。

我们先来看看,TextInputFormat、FileInputFormat和InputFormat三者之间的关系,如下:

 

[java] view plain copy
 
  1. public class TextInputFormat extends FileInputFormat;  
  2. public abstract class FileInputFormat<K, V> extends InputFormat;  
  3. public abstract class InputFormat。  

最顶的父类InputFormat只有两个未实现的抽象方法getSplits()和createRecordReader();而FileInputFormat包含的方法比较多,如下图:

 

技术分享

我们在自己的MapReduce程序中设置输入目录就是调用这里面的方法。

TextInputFormat这个类只有两个方法,源码如下:

 

[java] view plain copy
 
  1. /** An {@link InputFormat} for plain text files.  Files are broken into lines. 
  2.  * Either linefeed or carriage-return are used to signal end of line.  Keys are 
  3.  * the position in the file, and values are the line of text.. */  
  4. public class TextInputFormat extends FileInputFormat<LongWritable, Text> {  
  5.   
  6.   @Override  
  7.   public RecordReader<LongWritable, Text>   
  8.     createRecordReader(InputSplit split,  
  9.                        TaskAttemptContext context) {  
  10.     return new LineRecordReader();  
  11.   }  
  12.   
  13.   @Override  
  14.   protected boolean isSplitable(JobContext context, Path file) {  
  15.     CompressionCodec codec =   
  16.       new CompressionCodecFactory(context.getConfiguration()).getCodec(file);  
  17.     if (null == codec) {  
  18.       return true;  
  19.     }  
  20.     return codec instanceof SplittableCompressionCodec;  
  21.   }  
  22.   
  23. }  

isSplitable方法表示是否要切分文件,这个方法显示如果是压缩文件就不切分,非压缩文件就切分。

 

接下来,我们只关注那两个主要方法,首先来看:

一:getSplits()方法,这个方法在FileInputFormat类中,它的子类一般只需要实现TextInputFormat中的两个方法而已,getSplits()方法代码如下:

 

[java] view plain copy
 
  1. /**  
  2.    * Generate the list of files and make them into FileSplits. 
  3.    */   
  4.   public List<InputSplit> getSplits(JobContext job  
  5.                                     ) throws IOException {  
  6.     long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));  
  7.     long maxSize = getMaxSplitSize(job);    //Long.MAX_VALUE  
  8.   
  9.     // generate splits  
  10.     List<InputSplit> splits = new ArrayList<InputSplit>();  
  11.     List<FileStatus>files = listStatus(job);  
  12.     for (FileStatus file: files) {  
  13.       Path path = file.getPath();  
  14.       FileSystem fs = path.getFileSystem(job.getConfiguration());  
  15.       long length = file.getLen();    //整个文件的长度  
  16.       BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);  
  17.       if ((length != 0) && isSplitable(job, path)) {    //默认是true,但是如果是压缩的,则是false  
  18.         long blockSize = file.getBlockSize();        //64M,67108864B  
  19.         long splitSize = computeSplitSize(blockSize, minSize, maxSize);    //计算split大小  Math.max(minSize, Math.min(maxSize, blockSize))  
  20.   
  21.         long bytesRemaining = length;              
  22.         while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {  
  23.           int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);  
  24.           splits.add(new FileSplit(path, length-bytesRemaining, splitSize,   
  25.                                    blkLocations[blkIndex].getHosts()));        //hosts是主机名,name是IP  
  26.           bytesRemaining -= splitSize;        //剩余块的大小  
  27.         }  
  28.           
  29.         if (bytesRemaining != 0) {    //最后一个  
  30.           splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining,   
  31.                      blkLocations[blkLocations.length-1].getHosts()));  
  32.         }  
  33.       } else if (length != 0) {    //isSplitable(job, path)等于false  
  34.         splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));  
  35.       } else {  
  36.         //Create empty hosts array for zero length files  
  37.         splits.add(new FileSplit(path, 0, length, new String[0]));  
  38.       }  
  39.     }  
  40.       
  41.     // Save the number of input files in the job-conf  
  42.     job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());  
  43.   
  44.     LOG.debug("Total # of splits: " + splits.size());  
  45.     return splits;  
  46.   }  

代码分析:

 

(1):minSize=Math.max(getFormatMinSplitSize(),getMinSplitSize(job)):

getFormatMinSplitSize()返回了一个固定值即1;getMinSplitSize(job)是获取在mapred-site.xml文件中"mapred.min.split.size"属性配置的值,默认是0。

(2):maxSize=getMaxSplitSize(job):

getMaxSplitSize(job)是获取"mapred.max.split.size"属性配置的值,默认是Long.MAX_VALUE,即Long类型的最大值(注:这个属性在mapred-site.xml文件中并没有,所以不推荐配置)。

(3):接下来是遍历输入目录下所有文件的FileStatus信息列表。

(4):然后对每一个文件获取它的目录、文件的长度、文件对应的所有块信息(可能有多个块,每个块对应3个副本);

(5):然后如果文件长度不为0且支持分割(isSplitable方法等于true):获取block大小,默认是64MB,通过方法computeSplitSize(blockSize,minSize,maxSize)计算分片的大小,这个方法的源码返回Max.max(minSize,Math.min(maxSize,blockSize))。

(6):将bytesRemaiining(剩余分片字节数)设置为整个文件的长度。如果bytesRemaining超过分片大小splitSize一定量才会将文件分成多个InputSplit即当bytesRemaing/splitSize>SPLIT_SLOP(SPLIT_SLOP是固定值为1.1)时进入while循环执行getBlockIndex(blkLocations, length-bytesRemaining)获取block的索引,第二个参数就是这个block在整个文件中的偏移量,在循环中会从0越来越大,该方法代码如下:

 

[java] view plain copy
 
  1. protected int getBlockIndex(BlockLocation[] blkLocations,   
  2.                             long offset) {  
  3.   for (int i = 0 ; i < blkLocations.length; i++) {  
  4.     // is the offset inside this block?  
  5.     if ((blkLocations[i].getOffset() <= offset) &&  
  6.         (offset < blkLocations[i].getOffset() + blkLocations[i].getLength())){  
  7.       return i;  
  8.     }  
  9.   }  
  10.   BlockLocation last = blkLocations[blkLocations.length -1];  
  11.   long fileLength = last.getOffset() + last.getLength() -1;  
  12.   throw new IllegalArgumentException("Offset " + offset +   
  13.                                      " is outside of file (0.." +  
  14.                                      fileLength + ")");  
  15. }  

这个方法中的if语句的条件会限制获取到这个偏移量对应的block的索引。

 

(7)将这个索引对应的block信息的主机节点以及文件的路径名、开始的偏移量、分片大小splitSize封装到一个InputSplit中加入List<InputSplit> splits中。

(8)bytesRemaining -= splitSize是修改剩余字节大小。

(9)判断循环是否继续,如果满足条件则继续执行循环条件,否则跳出循环。

(10)跳出循环之后如果剩余bytesRemaining还不为0,表示还有未分配的数据,将剩余的数据及最后一个block加入到splits集合中。

(11)自此,我们已经走完了getSplits()方法中的第一个if条件,下面说第二个if条件,当不允许分割即isSplitable==false,则将第一个block、文件目录、开始位置0,长度为整个文件的长度封装到一个InputSplit中,加入到splits中。

(12)执行else条件即当文件的长度==0时,则会splits.add(new FileSplit(path,0,length,new String[0]))没有block,并且初始和长度都为0。

(13)将输入目录下文件的个数赋值给"mapreduce.input.num.files",方便以后的校对。

(14) 返回分片信息splits。

以上就是getSplits获取分片的过程,当使用基于FileInputFormat继承InputFormat时,为了提高MapTask的数据本地性,应尽量使InputSplit大小与block大小相同,因为当一个分片包含多个block的时候,总会从其他节点读取数据,也就是做不到所有的计算都是本地化,为了发挥计算本地化性能,应该尽量使InputSplit大小和块大小相当。

特殊问题:就是如果分片大小超过block大小,但是InputSplit中封装了单个block的所在主机信息啊,这样能读取多个block数据吗?这个问题留到最后讲解。


二:createRecordReader()方法,该方法返回一个RecordReader对象,实现了类似的迭代功能,将某个InputSplit解析成一个个key/value对。RecordReader应该注意两点:

(1):定位记录边界:为了能识别一条完整的记录,应该添加一些同步标示,TextInputFormat的标示是换行符;SequenceFileInputFormat的标示是每隔若干条记录会添加固定长度的同步字符串。为了解决InputSplit中第一条或者最后一条可能跨InputSplit的情况,RecordReader规定每个InputSplit的第一条不完整记录划给前一个InputSplit。

(2):解析key/value:将每个记录分解成key和value两个部分,TextInputFormat每一行的内容是value,该行在整个文件中的偏移量为key;SequenceFileInputFormat的记录共有四个字段组成:前两个字段分别是整个记录的长度和key的长度,均为4字节,后两个字段分别是key和value的内容。

TextInputFormat使用的RecordReader是org.apache.hadoop.mapreduce.lib.input.LineRecordReader。这个类中的方法:首先是initialize()方法,该方法主要是获取分片信息的初始化位置和结束位置,以及输入流(若没有则是压缩流);mapper的key/value是通过LineRecordReader.nextKeyValue()方法将key和value读取到key和value中的,在这个方法中key被设置为在文件中的偏移量,value通过LineReader.readLine(value,maxLineLength,Math.max((int)Math.min(Integer.Max_VALUE,end-pos),maxLineLength))这个方法读取一行数据放入value中,方法代码如下:

 

[java] view plain copy
 
  1. /** 
  2.    * Read one line from the InputStream into the given Text.  A line 
  3.    * can be terminated by one of the following: ‘\n‘ (LF) , ‘\r‘ (CR), 
  4.    * or ‘\r\n‘ (CR+LF).  EOF also terminates an otherwise unterminated 
  5.    * line. 
  6.    * 
  7.    * @param str the object to store the given line (without newline) 
  8.    * @param maxLineLength the maximum number of bytes to store into str; 
  9.    *  the rest of the line is silently discarded. 
  10.    * @param maxBytesToConsume the maximum number of bytes to consume 
  11.    *  in this call.  This is only a hint, because if the line cross 
  12.    *  this threshold, we allow it to happen.  It can overshoot 
  13.    *  potentially by as much as one buffer length. 
  14.    * 
  15.    * @return the number of bytes read including the (longest) newline 
  16.    * found. 
  17.    * 
  18.    * @throws IOException if the underlying stream throws 
  19.    */  
  20.   public int readLine(Text str, int maxLineLength,  
  21.                       int maxBytesToConsume) throws IOException {  
  22.     /* We‘re reading data from in, but the head of the stream may be 
  23.      * already buffered in buffer, so we have several cases: 
  24.      * 1. No newline characters are in the buffer, so we need to copy 
  25.      *    everything and read another buffer from the stream. 
  26.      * 2. An unambiguously terminated line is in buffer, so we just 
  27.      *    copy to str. 
  28.      * 3. Ambiguously terminated line is in buffer, i.e. buffer ends 
  29.      *    in CR.  In this case we copy everything up to CR to str, but 
  30.      *    we also need to see what follows CR: if it‘s LF, then we 
  31.      *    need consume LF as well, so next call to readLine will read 
  32.      *    from after that. 
  33.      * We use a flag prevCharCR to signal if previous character was CR 
  34.      * and, if it happens to be at the end of the buffer, delay 
  35.      * consuming it until we have a chance to look at the char that 
  36.      * follows. 
  37.      */  
  38.     str.clear();  
  39.     int txtLength = 0; //tracks str.getLength(), as an optimization  
  40.     int newlineLength = 0; //length of terminating newline  
  41.     boolean prevCharCR = false; //true of prev char was CR  
  42.     long bytesConsumed = 0;  
  43.     do {  
  44.       int startPosn = bufferPosn; //starting from where we left off the last time  
  45.       if (bufferPosn >= bufferLength) {  
  46.         startPosn = bufferPosn = 0;  
  47.         if (prevCharCR)  
  48.           ++bytesConsumed; //account for CR from previous read  
  49.         bufferLength = in.read(buffer);    //从输入流中读取一定数量的字节,并将其存储在缓冲区数组 b 中。以整数形式返回实际读取的字节数。  
  50.         if (bufferLength <= 0)        //结束了,没数据了  
  51.           break; // EOF  
  52.       }  
  53.       //‘\n‘,ASCII码:10,意义:换行NL;;;;‘\r‘ ,ASCII码:13,意义: 回车CR  
  54.       for (; bufferPosn < bufferLength; ++bufferPosn) { //search for newline  
  55.         if (buffer[bufferPosn] == LF) {            //如果是换行字符\n  
  56.           newlineLength = (prevCharCR) ? 2 : 1;  
  57.           ++bufferPosn; // at next invocation proceed from following byte,越过换行字符  
  58.           break;  
  59.         }  
  60.         if (prevCharCR) { //CR + notLF, we are at notLF,如果是回车字符\r  
  61.           newlineLength = 1;  
  62.           break;  
  63.         }  
  64.         prevCharCR = (buffer[bufferPosn] == CR);  
  65.       }  
  66.       int readLength = bufferPosn - startPosn;  
  67.       if (prevCharCR && newlineLength == 0)    //表示还没遇到换行,有回车字符,且缓存最后一个是\r  
  68.         --readLength; //CR at the end of the buffer  
  69.       bytesConsumed += readLength;  
  70.       int appendLength = readLength - newlineLength;    //newlineLength换行符个数  
  71.       if (appendLength > maxLineLength - txtLength) {  
  72.         appendLength = maxLineLength - txtLength;  
  73.       }  
  74.       if (appendLength > 0) {  
  75.         str.append(buffer, startPosn, appendLength);    //将数据加入str  
  76.         txtLength += appendLength;  
  77.       }  
  78.     } while (newlineLength == 0 && bytesConsumed < maxBytesToConsume);//循环条件没有换行并且没超过上限  
  79.   
  80.     if (bytesConsumed > (long)Integer.MAX_VALUE)  
  81.       throw new IOException("Too many bytes before newline: " + bytesConsumed);      
  82.     return (int)bytesConsumed;  
  83.   }  

这个方法的目的就是读取一行记录写入str中,bytesConsumed记录为读取的字节总数;bufferLength=in.read(buffer)从输入流读取bufferLength长度的字节数据放入buffer中;do-while中开始部分的if语句要保证将bufferLength个字节数据处理完毕之后再从输入流中读取下一行数据;newlineLength表示换行的标记符长度(0、1、2三种值),因为不同的系统换行标记可能不同,有三种:\r(回车符)、\n(换行符)、\r\n(\n:Unix系统换行符末结束符;\r\n:window系统行末结束符;\r:MAC OS系统行末结束符)。

 

接下来是进入for循环,for循环会挨个检查字符是否是\r\n,如果是回车符\r,还会将prevCharCR设置为true,当前字符如果是换行符\n,prevCharR==true时(表示上一个字符是回车符)则newlineLength=2(这表明当前系统的换行标记是\r\n),prevcharCR==false(表示上一个字符不是回车符)则newlineLength=1(这表明当前系统的换行标记是\n),并退出for循环;如果当前字符不是换行符\n且prevCharCR==true(表明当前系统的换行标记是\r)则newlineLength=1并退出for循环;这样就找到了换行标记,然后计算数据的长度appendLength(不包含换行符),将buffer中指定位置开始长度为appendLength的数据追加到str(这里其实就是value)中;txtLength表示的是str(这里其实是value中值的长度)。

do-while循环的条件是:

(1):没有发现换行标记即newlineLength==0;

(2):读取的字节数量没有超过上限即bytesconsumed <maxBytesToConsume。

当这两个条件要同时满足。这其中有个问题就是当前系统的换行标记是\r\n,但是这两个字符没有同时出现在这次读取的数据之中,\n在下一个批次中,这没有关系,上面的for循环会检查\r出现之后的下一个字符是否是\n再对newliineLength进行设置的。从这个方法可以看出,即使是记录跨split、跨block也不能阻止它完整读取一行数据的决心啊。

 

我们再回来看看LineRecordReader.nextKeyValue()方法,这个方法的代码如下:

 

[java] view plain copy
 
  1. public boolean nextKeyValue() throws IOException {  
  2.    if (key == null) {  
  3.      key = new LongWritable();  
  4.    }  
  5.    key.set(pos);  
  6.    if (value == null) {  
  7.      value = new Text();  
  8.    }  
  9.    int newSize = 0;  
  10.    // We always read one extra line, which lies outside the upper  
  11.    // split limit i.e. (end - 1)  
  12.    while (getFilePosition() <= end) {  
  13.      newSize = in.readLine(value, maxLineLength,  
  14.          Math.max(maxBytesToConsume(pos), maxLineLength));  
  15.      if (newSize == 0) {  
  16.        break;  
  17.      }  
  18.      pos += newSize;  
  19.      if (newSize < maxLineLength) {  
  20.        break;  
  21.      }  
  22.   
  23.      // line too long. try again  
  24.      LOG.info("Skipped line of size " + newSize + " at pos " +   
  25.               (pos - newSize));  
  26.    }  
  27.    if (newSize == 0) {  
  28.      key = null;  
  29.      value = null;  
  30.      return false;  
  31.    } else {  
  32.      return true;  
  33.    }  
  34.  }  

这个方法会控制split读取数据的结束位置,上面的readLine()方法只关注输入流不会管split的大小的。需要注意的是其中的while循环,其中的pos和end表示当前在文件中的偏移量和split的结束位置,即使这个split的最后一行跨split也会完整的读取一行。也就保证了一个记录的完整性。mapper获取key/value会通过getCurrentKey()和getCurrentValue()来达到。但是调用这两个方法前得先调用nextKeyValue()方法才能实现key和value的赋值。

 

 

到这里我们回头看看上面那个特殊问题,就是split的大小超过block的大小数据读取的问题,我们前面已经讲过split是逻辑分片,不是物理分片,当MapTask的数据本地性发挥作用时,会从本机的block开始读取,超过这个block的部分可能还在本机也可能不再本机,如果是后者的话就要从别的节点拉数据过来,因为实际数据是一个输入流,这个输入流面向整个问加你,不受什么block啊、split的影响,split越大可能需要从别的节点拉的数据也越大,从而效率也会越慢,拉数据的多少是由getSplits()方法中splitSize决定的。

 

至此,TextInputFormat的分片和数据读取过程就讲完了。这只是一个例子,其他InputFormat可以参考这个。

以上是关于Hadoop TextInputFormat源码分析的主要内容,如果未能解决你的问题,请参考以下文章

在 spark 中设置 textinputformat.record.delimiter

大数据之Hadoop(MapReduce):FileInputFormat实现类

大数据技术之Hadoop(MapReduce)框架原理数据压缩

大数据技术之Hadoop(MapReduce)框架原理数据压缩

4 weekend110的textinputformat对切片规划的源码分析 + 倒排索引的mr实现 + 多个job在同一个main方法中提交

hadoop离线day06--Hadoop MapReduceHDFS高阶