MapReduce数据流

1、InputFormat

InputFormat的作用有：

• 验证作业输入的正确性
• 将输入文件切割成逻辑分片(InputSplit)，一个InputSplit将会被分配给一个独立的MapTask
• 提供RecordReader实现，读取InputSplit中的“K-V对”供Mapper使用

InputFormat抽象类源码如下：

public   abstract   class   InputFormat<K, V> {
 
      /**
       * 仅仅是逻辑分片，并没有物理分片，所以每一个分片类似于这样一个元组 <input-file-path, start, offset>
       */
      public   abstract   List<InputSplit> getSplits(JobContext context)
              throws   IOException, InterruptedException;
 
      /**
       * 创建RecordReader，从InputSplit中读取数据，解析成<K,V>对
       */
      public   abstract   RecordReader<K, V> createRecordReader(InputSplit split,
              TaskAttemptContext context)   throws   IOException,
              InterruptedException;
 
}

InputFormat有两个抽象方法：

• List<InputSplit> getSplits()，将输入数据切分成Split。
• RecordReader<K,V> createRecordReader()，将InputSplit的内容转换为可以作为map输入的<k,v>键值对。

InputFormat主要类图图下：

• FileInputFormat切片机制：
  （1）简单地按照文件的内容长度进行切片
  （2）切片大小默认等于Block大小
  （3）切片时不考虑数据整体，而是逐个针对每一个文件单独切片
• TextInputFormat切片机制：
  切片方式：TextInputFormat是默认的切片机制，按文件规划进行切分。比如切片默认为128M，如果一个文件为200M，则会形成两个切片，一个是128M，一个是72M，启动两个MapTask任务进行处理任务。但是如果一个文件只有1M，也会单独启动一个MapTask执行此任务，如果是10个这样的小文件，就会启动10个MapTask处理小文件任务。
  读取方式：TextInputFormat是按行读取文件的每条记录，key代表读取的文件行在该文件中的起始字节偏移量，key为LongWritable类型；value为读取的行内容，不包括任何行终止符（换行符/回车符）, value为Text类型，相当于java中的String类型。
• CombineFileInputFormat切片机制：
  框架默认的TextInputFormat切片机制是对任务按文件规划切片，不管文件多小，都会是一个单独的切片，都会交给一个MapTask，这样如果有大量小文件，就会产生大量的MapTask，处理效率极其低下。
  （1）应用场景：CombineTextInputFormat用于小文件过多的场景，它可以将多个小文件从逻辑上规划到一个切片中，这样，多个小文件就可以交给一个MapTask处理。
  （2）切片机制：生成切片过程包括：虚拟存储过程和切片过程二部分。
• KeyValueTextInputFormat切片机制：
  KeyValueTextInputFormat与TextInputFormat相似，按行读入记录，每个文件形成一个切片，但KeyValueTextInputFormat在读入一行后可以指定切割符，把一行内容按切割符分割成键值对的形式。
  例如：

A-this is a
B-this is b
C-this is c
C-this is c

经过mapper阶段后被切割成：

(A,this is a)
(B,this is b)
(C,this is c)
(C,this is c)

• NLineInputForma切片机制：
  NLineInputFormat可以指定切分文件时按指定的行数进行切分，比如文件总行数为n，切分行数为N，则切片数为：如果n/N整除，切片数为n/N；如果不能整除，切片数为（n/N + 1）。
• SequenceFileInputFormat切片机制：
  SequenceFileInputFormat只能处理SequenceFile类型的文件，输出为普通文件

1.1 InputSplit

MapTask的并行度决定Map阶段的任务处理并发度，每一个Split切片分配一个MapTask并行实例处理。默认情况下，切片大小=BlockSize。切片时不考虑数据集整体，而是逐个针对每一个文件单独切片。看源码可知，InputSplit也是一个抽象类，它在逻辑上包含了提供给处理这个InputSplit的Mapper的所有K-V对。

public abstract class InputSplit {
      /**
       * 获取Split的大小，单位是字节
       */
      public abstract long getLength() throws IOException, InterruptedException;
 
      /**
       * 返回存储该数据块的数据节点的名称，例如：String[0]="Slave1",String[1]="Slave2".
       */
      public abstract
        String[] getLocations() throws IOException, InterruptedException;
}

InputSplit主要类图如下：

1.2 RecordReader

RecordReader的作用就是将数据切分成key/value的形式然后作为输入传给Mapper。

public abstract class RecordReader<KEYIN, VALUEIN> implements Closeable {
 
    /**
     * 初始化RecordReader，框架会在开始的时候调用一次
     */
    public abstract void initialize(InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException;
 
    /**
     * 读取分片下一个<key, value>对
     */
    public abstract boolean nextKeyValue() throws IOException,
            InterruptedException;
 
    /**
     * 获取当前的key
     */
    public abstract KEYIN getCurrentKey() throws IOException,
            InterruptedException;
 
    /**
     * 获取当前的value
     */
    public abstract VALUEIN getCurrentValue() throws IOException,
            InterruptedException;
 
    /**
     * 跟踪读取分片的进度
     */
    public abstract float getProgress() throws IOException,
            InterruptedException;
 
    /**
     * 关闭RecordReader
     */
    public abstract void close() throws IOException;
}

RecordReader主要类图如下：