mapreduce中,一個job的map個數(shù), 每個map處理的數(shù)據(jù)量是如何決定的呢? 另外每個map又是如何讀取輸入文件的內(nèi)容呢? 用戶是否可以自己決定輸入方式, 決定map個數(shù)呢? 這篇文章將詳細(xì)講述hadoop中各種InputFormat的功能和如何編寫自定義的InputFormat.
簡介: mapreduce作業(yè)會根據(jù)輸入目錄產(chǎn)生多個map任務(wù), 通過多個map任務(wù)并行執(zhí)行來提高作業(yè)運行速度, 但如果map數(shù)量過少, 并行量低, 作業(yè)執(zhí)行慢, 如果map數(shù)過多, 資源有限, 也會增加調(diào)度開銷. 因此, 根據(jù)輸入產(chǎn)生合理的map數(shù), 為每個map分配合適的數(shù)據(jù)量, 能有效的提升資源利用率, 并使作業(yè)運行速度加快.
在mapreduce中, 每個作業(yè)都會通過 InputFormat來決定map數(shù)量. InputFormat是一個接口, 提供兩個方法:
InputSplit[] getSplits(JobConf job, int numSplits) throws IOException;
RecordReader<K, V> getRecordReader(InputSplit split,
JobConf job,
Reporter reporter) throws IOException;
其中getSplits方法會根據(jù)輸入目錄產(chǎn)生InputSplit數(shù)組, 每個InputSplit會相應(yīng)產(chǎn)生一個map任務(wù), map的輸入定義在InputSplit中. getRecordReader方法返回一個RecordReader對象, RecordReader決定了map任務(wù)如何讀取輸入數(shù)據(jù), 例如一行一行的讀取還是一個字節(jié)一個字節(jié)的讀取, 等等.
下圖是InputFormat的實現(xiàn)類:
(暫時無法上傳)
這理詳細(xì)介紹FileInputFormat和CombineFileInputFormat, 其它不常用,有興趣的可以自己查看hadoop源碼.
FileInputFormat(舊接口org.apache.hadoop.mapred)
mapreduce默認(rèn)使用TextInputFormat，TextInputFormat沒有實現(xiàn)自己的getSplits方法,它繼承于FileInputFormat, 因此使用了FileInputFormat的.
org.apache.hadoop.mapred.FileInputFormat的getSplits流程:
兩個配置
mapred.min.split.size (一個map最小輸入長度),
mapred.map.tasks (推薦map數(shù)量)
如何決定每個map輸入長度呢? 首先獲取輸入目錄下所有文件的長度和, 除以mapred.map.tasks得到一個推薦長度goalSize, 然后通過式子: Math.max(minSize, Math.min(goalSize, blockSize))決定map輸入長度. 這里的minSize為mapred.min.split.size, blockSize為相應(yīng)文件的block長度. 這式子能保證一個map的輸入至少大于mapred.min.split.size, 對于推薦的map長度,只有它的長度小于blockSize且大于mapred.min.split.size才會有效果. 由于mapred.min.split.size默認(rèn)長度為1, 因此通常情況下只要小于blockSize就有效果,否則使用blockSize做為map輸入長度.
因此, 如果想增加map數(shù), 可以把mapred.min.split.size調(diào)小(其實默認(rèn)值即可), 另外還需要把mapred.map.tasks設(shè)置大.
如果需要減少map數(shù),可以把mapred.min.split.size調(diào)大, 另外把mapred.map.tasks調(diào)小.
這里要特別指出的是FileInputFormat會讓每個輸入文件至少產(chǎn)生一個map任務(wù), 因此如果你的輸入目錄下有許多文件, 而每個文件都很小, 例如幾十kb, 那么每個文件都產(chǎn)生一個map會增加調(diào)度開銷. 作業(yè)變慢.
那么如何防止這種問題呢? CombineFileInputFormat能有效的減少map數(shù)量.
FileInputFormat(新接口org.apache.hadoop.mapreduce.lib.input)
Hadoop 0.20開始定義了一套新的mapreduce編程接口, 使用新的FileInputFormat, 它與舊接口下的FileInputFormat主要區(qū)別在于, 它不再使用mapred.map.tasks, 而使用mapred.max.split.size參數(shù)代替goalSize, 通過Math.max(minSize, Math.min(maxSize, blockSize))決定map輸入長度, 一個map的輸入要大于minSize,小于
Math.min(maxSize, blockSize).
若需增加map數(shù),可以把mapred.min.split.size調(diào)小,把mapred.max.split.size調(diào)大. 若需減少map數(shù), 可以把mapred.min.split.size調(diào)大, 并把mapred.max.split.size調(diào)小.
CombineFileInputFormat
顧名思義, CombineFileInputFormat的作用是把許多文件合并作為一個map的輸入.
在它之前,可以使用MultiFileInputFormat,不過其功能太簡單, 它以文件為單位，一個文件至多分給一個map處理, 如果某個目錄下有許多小文件, 另外還有一個超大文件, 處理大文件的map會嚴(yán)重偏慢.
CombineFileInputFormat是一個被推薦使用的InputFormat. 它有三個配置:
mapred.min.split.size.per.node， 一個節(jié)點上split的至少的大小
mapred.min.split.size.per.rack 一個交換機下split至少的大小
mapred.max.split.size 一個split最大的大小
它的主要思路是把輸入目錄下的大文件分成多個map的輸入, 并合并小文件, 做為一個map的輸入. 具體的原理是下述三步:
1.根據(jù)輸入目錄下的每個文件,如果其長度超過mapred.max.split.size,以block為單位分成多個split(一個split是一個map的輸入),每個split的長度都大于mapred.max.split.size, 因為以block為單位, 因此也會大于blockSize, 此文件剩下的長度如果大于mapred.min.split.size.per.node, 則生成一個split, 否則先暫時保留.
2. 現(xiàn)在剩下的都是一些長度效短的碎片,把每個rack下碎片合并, 只要長度超過mapred.max.split.size就合并成一個split, 最后如果剩下的碎片比mapred.min.split.size.per.rack大, 就合并成一個split, 否則暫時保留.
3. 把不同rack下的碎片合并, 只要長度超過mapred.max.split.size就合并成一個split, 剩下的碎片無論長度, 合并成一個split.
舉例: mapred.max.split.size=1000
mapred.min.split.size.per.node=300
mapred.min.split.size.per.rack=100
輸入目錄下五個文件,rack1下三個文件,長度為2050,1499,10, rack2下兩個文件,長度為1010,80. 另外blockSize為500.
經(jīng)過第一步, 生成五個split: 1000,1000,1000,499,1000. 剩下的碎片為rack1下:50,10; rack2下10:80
由于兩個rack下的碎片和都不超過100, 所以經(jīng)過第二步, split和碎片都沒有變化.
第三步,合并四個碎片成一個split, 長度為150.
如果要減少map數(shù)量, 可以調(diào)大mapred.max.split.size, 否則調(diào)小即可.
其特點是: 一個塊至多作為一個map的輸入，一個文件可能有多個塊，一個文件可能因為塊多分給做為不同map的輸入， 一個map可能處理多個塊，可能處理多個文件。