Hadoop和YARN :map+shuffle+reduce走讀

 　　今天作了一個hadoop分享，總結下來，包括mapreduce，及shuffle深度講解，還有YARN框架的詳細說明等。

　一 引言

　　一、海量日誌數據，提取出某日訪問百度次數最多的那個IP

　　算法思想：分而治之+Hash

　　1.IP地址最多有2^32=4G種取值狀況，因此不能徹底加載到內存中處理； 
　　2.能夠考慮採用「分而治之」的思想，按照IP地址的Hash(IP)%1024值，把海量IP日誌分別存儲到1024個小文件中。這樣，每一個小文件最多包含4MB　　個IP地址； 
　　3.對於每個小文件，能夠構建一個IP爲key，出現次數爲value的Hash map，同時記錄當前出現次數最多的那個IP地址；
　　4.能夠獲得1024個小文件中的出現次數最多的IP，再依據常規的排序算法獲得整體上出現次數最多的IP；

　　二、Top K：統計最熱門的10個查詢串

　　假設目前有一千萬個記錄，請你統計最熱門的10個查詢串，要求使用的內存不能超過1G。

　　第一步、維護一個Key爲Query字串，Value爲該Query出現次數的HashTable，每次讀取一個Query，若是該字串不在Table中，那麼加入該字串，而且將Value值設爲　　1；若是該字串在Table中，那麼將該字串的計數加一便可。採用Hash Table的緣由：查詢速度很是的快，幾乎是O(1)的時間複雜度。

　　第二步、堆頂存放的是整個堆中最小的數，如今遍歷N個數，把最早遍歷到的k個數存放到最小堆中，並假設它們就是咱們要找的最大的k個數，X1>X2...Xmin(堆頂)，　　然後遍歷後續的N-K個數，一一與堆頂元素進行比較，若是遍歷到的Xi大於堆頂元素Xmin，則把Xi放入堆中，然後更新整個堆，更新的時間複雜度爲logK，若是　　　　　　Xi<Xmin，則不更新堆

　　最終的時間複雜度是：O（N） + N'*O（logK）

 　　下面介紹另外一種處理大數據的編程模式：mapreduce

　二 Mapreduce

　　Hadoop 起源於 google’s 三篇papers:

　　1 mapreduce；

　　2 The google file system;

　　3 big table；

　2.1 Mapreduce機制 – from google paper

　Mapreduce 是一種處理大數據的編程模式。

　　1執行概述圖Execution Overview

 

 

　　2Master的數據結構Master Data Structures

　　對於每一個map和reduce任務，master中保存着三個狀態(idle, in-progress, or completed)和非空閒任務的機器id；master是map 到 reduce的管道；

　　3容錯機制Fault Tolerance

　　Master會週期性的ping每一個worker，若是超時沒有收到迴應，則認爲這個worker死了。

　　已完成和死掉的worker都會被master置成空閒狀態，爲了下一次調度

　　已經完成的map tasks不能從新執行，由於他們的輸出保存在本地硬盤裏；

　　已經完成的reduce task不須要從新執行，由於他們的輸出已經保存在了全局的HDFS裏面；

　　例如，Map task在worker A 執行失敗了，稍後會在worker B再次執行，這時全部的reduce都須要從新執行，覺得全部的reducer沒有準備好從worker A讀出改成從　　worker B讀入；

　　Map Reduce 是對大範圍的worker失敗有復原能力的，例如，有80臺機器持續是unreachable狀態，master爲了保證進度的向前推動，簡單的使unreachable機器　　再次執行任務；

　　4 存儲Locality

　　網絡帶寬是一個稀缺資源，因此咱們利用GFS把文件保存在本地磁盤以節省帶寬；

　　GFS把每一個文件劃分紅64M的block，而且有3個副本在不一樣的機器上；因此master把map任務分配到有輸入文件對應副本的機器上；若這個map任務失敗，會轉移到具備這個副本的同一網絡的其它機器上再次執行；

　　5 任務粒度Task Granularity

　　咱們把map階段劃分長M pieces，把 reduce階段劃分紅Rpieces，理想的M和R均比工做的機器數大不少。M和R應該多大好呢？master須要O(M+R)個調度決策和　　　　O(M*R)個狀態保存在內存；推薦M=200,000 and R=5,000, using 2,000 工做機器；

　　6備份任務Backup Tasks

　　形成總的執行時間過長的緣由主要是由某個任務的執行相對其它全部任務過慢，稱爲掉隊者；形成這個任務掉隊的可能緣由是硬盤讀寫慢，或者這臺機器在調度其它任　　　務，cpu，內存，硬盤或網絡帶寬資源被佔用；

　　解決辦法是當mapreduce任務將要執行完成時，master調度in-progress狀態掉隊任務的備份執行任務；咱們調整這種機制，使佔用的計算資源不高於必定的百分比而且實驗代表不採用這種機制，時間會增加44%。

　2.2 MapReduce 例子  –  from yahoo

　　「Hello World」 of MapReduce : the Word Count algorithm ：

 

 

　　流程介紹：

　　Multiple Input files -> Map (Discarded the offset and Split the text) -> Sort and Shuffle(sorts the key-value pairs by key and it 「shuffles」 all pairs with the same key to the same Reducer) -> Reduce (add up all counts)-> Output to R files

