十、MapReduce--InputFormat以及RecordReader抽象類

一、基本原理

? 在map執(zhí)行之前，需要將數(shù)據(jù)進行切片，每個切片對應(yīng)一個map任務(wù)。而每個map任務(wù)并不是直接處理這些切片數(shù)據(jù)的，它是處理KV的。所以問題有兩個：數(shù)據(jù)是如何切片的、切片是如何轉(zhuǎn)為KV給map處理的。
? 這就涉及到兩個抽象類，InputFormat以及 RecordReader。具體為什么是這兩個抽象類，請看之前input的源碼分析

1、InputFormat

public abstract class InputFormat<K, V> {
    public InputFormat() {
    }

    public abstract List<InputSplit> getSplits(JobContext var1) throws IOException, InterruptedException;

    public abstract RecordReader<K, V> createRecordReader(InputSplit var1, TaskAttemptContext var2) throws IOException, InterruptedException;
}

我們看到，這個抽象類就兩個方法
getSplits：看名字就知道是用來將數(shù)據(jù)處理成切片的了
createRecordReader：就是用來創(chuàng)建RecordReader對象的。
所以這就是一個InputFormat基本的功能

2、 RecordReader

public abstract class RecordReader<KEYIN, VALUEIN> implements Closeable {
    public RecordReader() {
    }

    //初始化，一般就是讀取切片的數(shù)據(jù)
    public abstract void initialize(InputSplit var1, TaskAttemptContext var2) throws IOException, InterruptedException;

    //檢查是否還有下一對KV，并且如果有，實際上會將其處理成KV，并賦值給this.key和this.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;

    //關(guān)閉reader
    public abstract void close() throws IOException;
}

這個抽象類就涉及到讀取切片的數(shù)據(jù)，處理成KV結(jié)構(gòu)。而在input源碼分析中說到，mapper.run方法中通過 context.getCurrentKey() 類似的方法獲取key其實就是調(diào)用這個RecordReader中的這些get方法而已。

3、InputFormat以及 RecordReader的關(guān)系

從上面的源碼可以看到。
InputFormat：負(fù)責(zé)規(guī)劃切片信息，以及創(chuàng)建RecordReader對象
RecordReader：負(fù)責(zé)按照切片規(guī)劃去讀取當(dāng)前mapper處理的切片數(shù)據(jù)，并將其處理成KV形式，然后通過context傳遞給mapper。

二、InputFormat以及 RecordReader常用實現(xiàn)類

常用的有：TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定義InputFormat（自定義有另外的文章講）

1、TextInputFormat

? 這是默認(rèn)的InputFormat，切片方式是按數(shù)據(jù)塊的方式切割，默認(rèn)大小block大小。一個文件至少是一個切片（無論多小）。因為這個類繼承FileInputFormat，使用的是其父類定義的getsplit() 方法進行切片。
? 使用的RecordReader是LineRecordReader。處理切片成KV時，每條記錄是一行輸入。鍵K是LongWritable類型，存儲該行在整個文件中的字節(jié)偏移量。值是這行的內(nèi)容，不包括任何行終止符（換行符和回車符）。

2、KeyValueTextInputFormat

? 這個類也是使用父類FileInputFormat的getsplit() 方法進行切片，所以切片方式和上面一致。
? 使用的RecordReader是KeyValueLineRecordReader。每一行均為一條記錄，被分隔符分割為key，value?？梢酝ㄟ^在驅(qū)動類中設(shè)置conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, " ");來設(shè)定分隔符。默認(rèn)分隔符是tab（\t）。

3、NLineInputFormat

? 這個類雖然繼承了FileInputFormat，但是自己重寫了getSplit方法，使用另外的方式來切片。是按指定的行數(shù)來切片，比如5行，那就5行作為一個切片，無論數(shù)據(jù)大小。通過mapreduce.input.lineinputformat.linespermap 這個參數(shù)設(shè)置切片行數(shù)。
? 使用的RecordReader是LineRecordReader。和上面類似，不重復(fù)說。

4、CombineTextInputFormat

? 這個類繼承于 CombineFileInputFormat，父類繼承于FileInputFormat。在CombineFileInputFormat中重寫了 getSplits方法。因為FileInputFormat默認(rèn)無論多小的文件，一個文件至少是一個切片。如果遇到很多小文件，就會導(dǎo)致很多切片。而這里的切片方式就是嚴(yán)格按照大小來切片，會將小文件集合在一起，達(dá)到指定大小，才作為一個切片。
? 使用的RecordReader是CombineFileRecordReader。處理方式和 LineRecordReader類似，只不過切片可能是來自多個文件，讀取方式上略顯麻煩。

三、設(shè)置使用指定的inputformat

job.setInputFormatClass(xxxInputFormat.class);