Hadoop之InputFormat數據輸入詳解

---

Job提交流程和切片源碼詳解

FileInputFormat切片機制

CombineTextInputFormat切片機制

InputFormat接口實現類

自定義InputFormat

---

1. Job提交流程和切片源碼詳解

job提交流程源碼詳解，如下

waitForCompletion() submit(); // 1建立連接connect(); // 1）創建提交job的代理new Cluster(getConfiguration());// （1）判斷是本地yarn還是遠程initialize(jobTrackAddr, conf); // 2 提交job submitter.submitJobInternal(Job.this, cluster)// 1）創建給集群提交數據的Stag路徑Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);// 2）獲取jobid ，并創建job路徑JobID jobId = submitClient.getNewJobID();// 3）拷貝jar包到集群 copyAndConfigureFiles(job, submitJobDir); rUploader.uploadFiles(job, jobSubmitDir); // 4）計算切片，生成切片規劃文件 writeSplits(job, submitJobDir);maps = writeNewSplits(job, jobSubmitDir);input.getSplits(job); // 5）向Stag路徑寫xml配置文件 writeConf(conf, submitJobFile);conf.writeXml(out); // 6）提交job,返回提交狀態 status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());

FileInputFormat源碼解析(input.getSplits(job))

找到你數據存儲的目錄。

開始遍歷處理（規劃切片）目錄下的每一個文件

遍歷第一個文件ss.txt

獲取文件大小fs.sizeOf(ss.txt)

計算切片大小 computeSliteSize(Math.max(minSize,Math.min(maxSize,blocksize)))=blocksize=128M

默認情況下，切片大小=blocksize

開始切，形成第1個切片：ss.txt—0:128M 第2個切片ss.txt—128:256M 第3個切片ss.txt—256M:300M（每次切片時，都要判斷切完剩下的部分是否大于塊的1.1倍，不大于1.1倍就劃分一塊切片）

將切片信息寫到一個切片規劃文件中

整個切片的核心過程在getSplit()方法中完成

數據切片只是在邏輯上對輸入數據進行分片，并不會在磁盤上將其切分成分片進行存儲。InputSplit只記錄了分片的元數據信息，比如起始位置、長度以及所在的節點列表等

注意：block是HDFS物理上存儲的數據，切片是對數據邏輯上的劃分

提交切片規劃文件到yarn上，yarn上的MrAppMaster就可以根據切片規劃文件計算開啟maptask個數。

---

2. FileInputFormat切片機制

FileInputFormat中默認的切片機制

簡單地按照文件的內容長度進行切片

切片大小，默認等于block大小

切片時不考慮數據集整體，而是逐個針對每一個文件單獨切片

比如待處理數據有兩個文件：

file1.txt 320M file2.txt 10M

經過FileInputFormat的切片機制運算后，形成的切片信息如下：

file1.txt.split1-- 0~128 file1.txt.split2-- 128~256 file1.txt.split3-- 256~320 file2.txt.split1-- 0~10M

---

FileInputFormat切片大小的參數配置

通過分析源碼，在FileInputFormat中，計算切片大小的邏輯：
Math.max(minSize, Math.min(maxSize, blockSize));

切片主要由這幾個值來運算決定
mapreduce.input.fileinputformat.split.minsize=1 默認值為1
mapreduce.input.fileinputformat.split.maxsize= Long.MAXValue 默認值Long.MAXValue
因此，默認情況下，切片大小=blocksize。
maxsize（切片最大值）：參數如果調得比blocksize小，則會讓切片變小，而且就等于配置的這個參數的值。
minsize（切片最小值）：參數調的比blockSize大，則可以讓切片變得比blocksize還大。

---

獲取切片信息API

// 根據文件類型獲取切片信息 FileSplit inputSplit = (FileSplit) context.getInputSplit(); // 獲取切片的文件名稱 String name = inputSplit.getPath().getName();

---

3. CombineTextInputFormat切片機制

