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