Hadoop之InputFormat數(shù)據(jù)輸入詳解

---

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

FileInputFormat切片機(jī)制

CombineTextInputFormat切片機(jī)制

InputFormat接口實(shí)現(xiàn)類

自定義InputFormat

---

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

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

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

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

找到你數(shù)據(jù)存儲(chǔ)的目錄。

開始遍歷處理（規(guī)劃切片）目錄下的每一個(gè)文件

遍歷第一個(gè)文件ss.txt

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

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

默認(rèn)情況下，切片大小=blocksize

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

將切片信息寫到一個(gè)切片規(guī)劃文件中

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

數(shù)據(jù)切片只是在邏輯上對(duì)輸入數(shù)據(jù)進(jìn)行分片，并不會(huì)在磁盤上將其切分成分片進(jìn)行存儲(chǔ)。InputSplit只記錄了分片的元數(shù)據(jù)信息，比如起始位置、長(zhǎng)度以及所在的節(jié)點(diǎn)列表等

注意：block是HDFS物理上存儲(chǔ)的數(shù)據(jù)，切片是對(duì)數(shù)據(jù)邏輯上的劃分

提交切片規(guī)劃文件到y(tǒng)arn上，yarn上的MrAppMaster就可以根據(jù)切片規(guī)劃文件計(jì)算開啟maptask個(gè)數(shù)。

---

2. FileInputFormat切片機(jī)制

FileInputFormat中默認(rèn)的切片機(jī)制

簡(jiǎn)單地按照文件的內(nèi)容長(zhǎng)度進(jìn)行切片

切片大小，默認(rèn)等于block大小

切片時(shí)不考慮數(shù)據(jù)集整體，而是逐個(gè)針對(duì)每一個(gè)文件單獨(dú)切片

比如待處理數(shù)據(jù)有兩個(gè)文件：

file1.txt 320M file2.txt 10M

經(jīng)過FileInputFormat的切片機(jī)制運(yùn)算后，形成的切片信息如下：

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

---

FileInputFormat切片大小的參數(shù)配置

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

切片主要由這幾個(gè)值來(lái)運(yùn)算決定
mapreduce.input.fileinputformat.split.minsize=1 默認(rèn)值為1
mapreduce.input.fileinputformat.split.maxsize= Long.MAXValue 默認(rèn)值Long.MAXValue
因此，默認(rèn)情況下，切片大小=blocksize。
maxsize（切片最大值）：參數(shù)如果調(diào)得比blocksize小，則會(huì)讓切片變小，而且就等于配置的這個(gè)參數(shù)的值。
minsize（切片最小值）：參數(shù)調(diào)的比blockSize大，則可以讓切片變得比blocksize還大。

---

獲取切片信息API

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

---

3. CombineTextInputFormat切片機(jī)制

關(guān)于大量小文件的優(yōu)化策略
默認(rèn)情況下TextInputformat對(duì)任務(wù)的切片機(jī)制是按文件規(guī)劃切片，不管文件多小，都會(huì)是一個(gè)單獨(dú)的切片，都會(huì)交給一個(gè)maptask，這樣如果有大量小文件，就會(huì)產(chǎn)生大量的maptask，處理效率極其低下。

優(yōu)化策略 多種辦法

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

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

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

具體實(shí)現(xiàn)步驟

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

---

4. InputFormat接口實(shí)現(xiàn)類

MapReduce任務(wù)的輸入文件一般是存儲(chǔ)在HDFS里面。輸入的文件格式包括：基于行的日志文件、二進(jìn)制格式文件等。這些文件一般會(huì)很大，達(dá)到數(shù)十GB，甚至更大。那么MapReduce是如何讀取這些數(shù)據(jù)的呢？下面我們首先學(xué)習(xí)InputFormat接口。

InputFormat常見的接口實(shí)現(xiàn)類包括：TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定義InputFormat等。

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

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

每條記錄表示為以下鍵/值對(duì)：

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

很明顯，鍵并不是行號(hào)。一般情況下，很難取得行號(hào)，因?yàn)槲募醋止?jié)而不是按行切分為分片。

---

KeyValueTextInputFormat
每一行均為一條記錄，被分隔符分割為key，value?？梢酝ㄟ^在驅(qū)動(dòng)類中設(shè)置conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, " ");來(lái)設(shè)定分隔符。默認(rèn)分隔符是tab（\t）。

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

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

每條記錄表示為以下鍵/值對(duì)：

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

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

---

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

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

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

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

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

另一個(gè) mapper 則收到后兩行：

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

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

---

5. 自定義InputFormat

概述

自定義一個(gè)類繼承FileInputFormat。

改寫RecordReader，實(shí)現(xiàn)一次讀取一個(gè)完整文件封裝為KV。

在輸出時(shí)使用SequenceFileOutPutFormat輸出合并文件。

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。