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主要类图如下：