　　Java中Key-value pair格式：

　　Map’s input pattern  (offset,text) -> (1： ‘Hadoop uses’),(2: ‘ MapReduce‘)

　　Reduce’s input pattern (word,count) -> (word1:1),(word2,[1,1])

　　Coding：見ssh

　2.3 mapreduce 與機器學習

　　機器學習算法要是適用於hadoop必須重構算法，改寫成mapreduce的分佈式程序；有些機器學習算法很好改進，好比樸素貝葉斯就是自然的適用於mapreduce模式，由於先驗機率和條件機率能夠很好的分佈式統計，然而有些機器學習算法卻不容易改爲分佈式。

　　例子，k-means的分佈式程序：

　　代碼： 見ssh

　2.4 詳解shaffle過程

　　參考：http://langyu.iteye.com/blog/992916

 

 

 

         shuffle整體概述圖如上(來自官方文檔)；shuffle又叫洗牌或打亂，把ｍａｐ結果輸入到ｒｅｄｕｃｅ中，具體過程以下；左半部分細節圖以下(來自博客)。

 

 　  1 一個輸入文件會被劃分紅不少64M的block，split分佈式的輸入到各個map中；

　　2 partition的做用是map以後的鍵值對須要肯定分配給哪一個reduce，partition會把hash(key)%R的reduce序號追加到map處理後的鍵值對尾，保存到中間緩存中；

　　3 spill：sort和combiner組成，網絡帶寬是十分寶貴的資源，爲了節省帶寬，排序併合並緩存中的map生成的鍵值對，例如(a,1)(a,1)合併（a,2）;

　　4 merge：緩存快滿的時候，會寫入磁盤，上一步spill只是合併了緩存中的鍵值對，硬盤中有來自不一樣文件的鍵值對，須要再次合併相同key的value，也是爲了減小帶寬；例如 (a,2)(a,3)合併爲(a,[2],[3])，再調用combine方法,造成(a,5)，最後根據partition的hash地址發送到目標reduce中。

         右半部分細節圖:

 

         1 copy:經過HTTP方式請求map task所在的TaskTracker獲取map task的輸出文件，保存在緩存中(map後的結果存儲在磁盤，由TaskTracker管理);

         2 merge：跟map過程同樣，緩存內容溢寫到了磁盤，merge磁盤來自不一樣map的輸出，合併相同key的value，輸出到reduce中。

         3 input：把所有緩存和磁盤的文件造成最終文件，輸入到reduce中。

         備註：至此兩個步驟合起來纔是整個shuffle的完整過程。我認爲整個mapreduce過程應該是： map + Shuffle＋reduce才貼切。

　3 hadoop1 和 hadoop2架構對比

　　因爲MRv1 在擴展性、可靠性、資源利用率和多框架等方面存在明顯不足，Apache開始嘗試對MapReduce 進行升級改造，因而誕生了更加先進的下一代MapReduce 計算框架MRv2。因爲MRv2 將資源管理模塊構建成了一個獨立的通用系統YARN，這直接使得MRv2 的核心從計算框架MapReduce 轉移爲資源管理系統YARN。

　　3.1 編程模型對比

Map Reduce 1 架構

Map Reduce 2 架構

 

YARN 負責資源管理和調度，而ApplicationMaster 僅負責一個做業的管理；MRv1 僅是一個獨立的離線計算框架，而MRv2 則是運行於YARN 之上的MapReduce

      YARN上能夠支持運行出mapreduce外的多種計算框架，並可以統一管理和調度。

3.2 Hadoop1架構和Hadoop2架構對比

Hadoop1架構

Hadoop2架構

 

經過將原有JobTracker 中與應用程序相關和無關的模塊分開，不只減輕了JobTracker 負載，也使得Hadoop 支持更多的計算框架。

　　3.3  YARN基本組成結構

　　YARN 主要由ResourceManager、NodeManager、ApplicationMaster和Container 等幾個組件構成。

 

 

　　1 Resource Manager：主要由兩個組件構成：調度器（Scheduler）和（Applications Manager，ASM）；調度器：將系統中的資源分配給各個正在運行的應用程序；應用程序管理器：應用程序管理器負責管理整個系統中全部應用程序，包括應用程序提交、與調度器協商資源以啓動ApplicationMaster、監控ApplicationMaster 運行狀態並在失敗時從新啓動它等。

　　2 ApplicationMaster：

　　1）與RM 調度器協商以獲取資源（用 Container 表示）；

　　2）將獲得的任務進一步分配給內部的任務；

　　3）與 NM 通訊以啓動 / 中止任務；

　　4）監控全部任務運行狀態，並在任務運行失敗時從新爲任務申請資源以重啓任務。

　　3 NodeManager（NM）：NM 是每一個節點上的資源和任務管理器；

　　(NM與ApplicationMaster的區別是NM是控制整臺機器資源和任務管理，而AM是控制每一個應用的資源和任務管理)

　　4 Container：是YARN 中的資源抽象，它封裝了某個節點上的多維度資源，如內存、CPU、磁盤、網絡等。

　　總結：hadoop2統稱爲YARN，YARN負責資源管理和任務調度，MR2只是上面的一個計算框架，YARN還能夠支持Spark和Hbase等框架。