Hadoop–Task 相關

在MapReduce計算框架中，一個應用程序被劃分爲Map和Reduce兩個計算階段。他們分別由一個或多個Map Task 和Reduce Task組成。

1. Map Task： 處理輸入數據集合中的一片數據，並將產生的若干個數據片斷寫到本地磁盤。
1. 按照用戶提供的InputFormat將對應的InputSpilt解析成一系列的key/value， 並以此交給用戶編寫的map()函數處理。
2. 按照指定的Partitioner對數據分片，以肯定每一個key/value將交給哪一個Reducer Task處理。
3. 將數據交給用戶定義的Combiner進行以此本地規約(用戶沒有定義則直接跳過)
4. 將處理結果保存到本地磁盤。
2. Reduce Task: 從每一個Map Task上遠程拷貝相應的數據片斷，通過分組彙集和規約後，將結果寫到HDFS上做爲最終結果。
1. 經過HTTP請求從各個已經運行完成的Map Task上拷貝對應的數據分片。
2. 待數據拷貝完成，以key爲關鍵字對全部數據進行排序。經過排序，key相同的記錄彙集在一塊兒造成若干分組。
3. 將每組數據交給用戶編寫的reduce()函數處理。
4. 將結果直接寫到HSFS上面做爲最終輸出結果。

 

IFile

    IFile是一種支持行壓縮的存儲格式。爲了減小MapTask寫入磁盤的數據量和跨網絡傳輸的數據量，IFile支持按行壓縮數據記錄。當前Hadoop提供了ZLib(默認壓縮方式）、BZip2等壓縮算法。

    IFile文件格式：<key-len, value-len, key, value>

排序

     排序是MapReduce框架中最重要的從a組之一。Map Task和Reduce Task均會對數據(按照key)進行排序。該操做屬於Hadoop的默認行爲。任何應用程序均會被排序，而無論邏輯上是否須要。

     對於Map Task，它會將處理的結果暫時存放到一個緩衝區，當緩衝區使用率達到必定閾值後，在對緩衝區中的數據進行以此排序。並將這些有序集合以IFile文件的形式寫到磁盤上。而當數據處理完畢後，它會對磁盤上全部文件進行一次合併。已將這些文件造成一個大的有序文件。

     對於Reduce Task，它從每一個Map Task上面遠程拷貝相應的數據文件，若是文件大小超過必定閾值，則放到磁盤，不然放到內存。若是磁盤上文件數目達到必定閾值，則進行一次合併以生成一個更大文件；若是內存中文件大小或者數目超過必定閾值，則進行一次合併後寫到磁盤上。當全部數據拷貝完畢後，Reduce Task統一對內存和磁盤上的全部數據進行一次合併。

     Map Task和Reduce Task的緩衝區數據合併使用Hadoop本身實現的快排算法，而IFile文件合併則使用了基於堆實現的優先隊列。

快排

1. 樞軸選擇：使用序列的首尾和中間元素的中位數做爲樞軸
2. 子序列劃分：兩個索引i，j分別從左右兩端掃描，i掃描到大於等於樞軸的等值，j掃描到小於等於樞軸的元素中止，而後交換兩個元素。重複直到相遇
3. 相同元素的優化： 每次劃分子序列，將於樞軸相同的元素集中存放到中間位置，讓它們再也不參與後續的遞歸處理過程。即序列劃分三部分：小於樞軸、等於樞軸、大於樞軸
4. 減小遞歸次數：當子序列中元素數目小於13時，直接使用插入排序算法，再也不遞歸。

優先隊列

  文件歸併由類Merger完成。其採用多輪遞歸合併的方式。每輪選取最小的前io.sort.factor（默認是10，用戶可配置)個文件進行合併。並將產生的文件從新加入帶合併列表中。知道剩下的文件數目小於io.sort.factor個，此時，他會返回指向由這些文件組成的小頂堆的迭代器。

Reporter

   Reporter用來完成Task週期性的向TaskTracker彙報最新進度和計數器值。TaskReporter類實現了Reporter接口，並以線程形式啓動。其彙報的信息中包含兩部分：

1. 任務執行進度
1. Map Task 而言： 使用已讀取數據量佔數據總量的比例做爲任務當前進度值
2. Reduce Task： 其能夠分解爲三個階段： Shuffle、Sort、Reduce。每一個階段佔任務總進度的1/3.考慮在Shuttle階段，Reduce Task須要從M(M爲Map Task數目)個Map Task上讀取數據。所以，可被分解爲M個階段，每一個階段佔Shuffle進度的1/M。
2. 任務計數器值： 是由Hadoop提供的，用於實現跟蹤任務運行進度的全局技術功能。任務計數器由兩部分組成<name, value>.計數器以組爲單位進行管理，一個計數器屬於一個計數器組。Hadoop規定一個做業最多包含120個計數器(可經過參數mapreduce.job.counters.limit設定)，50個計數器組。