Hadoop分佈式計算架構流程分析-Hadoop商業環境實戰

版權聲明：本套技術專欄是做者（秦凱新）平時工做的總結和昇華，經過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和集羣環境容量規劃等內容，請持續關注本套博客。版權聲明：禁止轉載，歡迎學習。

1. Hadoop分佈式計算架構流程分析

• 1）在MapReduce程序讀取文件的輸入目錄上存放相應的文件。
• 2）客戶端程序在submit()方法執行前，獲取待處理的數據信息，而後根據集羣中參數的配置造成一個任務分配規劃。
• 3）客戶端提交job.split、jar包、job.xml等文件給yarn，yarn中的resourcemanager啓動MRAppMaster。
• 4）MRAppMaster啓動後根據本次job的描述信息，計算出須要的maptask實例數量，而後向集羣申請機器啓動相應數量的maptask進程。
• 5）maptask利用客戶指定的inputformat來讀取數據，造成輸入KV對。
• 6）maptask將輸入KV對傳遞給客戶定義的map()方法，作邏輯運算
• 7）map()運算完畢後將KV對序列化後寫到環形緩衝區，環形緩衝區默認是100M,寫滿80%後，會根據Hash（結合Reduce Task 的數量，不會超過Reduce Task的數量）溢寫，溢寫過程當中使緩存中的KV對按照K分區排序後不斷寫到磁盤文件從而造成不一樣的分區，若設置了本地聚合，則會合並Combiner多個溢寫文件爲一個。
• 9）MRAppMaster監控到全部maptask進程任務完成以後，會根據客戶指定的參數啓動相應數- 量的reducetask進程，並告知reducetask進程要處理的數據分區。
• 10）Reducetask進程啓動以後，根據MRAppMaster告知的待處理數據所在位置，從若干臺maptask運行所在機器上獲取到若干個maptask輸出結果文件，並在本地進行從新歸併排序，而後按照相同key的KV爲一個組，調用客戶定義的reduce()方法進行邏輯運算。
• 11）Reducetask運算完畢後，調用客戶指定的outputformat將結果數據輸出到外部存儲。

2. Hadoop分佈式計算流程圖

• 1）分佈式的運算程序每每須要分紅至少2個階段。
• 2）第一個階段的maptask併發實例，徹底並行運行，互不相干。
• 3）第二個階段的reduce task併發實例互不相干，可是他們的數據依賴於上一個階段的全部maptask併發實例的輸出。
• 4）MapReduce編程模型只能包含一個map階段和一個reduce階段，若是用戶的業務邏輯很是複雜，那就只能多個mapreduce程序，串行運行。 以下圖展現了這個MapReduce過程：

  

3 TDW計算引擎解析——Shuffle

3.1 MapReduce的Shuffle Collect 過程詳解（下圖Spill過程簡化了）

MapReduce中的Shuffle更像是洗牌的逆過程，把一組無規則的數據儘可能轉換成一組具備必定規則的數據。

爲何MapReduce計算模型須要Shuffle過程？咱們都知道MapReduce計算模型通常包括兩個重要的階段：Map是映射，負責數據的過濾分發；Reduce是規約，負責數據的計算歸併。Reduce的數據來源於Map，Map的輸出便是Reduce的輸入，Reduce須要經過Shuffle來獲取數據。

從Map輸出到Reduce輸入的整個過程能夠廣義地稱爲Shuffle。Shuffle橫跨Map端和Reduce端，在Map端包括Spill過程，在Reduce端包括copy和sort過程，如圖所示：

• Spill過程包括輸出、排序、溢寫、合併等步驟，如圖所示，期間會產生不少溢出文件，須要最終歸併排序

3.2 環形數據區Collect過程

• 每一個Map任務不斷地以<key, value>對的形式把數據輸出到在內存中構造的一個環形數據結構中。使用環形數據結構是爲了更有效地使用內存空間，在內存中放置儘量多的數據。

• 這個數據結構其實就是個字節數組，叫Kvbuffer，名如其義，可是這裏面不光放置了<key, value>數據，還放置了一些索引數據，給放置索引數據的區域起了一個Kvmeta的別名，在Kvbuffer的一塊區域上穿了一個IntBuffer（字節序採用的是平臺自身的字節序）的馬甲。<key, value>數據區域和索引數據區域在Kvbuffer中是相鄰不重疊的兩個區域，用一個分界點來劃分二者，分界點不是亙古不變的，而是每次Spill以後都會更新一次。初始的分界點是0，<key, value>數據的存儲方向是向上增加，索引數據的存儲方向是向下增加，

• Kvbuffer的存放指針bufindex是一直悶着頭地向上增加，好比bufindex初始值爲0，一個Int型的key寫完以後，bufindex增加爲4，一個Int型的value寫完以後，bufindex增加爲8。

