Hadoop深入學習：MapTask詳解

我們主要來學習MapTask的內部實現。 
         
         整體執行流程 
 
         如上圖示，MapTask的整個處理流程分五個階段： 
         ●read階段：通過RecordReader從InputSplit分片中將數據解析成一個個key/value。 
         ●map階段：將由RecordReader解析出的key/value交給map()方法處理，并生成一個個新的key/value。 
         ●collect階段：將map()中新生成key/value由OutpCollector.collect()寫入內存中的環形數據緩沖區。 
         ●spill階段：當環形緩沖區達到一定閥值后，會將數據寫到本地磁盤上，生成一個spill文件。在寫文件之前，會先將數據進行一次本地排序，必要的時候(按配置要求)還會對數據進行壓縮。 
         ●combine階段：當所有數據處理完后，將所有的臨時的spill文件進行一次合并，最終之生成一個數據文件。 
     
         接下來我們會對該流程中最重要的collect、spill和combine三個階段進行更深入的學習。 
         Collect過程 
         前階段的map中新生成key/value對后，會調用OutpCollector.collect(key,value)，在該方法內部，先調用 Partitioner.getPartition()獲取該記錄的分區號，然后將<key,value,partition>傳給 MapOutputBuffer.collect()作進一步的處理。 
         MapOutputBuffer內部使用了一個內部的環形的緩沖區來暫時保存用戶的輸出數據，當緩沖區使用率達到一定閥值后，由SpillThread線 程將緩沖區中的數據spill到本地磁盤上，當所有的數據處理完畢后，對所有的文件進行合并，最終只生成一個文件。該數據緩沖區直接用想到MapTask 的寫效率。 
         環形緩沖區使得collect階段和spill階段可以并行處理。 
         MapOutputBuffer內部采用了兩級索引結構，涉及三個環形的內存緩沖區，分別是kvoffsets、kvindices和kvbuffer，這個環形緩沖區的大小可以通過io.sot.mb來設置，默認大小是100MB，圖示如下： 

         kvoffsets即 偏移量索引數組，用于保存key/value在kvindices中的偏移量。一個key/value對在kvoffsets數組中占一個int的大小， 而在kvindices數組中站3個int的大小(如上圖示，包括分區號partition,key的起始位置和value的起始位置)。 
         當kvoffsets的使用率超過io.sort.spill.percent(默認為80%)后，便會觸發SpillTread線程將數據spill到磁盤上。 
         kvindices即文職索引數組，用于保存實際的key/value在數據緩沖區kvbuffer中的起始位置。 
         kvbuffer即數據局緩沖區，用于實際保存key/value，默認情況下可使用io.sort.mb的95%，當該緩沖區使用率使用率超過io.sort.spill.percent后，便會觸發SpillTread線程將數據spill到磁盤上。 

         Spill過程 
         在collect階段的執行過程中，當內存中的環形數據緩沖區中的數據達到一定發之后，便會觸發一次Spill操作，將部分數據spill到本地磁盤上。SpillThread線程實際上是kvbuffer緩沖區的消費者，主要代碼如下：

spillLock.lock();  

while(true){  

   spillDone.sinnal();  

   while(kvstart == kvend){  

      spillReady.await();  

   }  

   spillDone.unlock();  

   //排序并將緩沖區kvbuffer中的數據spill到本地磁盤上  

   sortAndSpill();  

   spillLock.lock;  

   //重置各個指針，為下一下spill做準備  

   if(bufend < bufindex && bufindex < bufstart){  

      bufvoid = kvbuffer.length;  

   }  

   vstart = vend;  

   bufstart = bufend;  

}  

spillLock.unlock();