Hadoop中MapTask如何實(shí)現(xiàn)

小編給大家分享一下Hadoop中MapTask如何實(shí)現(xiàn)，希望大家閱讀完這篇文章之后都有所收獲，下面讓我們一起去探討吧！

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

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

         Spill過程 
         在collect階段的執(zhí)行過程中，當(dāng)內(nèi)存中的環(huán)形數(shù)據(jù)緩沖區(qū)中的數(shù)據(jù)達(dá)到一定發(fā)之后，便會觸發(fā)一次Spill操作，將部分?jǐn)?shù)據(jù)spill到本地磁盤上。SpillThread線程實(shí)際上是kvbuffer緩沖區(qū)的消費(fèi)者，主要代碼如下：

Java代碼  

1. spillLock.lock();  

2. while(true){  

3.    spillDone.sinnal();  

4.    while(kvstart == kvend){  

5.       spillReady.await();  

6.    }  

7.    spillDone.unlock();  

8.    //排序并將緩沖區(qū)kvbuffer中的數(shù)據(jù)spill到本地磁盤上  

9.    sortAndSpill();  

10.    spillLock.lock;  

11.    //重置各個指針，為下一下spill做準(zhǔn)備  

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

13.       bufvoid = kvbuffer.length;  

14.    }  

15.    vstart = vend;  

16.    bufstart = bufend;  

17. }  

18. spillLock.unlock();  

19.            

         sortAndSpill()方法中的內(nèi)部流程是這樣的： 
         第一步，使用用快速排序算法對kvbuffer[bufstart,bufend)中的數(shù)據(jù)排序，先對partition分區(qū)號排序，然后再按照key排序，經(jīng)過這兩輪排序后，數(shù)據(jù)就會以分區(qū)為單位聚集在一起，且同一分區(qū)內(nèi)的數(shù)據(jù)按key有序； 
         第二步，按分區(qū)大小由小到大依次將每個分區(qū)中的數(shù)據(jù)寫入任務(wù)的工作目錄下的臨時文件中，如果用戶設(shè)置了Combiner，則寫入文件之前，會對每個分區(qū)中的數(shù)據(jù)做一次聚集操作，比如<key1,val1>和<key1,val2>合并成<key1,<val1,val2>>； 
         第三步，將分區(qū)數(shù)據(jù)的元信息寫到內(nèi)存索引數(shù)據(jù)結(jié)構(gòu)SpillRecord中。分區(qū)的元數(shù)據(jù)信息包括臨時文件中的偏移量、壓縮前數(shù)據(jù)的大小和壓縮后數(shù)據(jù)的大小。 

         Combine過程 
         當(dāng)任務(wù)的所有數(shù)據(jù)都處理完后，MapTask會將該任務(wù)所有的臨時文件年合并成一個大文件，同時生成相應(yīng)的索引文件。在合并過程中，是以分區(qū)文單位進(jìn)行合并的。 
         讓每個Task最終生成一個文件，可以避免同時打開大量文件和對小文件產(chǎn)生隨機(jī)讀帶來的開銷。

看完了這篇文章，相信你對“Hadoop中MapTask如何實(shí)現(xiàn)”有了一定的了解，如果想了解更多相關(guān)知識，歡迎關(guān)注億速云行業(yè)資訊頻道，感謝各位的閱讀！