深刻淺出Hadoop數據流（1）

1. 首先是一些術語的說明。MapReduce做業(job)是客戶端執行的單位：它包括輸入數據、MapReduce程序和配置信息。Hadoop經過把做業分紅若干個小任務(task)來工做，其包括兩種類型的任務：map任務和reduce任務。

2. 有兩種類型的節點控制着做業執行過程：jobtracker和多個tasktracker。jobtracker經過調度任務在tasktracker上運行，來協調全部運行在系統上的做業。Tasktracker運行任務的同時，把進度報告傳送到jobtracker，jobtracker則記錄着每項任務的總體進展狀況。若是其中一個任務失敗，jobtracker能夠從新調度任務到另一個tasktracker。Hadoop把輸入數據劃分紅等長的小數據發送到MapReduce，稱爲輸入分片(input split)或分片。Hadoop爲每一個分片(split)建立一個map任務，由它來運行用戶自定義的map函數來分析每一個分片中的記錄。

3. 擁有許多分片就意味着處理每一個分片的時間與處理整個輸入的時間相比是比較小的。所以，若是咱們並行處理每一個分片，且分片是小塊的數據，那麼處理過程將有一個更好的負載平衡，由於更快的計算機將可以比一臺速度較慢的機器在做業過程當中處理完比例更多的數據分片。即便是相同的機器，沒有處理的或其餘同時運行的做業也會使負載平衡得以實現，而且在分片變得更細時，負載平衡質量也會更佳。

4. 另外一方面，若是分片過小，那麼管理分片的總時間和map任務建立的總時間將決定做業的執行的總時間。對於大多數做業，一個理想的分片大小每每是一個HDFS塊的大小，默認是64 MB，雖然這能夠根據集羣進行調整(對於全部新建文件)或在新建每一個文件時具體進行指定。

5. map任務的執行節點和輸入數據的存儲節點是同一節點，Hadoop的性能達到最佳。這就是所謂的data locality optimization(數據局部性優化)。如今咱們應該清楚爲何最佳分片的大小與塊大小相同：它是最大的可保證存儲在單個節點上的數據量。若是分區跨越兩個塊，那麼對於任何一個HDFS節點而言，基本不可能同時存儲這兩數據塊，所以此分佈的某部分必須經過網絡傳輸到節點，這與使用本地數據運行map任務相比，顯然效率更低。

6. map任務把輸出寫入本地硬盤，而不是HDFS。這是爲何？由於map的輸出做爲中間輸出：而中間輸出則被reduce任務處理後產生最終的輸出，一旦做業完成，map的輸出就能夠刪除了。所以，把它及其副本存儲在HDFS中，不免有些小題大作。若是該節點上運行的map任務在map輸出給reduce任務處理以前崩潰，那麼Hadoop將在另外一個節點上從新運行map任務以再次建立map的輸出。

7. reduce任務並不具有數據本地讀取的優點-- 一個單一的reduce任務的輸入每每來自於全部mapper的輸出。在本例中，咱們有一個單獨的reduce任務，其輸入是由全部map任務的輸出組成的。所以，有序map的輸出必須經過網絡傳輸到reduce任務運行的節點，並在那裏進行合併，而後傳遞到用戶定義的reduce函數中。爲增長其可靠性，reduce的輸出一般存儲在HDFS中。如第3章所述，對於每一個reduce輸出的HDFS塊，第一個副本存儲在本地節點上，其餘副本存儲在其餘機架節點中。所以，編寫reduce的輸出確實十分佔用網絡帶寬，可是隻和正常的HDFS寫管線的消耗同樣。

8. 一個單一的reduce 任務的整個數據流如圖2-2所示。虛線框表示節點，虛線箭頭表示數據傳輸到一個節點上，而實線的箭頭表示節點之間的數據傳輸。

9. reduce任務的數目並非由輸入的大小來決定，而是單獨具體指定的。在第7章的7.1節中，將介紹如何爲一個給定的做業選擇reduce任務數量。

10. 若是有多個reducer，map任務會對其輸出進行分區，爲每一個reduce任務建立一個分區(partition)。每一個分區包含許多鍵(及其關聯的值)，但每一個鍵的記錄都在同一個分區中。分區能夠經過用戶定義的partitioner來控制，但一般是用默認的分區工具，它使用的是hash函數來造成"木桶"鍵/值，這種方法效率很高。

11. 通常狀況下，多個reduce任務的數據流如圖2-3所示。此圖清楚地代表了map和reduce任務之間的數據流爲何要稱爲"shuffle"(洗牌)，由於每一個reduce任務的輸入都由許多map任務來提供。shuffle其實比此圖所顯示的更復雜，而且調整它可能會對做業的執行時間產生很大的影響，詳見6.4節。

（點擊查看大圖）圖2-2：MapReduce中單一reduce任務的數據流圖

1.  

（點擊查看大圖）圖2-3：多個reduce任務的MapReduce數據流

2.4.1  數據流（2）

最後，也有可能不存在reduce任務，不須要shuffle的時候，這樣的狀況是可能的，由於處理能夠並行進行(第7章有幾個例子討論了這個問題)。在這種狀況下，惟一的非本地節點數據傳輸是當map任務寫入到HDFS中(見圖2-4)。

圖2-4：MapReduce中沒有reduce任務的數據流

集羣的可用帶寬限制了MapReduce做業的數量，所以map和reduce任務之間數據傳輸的代價是最小的。Hadoop容許用戶聲明一個combiner，運行在map的輸出上-- 該函數的輸出做爲reduce函數的輸入。因爲combiner是一個優化方法，因此Hadoop不保證對於某個map的輸出記錄是否調用該方法，調用該方法多少次。換言之，不調用該方法或者調用該方法屢次，reducer的輸出結果都同樣。

combiner的規則限制着可用的函數類型。咱們將用一個例子來巧妙地加以說明。之前面的最高氣溫例子爲例，1950年的讀數由兩個map處理(由於它們在不一樣的分片中)。假設第一個map的輸出以下：

1.  (1950, 0)  

2.  (1950, 20)  

3.  (1950, 10) 

第二個map的輸出以下：

1.  (1950, 25)  

2.  (1950, 15) 

reduce函數再調用時被傳入如下數字：

1.  (1950, [0, 20, 10, 25, 15]) 

由於25是輸入值中的最大數，因此輸出以下：

1.  (1950, 25) 

咱們能夠用combiner，像reduce函數那樣，爲每一個map輸出找到最高氣溫。reduce函數被調用時將被傳入以下數值：

1.  (1950, [20, 25]) 

然而，reduce輸出的結果和之前同樣。更簡單地說，咱們能夠像下面這樣，對本例中的氣溫值進行以下函數調用：

1.  max(0, 20, 10, 25, 15) = max(max(0, 20, 10), 
max(25, 15)) = max(20, 25) = 25 

並不是全部函數都有此屬性。例如，若是咱們計算平均氣溫，便不能用mean做爲combiner，由於：

1.  mean(0, 20, 10, 25, 15) = 14 

可是：

1.  mean(mean(0, 20, 10), mean(25, 15)) = mean(10, 20) = 15 

combiner並不能取代reduce函數。(爲何呢？reduce函數仍然須要處理來自不一樣的map給出的相同鍵記錄。)但它能夠幫助減小map和reduce之間的數據傳輸量，而正由於此，是否在MapReduce做業中使用combiner是須要慎重考慮的。