這樣的操作在map端或者reduce端均可。下面以一個實際業(yè)務場景中的例子來簡要說明。

問題簡要描述：

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

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

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

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

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

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

(1)文件流沖突。

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

(2)mapreduce推測執(zhí)行機制

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

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

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

2.文件異常處理

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

3.多目錄以及文件拼接

如果我們將key的種類設為1000萬，上述方法會生成太多的小文件從而影響hdfs的性能，另外，因為所有文件都在同一個目錄下，會導致同一個目錄下文件數(shù)目過多而影響訪問效率。

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

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

最新評論