• 索引是對<key, value>在kvbuffer中的索引，是個四元組，包括：value的起始位置、key的起始位置、partition值、value的長度，佔用四個Int長度，Kvmeta的存放指針Kvindex每次都是向下跳四個「格子」，而後再向上一個格子一個格子地填充四元組的數據。好比Kvindex初始位置是-4，當第一個<key, value>寫完以後，(Kvindex+0)的位置存放value的起始位置、(Kvindex+1)的位置存放key的起始位置、(Kvindex+2)的位置存放partition的值、(Kvindex+3)的位置存放value的長度，而後Kvindex跳到-8位置，等第二個<key, value>和索引寫完以後，Kvindex跳到-32位置。

• 關於Spill觸發的條件，也就是Kvbuffer用到什麼程度開始Spill，仍是要講究一下的。若是把Kvbuffer用得死死得，一點縫都不剩的時候再開始Spill，那Map任務就須要等Spill完成騰出空間以後才能繼續寫數據；若是Kvbuffer只是滿到必定程度，好比80%的時候就開始Spill，那在Spill的同時，Map任務還能繼續寫數據，若是Spill夠快，Map可能都不須要爲空閒空間而發愁。兩利相衡取其大，通常選擇後者。

3.3 Sort過程（hadoop環形緩衝區排序，最終改變索引）

先把Kvbuffer中的數據按照partition值和key兩個關鍵字升序排序，移動的只是索引數據，排序結果是Kvmeta中數據以partition爲單位彙集在一塊兒，同一partition內的數據按照key有序。

3.4 Spill 過程（Spill包括輸出、索引排序、溢寫、合併等步驟,Spill過程會產生不少溢出文件）

• Spill線程爲此次Spill過程建立一個磁盤文件：從全部的本地目錄中輪訓查找能存儲這麼大空間的目錄，找到以後在其中建立一個相似於「spill12.out」的文件。Spill線程根據排過序的Kvmeta挨個partition的把<key, value>數據吐到這個文件中，一個partition對應的數據吐完以後順序地吐下個partition，直到把全部的partition遍歷完。一個partition在文件中對應的數據也叫段(segment)。
• 每一次Spill過程就會最少生成一個out文件，有時還會生成index文件，Spill的次數也烙印在文件名中。索引文件和數據文件的對應關係以下圖所示

3.5 SortAndSpill如何一無既往地進行着數據輸出

• <key, value>只顧着悶頭按照bufindex指針向上增加，kvmeta只顧着按照Kvindex向下增加，Map取kvbuffer中剩餘空間的中間位置，用這個位置設置爲新的分界點，而後二者就能夠和諧地按照本身既定的軌跡放置數據了，當Spill完成，空間騰出以後，不須要作任何改動繼續前進。

3.6 Merge（屢次Spill後，out文件和Index文件會產生不少，須要歸併排序成一個segment文件）

• 在以前Spill過程當中的時候爲何不直接把這些索引信息存儲在內存中呢，何須又多了這步掃描的操做？特別是Spill的索引數據，以前當內存超限以後就把數據寫到磁盤，如今又要從磁盤把這些數據讀出來，仍是須要裝到更多的內存中。之因此畫蛇添足，是由於這時kvbuffer這個內存大戶已經再也不使用能夠回收，有內存空間來裝這些數據了。

• Merge過程主要是針對一個個partition對應的全部的segment進行合併，目標是合併成一個segment。當這個partition對應不少個segment時，會分批地進行合併：先從segment列表中把第一批取出來，以key爲關鍵字放置成最小堆，而後從最小堆中每次取出最小的<key, value>輸出到一個臨時文件中，這樣就把這一批段合併成一個臨時的段，把它加回到segment列表中；再從segment列表中把第二批取出來合併輸出到一個臨時segment，把其加入到列表中；這樣往復執行，直到剩下的段是一批，輸出到最終的文件中。

3.7 Copy 和 Merge Sort （shuffle reduce 過程，緩存區大小是64K,超事後輸出到磁盤）

• Reduce任務經過HTTP向各個Map任務拖取它所須要的數據。每一個節點都會啓動一個常駐的HTTP server，其中一項服務就是響應Reduce拖取Map數據。當有MapOutput的HTTP請求過來的時候，HTTP server就讀取相應的Map輸出文件中對應這個Reduce部分的數據經過網絡流輸出給Reduce。
• Reduce任務拖取某個Map對應的數據，若是在內存中能放得下此次數據的話就直接把數據寫到內存中。Reduce要向每一個Map去拖取數據，在內存中每一個Map對應一塊數據，當內存中存儲的Map數據佔用空間達到必定程度的時候，開始啓動內存中merge，把內存中的數據merge輸出到磁盤上一個文件中。
• 若是在內存中不能放得下這個Map的數據的話，直接把Map數據寫到磁盤上，在本地目錄建立一個文件，從HTTP流中讀取數據而後寫到磁盤，使用的緩存區大小是64K。拖一個Map數據過來就會建立一個文件，當文件數量達到必定閾值時，開始啓動磁盤文件merge，把這些文件合併輸出到一個文件。
• Map的輸出數據已是有序的，Merge進行一次合併排序，所謂Reduce端的sort過程就是這個合併的過程。通常Reduce是一邊copy一邊sort，即copy和sort兩個階段是重疊而不是徹底分開的。

4 總結

秦凱新 於深圳