這樣的操作在map端或者reduce端均可。下面以一個(gè)實(shí)際業(yè)務(wù)場(chǎng)景中的例子來簡(jiǎn)要說明。

問題簡(jiǎn)要描述：

假如reduce輸入的key是Text(String),value是BytesWritable(byte[]),不同key的種類為100萬個(gè)，value的大小平均為30k左右，每個(gè)key大概對(duì)應(yīng) 100個(gè)value,要求對(duì)每一個(gè)key建立兩個(gè)文件，一個(gè)用來不斷添加value中的二進(jìn)制數(shù)據(jù)，一個(gè)用來記錄各個(gè)value在文件中的位置索引。(大量的小文件會(huì)影響HDFS的性能，所以最好對(duì)這些小文件進(jìn)行拼接）

當(dāng)文件數(shù)量較小時(shí)，可以考慮使用MultipleOutput來進(jìn)行key-value的分流，可以按照key的不同，將其輸出到不同的文件或者目錄中。但是reduce的數(shù)量只能為1,不然每個(gè)reduce都會(huì)生成相同的目錄或者文件，不能達(dá)到最終的目的。此外最重要的是，操作系統(tǒng)對(duì)每個(gè)進(jìn)程打開的文件數(shù)量的限制，默認(rèn)為1024，集群的各個(gè)datanode可能會(huì)配置更高的值，但最多在幾萬左右，仍然是一個(gè)限制因素。不能滿足百萬文件的需求。

reduce的主要目的是用來歸并key-value并輸出到HDFS上，我們當(dāng)然也可以在reduce中進(jìn)行其他的操作，比如文件讀寫。因?yàn)槟J(rèn)的partitioner保證同一個(gè)key的數(shù)據(jù)肯定會(huì)在同一個(gè)reduce中，所以在每個(gè)reduce中只用打開兩個(gè)文件進(jìn)行讀寫即可(一個(gè)索引文件，一個(gè)數(shù)據(jù)文件)。并發(fā)度由reduce數(shù)量決定，將reduce數(shù)量設(shè)為256，那我們就可以同時(shí)處理256個(gè)key的數(shù)據(jù)(partioner保證了不同reduce處理的key不同，不會(huì)引起文件讀寫沖突)。這樣的并發(fā)度的效率是很客觀的，可以在較短的時(shí)間內(nèi)完成需求。

思路是這樣，但同時(shí)由于hdfs的特性以及hadoop的任務(wù)調(diào)度，在文件讀寫過程中，仍有可能會(huì)出現(xiàn)很多問題，下面簡(jiǎn)要說些一些常見的會(huì)碰到的問題。

1.org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException異常

這可能是最經(jīng)常碰到的一個(gè)問題。可能的原因如下：

(1)文件流沖突。

一般創(chuàng)建文件時(shí)都會(huì)打開一個(gè)供寫入的文件流。而我們希望是追加，所以如果使用了錯(cuò)誤的API ，就有可能引起上述問題。以FileSystem類為例，如果使用create()方法之后再調(diào)用append()方法，就會(huì)拋出上述異常。所以最好使用createNewFile方法，只創(chuàng)建文件，不打開流。

(2)mapreduce推測(cè)執(zhí)行機(jī)制

mapreduce 為了提高效率，會(huì)在一個(gè)任務(wù)啟動(dòng)之后，同時(shí)啟動(dòng)一些相同的任務(wù)(attempt)，其中有一個(gè)attempt成功完成之后，視為整個(gè)task完成，其結(jié)果 作為最終結(jié)果，并且殺掉那些較慢的attempt。集群一般會(huì)開啟此選項(xiàng)以優(yōu)化性能(以空間換時(shí)間)。但在本問題環(huán)境下推測(cè)執(zhí)行卻不太合適。因?yàn)槲覀円话阆Ｍ粋€(gè)task 用來處理一個(gè)文件，但如果啟動(dòng)推測(cè)執(zhí)行，會(huì)有幾個(gè)attempt同時(shí)試圖操作同一個(gè)文件，就會(huì)引發(fā)異常。所以最好關(guān)掉此選項(xiàng)，將 mapred.reduce.max.attempts 設(shè)為1,或者將mapred.reduce.tasks.speculative.execution設(shè)為false.

但此時(shí)仍有可能會(huì)出現(xiàn)問題。因?yàn)槿绻粋€(gè)task的唯一attempt出現(xiàn)問題，在被kill掉之后，task仍會(huì)另起一個(gè)attempt，此時(shí)因?yàn)榍耙粋€(gè)attempt異常終止，仍有可能會(huì)影響到新起的attempt的文件操作，引發(fā)異常。所以最安全的方法是，借鑒推測(cè)執(zhí)行的機(jī)制(每個(gè)attempt各自生成自己的結(jié)果，最終選擇一個(gè)作為最終結(jié)果)，以每個(gè)attempt的id號(hào)為后綴附加到所操作的文件上，同時(shí)捕獲所有文件操作的異常并處理，這樣可避免文件的讀寫沖突。Context可以用來獲取運(yùn)行時(shí)的一些上下文信息，可以很容易得到attempt的id號(hào)。注意，此時(shí)如果開啟推測(cè)執(zhí)行也可以，但是會(huì)生成很多相同的文件(每個(gè)attempt一份),仍然不是最好的解決方法。

同時(shí)，我們可以利用reduce的輸出來記錄運(yùn)行“不正常的” key.這些task大多數(shù)是attempt_0被殺掉而重啟了一個(gè)attempt_1，所以下面的文件一般為兩份。可以對(duì)這些情況的key輸出(文件異常或者attemptID > 0)，并進(jìn)行一些后續(xù)處理，比如文件重命名，或者緊對(duì)這些key重新寫入。因?yàn)榇朔N情況的key一般只占極少數(shù)，所以并不影響總體的效率。

2.文件異常處理

最好能將mapreduce中的所有文件操作都設(shè)置好異常處理。不然一個(gè)文件異常就有可能會(huì)使整個(gè)job失敗。所以從效率來講，最好是在文件發(fā)生異常時(shí)將其key作為reduce的輸出以進(jìn)行記錄。因?yàn)橥瑫r(shí)mapreduce會(huì)重啟一個(gè)task attempts重新進(jìn)行文件讀寫，可保證我們得到最終的數(shù)據(jù)，最后所需的只是對(duì)那些異常的key進(jìn)行一些簡(jiǎn)單的文件重命名操作即可。

3.多目錄以及文件拼接

如果我們將key的種類設(shè)為1000萬，上述方法會(huì)生成太多的小文件從而影響hdfs的性能，另外，因?yàn)樗形募荚谕粋€(gè)目錄下，會(huì)導(dǎo)致同一個(gè)目錄下文件數(shù)目過多而影響訪問效率。

在創(chuàng)建文件的同時(shí)建立多個(gè)子目錄，一個(gè)有用的方法是以reduce的taskid來建立子目錄。這樣有多少個(gè)reduce就可以建立多少個(gè)子目錄，不會(huì)有文件沖突。同一個(gè)reduce處理的key都會(huì)在同一個(gè)目錄下。

文件拼接要考慮的一個(gè)索引的問題。為了將文件索引建立的盡量簡(jiǎn)單，應(yīng)該盡量保證同一個(gè)key的所有數(shù)據(jù)都在同一個(gè)大文件中。這可以利用key的hashCode來實(shí)現(xiàn)。如果我們想在每個(gè)目錄下建立1000個(gè)文件，只需將hashCode對(duì)1000取余即可。