關于大量小文件的優化策略
默認情況下TextInputformat對任務的切片機制是按文件規劃切片，不管文件多小，都會是一個單獨的切片，都會交給一個maptask，這樣如果有大量小文件，就會產生大量的maptask，處理效率極其低下。

優化策略 多種辦法

最好的辦法，在數據處理系統的最前端（預處理/采集），將小文件先合并成大文件，再上傳到HDFS做后續分析。

補救措施：如果已經是大量小文件在HDFS中了，可以使用另一種InputFormat來做切片（CombineTextInputFormat），它的切片邏輯跟TextFileInputFormat不同：它可以將多個小文件從邏輯上規劃到一個切片中，這樣，多個小文件就可以交給一個maptask。

優先滿足最小切片大小，不超過最大切片大小
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m
CombineTextInputFormat.setMinInputSplitSize(job, 2097152);// 2m
舉例：0.5m+1m+0.3m+5m=2m + 4.8m=2m + 4m + 0.8m

具體實現步驟

// 如果不設置InputFormat,它默認用的是TextInputFormat.class job.setInputFormatClass(CombineTextInputFormat.class) CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m CombineTextInputFormat.setMinInputSplitSize(job, 2097152);// 2m

---

4. InputFormat接口實現類

MapReduce任務的輸入文件一般是存儲在HDFS里面。輸入的文件格式包括：基于行的日志文件、二進制格式文件等。這些文件一般會很大，達到數十GB，甚至更大。那么MapReduce是如何讀取這些數據的呢？下面我們首先學習InputFormat接口。

InputFormat常見的接口實現類包括：TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定義InputFormat等。

TextInputFormat
TextInputFormat是默認的InputFormat。每條記錄是一行輸入。鍵是LongWritable類型，存儲該行在整個文件中的起始字節偏移量。值是這行的內容，不包括任何行終止符（換行符和回車符）。
以下是一個示例，比如，一個分片包含了如下4條文本記錄。

Rich learning form Intelligent learning engine Learning more convenient From the real demand for more close to the enterprise

每條記錄表示為以下鍵/值對：

(0,Rich learning form) (19,Intelligent learning engine) (47,Learning more convenient) (72,From the real demand for more close to the enterprise)

很明顯，鍵并不是行號。一般情況下，很難取得行號，因為文件按字節而不是按行切分為分片。

---

KeyValueTextInputFormat
每一行均為一條記錄，被分隔符分割為key，value。可以通過在驅動類中設置conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, " ");來設定分隔符。默認分隔符是tab（\t）。

以下是一個示例，輸入是一個包含4條記錄的分片。其中——>表示一個（水平方向的）制表符。

line1 ——>Rich learning form line2 ——>Intelligent learning engine line3 ——>Learning more convenient line4 ——>From the real demand for more close to the enterprise

每條記錄表示為以下鍵/值對：

(line1,Rich learning form) (line2,Intelligent learning engine) (line3,Learning more convenient) (line4,From the real demand for more close to the enterprise)

此時的鍵是每行排在制表符之前的Text序列。

---

NLineInputFormat
如果使用NlineInputFormat，代表每個map進程處理的InputSplit不再按block塊去劃分，而是按NlineInputFormat指定的行數N來劃分。即輸入文件的總行數/N=切片數，如果不整除，切片數=商+1。

以下是一個示例，仍然以上面的4行輸入為例。

Rich learning form Intelligent learning engine Learning more convenient From the real demand for more close to the enterprise

例如，如果N是2，則每個輸入分片包含兩行。開啟2個maptask。

(0,Rich learning form) (19,Intelligent learning engine)

另一個 mapper 則收到后兩行：

(47,Learning more convenient) (72,From the real demand for more close to the enterprise)

這里的鍵和值與TextInputFormat生成的一樣。

---

5. 自定義InputFormat

概述

自定義一個類繼承FileInputFormat。

改寫RecordReader，實現一次讀取一個完整文件封裝為KV。

在輸出時使用SequenceFileOutPutFormat輸出合并文件。

總結

以上是生活随笔為你收集整理的Hadoop之InputFormat数据输入详解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。