mapreduce中,一個(gè)job的map個(gè)數(shù), 每個(gè)map處理的數(shù)據(jù)量是如何決定的呢? 另外每個(gè)map又是如何讀取輸入文件的內(nèi)容呢? 用戶是否可以自己決定輸入方式, 決定map個(gè)數(shù)呢? 這篇文章將詳細(xì)講述hadoop中各種InputFormat的功能和如何編寫自定義的InputFormat.

 

簡(jiǎn)介: mapreduce作業(yè)會(huì)根據(jù)輸入目錄產(chǎn)生多個(gè)map任務(wù), 通過多個(gè)map任務(wù)并行執(zhí)行來提高作業(yè)運(yùn)行速度, 但如果map數(shù)量過少, 并行量低, 作業(yè)執(zhí)行慢, 如果map數(shù)過多, 資源有限, 也會(huì)增加調(diào)度開銷. 因此, 根據(jù)輸入產(chǎn)生合理的map數(shù), 為每個(gè)map分配合適的數(shù)據(jù)量, 能有效的提升資源利用率, 并使作業(yè)運(yùn)行速度加快.

    在mapreduce中, 每個(gè)作業(yè)都會(huì)通過 InputFormat來決定map數(shù)量. InputFormat是一個(gè)接口, 提供兩個(gè)方法:

InputSplit[] getSplits(JobConf job, int numSplits) throws IOException;

RecordReader<K, V> getRecordReader(InputSplit split,

                                     JobConf job,

                                     Reporter reporter) throws IOException;

    其中getSplits方法會(huì)根據(jù)輸入目錄產(chǎn)生InputSplit數(shù)組, 每個(gè)InputSplit會(huì)相應(yīng)產(chǎn)生一個(gè)map任務(wù), map的輸入定義在InputSplit中. getRecordReader方法返回一個(gè)RecordReader對(duì)象, RecordReader決定了map任務(wù)如何讀取輸入數(shù)據(jù), 例如一行一行的讀取還是一個(gè)字節(jié)一個(gè)字節(jié)的讀取, 等等.

    下圖是InputFormat的實(shí)現(xiàn)類:

       (暫時(shí)無法上傳)

    這理詳細(xì)介紹FileInputFormat和CombineFileInputFormat, 其它不常用,有興趣的可以自己查看hadoop源碼.

 

 

mapreduce默認(rèn)使用TextInputFormat，TextInputFormat沒有實(shí)現(xiàn)自己的getSplits方法,它繼承于FileInputFormat, 因此使用了FileInputFormat的.

org.apache.hadoop.mapred.FileInputFormat的getSplits流程:

兩個(gè)配置

mapred.min.split.size        (一個(gè)map最小輸入長(zhǎng)度),

mapred.map.tasks                (推薦map數(shù)量)

如何決定每個(gè)map輸入長(zhǎng)度呢? 首先獲取輸入目錄下所有文件的長(zhǎng)度和, 除以mapred.map.tasks得到一個(gè)推薦長(zhǎng)度goalSize, 然后通過式子: Math.max(minSize, Math.min(goalSize, blockSize))決定map輸入長(zhǎng)度. 這里的minSize為mapred.min.split.size, blockSize為相應(yīng)文件的block長(zhǎng)度. 這式子能保證一個(gè)map的輸入至少大于mapred.min.split.size, 對(duì)于推薦的map長(zhǎng)度,只有它的長(zhǎng)度小于blockSize且大于mapred.min.split.size才會(huì)有效果. 由于mapred.min.split.size默認(rèn)長(zhǎng)度為1, 因此通常情況下只要小于blockSize就有效果,否則使用blockSize做為map輸入長(zhǎng)度.

因此, 如果想增加map數(shù), 可以把mapred.min.split.size調(diào)小(其實(shí)默認(rèn)值即可), 另外還需要把mapred.map.tasks設(shè)置大.

如果需要減少map數(shù),可以把mapred.min.split.size調(diào)大, 另外把mapred.map.tasks調(diào)小.

這里要特別指出的是FileInputFormat會(huì)讓每個(gè)輸入文件至少產(chǎn)生一個(gè)map任務(wù), 因此如果你的輸入目錄下有許多文件, 而每個(gè)文件都很小, 例如幾十kb, 那么每個(gè)文件都產(chǎn)生一個(gè)map會(huì)增加調(diào)度開銷. 作業(yè)變慢.

那么如何防止這種問題呢? CombineFileInputFormat能有效的減少map數(shù)量.

CombineFileInputFormat

顧名思義, CombineFileInputFormat的作用是把許多文件合并作為一個(gè)map的輸入.

在它之前,可以使用MultiFileInputFormat,不過其功能太簡(jiǎn)單, 它以文件為單位，一個(gè)文件至多分給一個(gè)map處理, 如果某個(gè)目錄下有許多小文件, 另外還有一個(gè)超大文件, 處理大文件的map會(huì)嚴(yán)重偏慢.

CombineFileInputFormat是一個(gè)被推薦使用的InputFormat. 它有三個(gè)配置:

mapred.min.split.size.per.node， 一個(gè)節(jié)點(diǎn)上split的至少的大小

mapred.min.split.size.per.rack   一個(gè)交換機(jī)下split至少的大小

mapred.max.split.size             一個(gè)split最大的大小

它的主要思路是把輸入目錄下的大文件分成多個(gè)map的輸入, 并合并小文件, 做為一個(gè)map的輸入. 具體的原理是下述三步:

1.根據(jù)輸入目錄下的每個(gè)文件,如果其長(zhǎng)度超過mapred.max.split.size,以block為單位分成多個(gè)split(一個(gè)split是一個(gè)map的輸入),每個(gè)split的長(zhǎng)度都大于mapred.max.split.size, 因?yàn)橐?/span>block為單位, 因此也會(huì)大于blockSize, 此文件剩下的長(zhǎng)度如果大于mapred.min.split.size.per.node, 則生成一個(gè)split, 否則先暫時(shí)保留.

2. 現(xiàn)在剩下的都是一些長(zhǎng)度效短的碎片,把每個(gè)rack下碎片合并, 只要長(zhǎng)度超過mapred.max.split.size就合并成一個(gè)split, 最后如果剩下的碎片比mapred.min.split.size.per.rack大, 就合并成一個(gè)split, 否則暫時(shí)保留.

3. 把不同rack下的碎片合并, 只要長(zhǎng)度超過mapred.max.split.size就合并成一個(gè)split, 剩下的碎片無論長(zhǎng)度, 合并成一個(gè)split.

舉例: mapred.max.split.size=1000

     mapred.min.split.size.per.node=300

      mapred.min.split.size.per.rack=100

輸入目錄下五個(gè)文件,rack1下三個(gè)文件,長(zhǎng)度為2050,1499,10, rack2下兩個(gè)文件,長(zhǎng)度為1010,80. 另外blockSize為500.

經(jīng)過第一步, 生成五個(gè)split: 1000,1000,1000,499,1000. 剩下的碎片為rack1下:50,10; rack2下10:80

由于兩個(gè)rack下的碎片和都不超過100, 所以經(jīng)過第二步, split和碎片都沒有變化.

第三步,合并四個(gè)碎片成一個(gè)split, 長(zhǎng)度為150.

 

如果要減少map數(shù)量, 可以調(diào)大mapred.max.split.size, 否則調(diào)小即可.

其特點(diǎn)是: 一個(gè)塊至多作為一個(gè)map的輸入，一個(gè)文件可能有多個(gè)塊，一個(gè)文件可能因?yàn)閴K多分給做為不同map的輸入， 一個(gè)map可能處理多個(gè)塊，可能處理多個(gè)文件。

 

Yarn做為hadoop下一代集群資源管理和調(diào)度平臺(tái), 其上能支持多種計(jì)算框架, 本文就簡(jiǎn)要介紹一下這些計(jì)算框架.

1.       MapReduce

首先是大家熟悉的mapreduce, 在MR2之前, hadoop包括HDFS和mapreduce, 做為hadoop上唯一的分布式計(jì)算框架, 其優(yōu)點(diǎn)是用戶可以很方便的編寫分布式計(jì)算程序, 并支持許多的應(yīng)用, 如hive, mahout, pig等. 但是其缺點(diǎn)是無法充分利用集群資源, 不支持DAG, 迭代式計(jì)算等. 為了解決這些問題, yahoo提出了Yarn (next generation mapreduce), 一個(gè)分布式集群集群資源管理和調(diào)度平臺(tái). 這樣除了mapreduce外, 還可以支持各種計(jì)算框架.

2.       Spark

Spark是一種與mapreduce相似的開源計(jì)算框架, 不同之處在于Spark在某些工作負(fù)載方面表現(xiàn)更優(yōu), 因?yàn)樗褂昧藘?nèi)存分布式數(shù)據(jù)集, 另外除了提供交互式查詢外, 它還可以優(yōu)化迭代工作負(fù)載.

3.       Apache HAMA

Apache Hama 是一個(gè)運(yùn)行在HDFS上的BSP(Bulk Synchronous Parallel大容量同步并行) 計(jì)算框架, 主要針對(duì)大規(guī)模科學(xué)計(jì)算,如矩陣, 圖像, 網(wǎng)絡(luò)算法等.當(dāng)前它有一下功能:

• 作業(yè)提交和管理接口
• 單節(jié)點(diǎn)上運(yùn)行多個(gè)任務(wù)
• 輸入/輸出格式化
• 備份恢復(fù)
• 支持通過Apache Whirr運(yùn)行在云端
• 支持與Yarn一起運(yùn)行

4.       Apache Giraph

圖像處理平臺(tái)上運(yùn)行這大型算法(如page rank, shared connections, personalization-based popularity 等)已經(jīng)很流行, Giraph采用BSP模型（bulk-synchronous parallel model），可用于等迭代類算法。

5.       Open MPI

這是一個(gè)高性能計(jì)算函數(shù)庫(kù)，通常在HPC（High Performance Computing）中采用，與MapReduce相比，其性能更高，用戶可控性更強(qiáng)，但編程復(fù)雜，容錯(cuò)性差，可以說，各有所長(zhǎng)，在實(shí)際應(yīng)用中，針對(duì)不同 該應(yīng)用會(huì)采用MPI或者MapReduce。

6.       Apache HBase

HBase是一個(gè)hadoop數(shù)據(jù)庫(kù), 其特點(diǎn)是分布式，可擴(kuò)展的，存儲(chǔ)大數(shù)據(jù)。當(dāng)有需要隨機(jī),實(shí)時(shí)讀寫的大數(shù)據(jù)時(shí), 使用HBase很適合.

本文參考:

http://hbase.apache.org/