數分-理論-大資料5-MapReduce（分布式并行編程模型）

(資料分析系列)

1知識點

• 分布式并行編程
• 簡介
• 作業流程
• 執行階段
• shuffle程序詳解
• WordCount理解MapReduce

2具體內容

2.1分布式并行編程

• 分布式并行編程-》提高程式性能
• 集群：大量廉價服務器，并行執行大規模資料處理任務，獲取海量計算能力
• 分布式并行程式可以運行在由大量計算機構成的集群上，從而可以充分利用集群的并行處理能力，同時，通過向集群中增加新的計算節點，就可以很容易實作集群計算能力的擴充，
  

2.2簡介

MapReduce:計算向資料靠攏

• 資料需要大量的網路傳輸開銷
• 移動計算比移動資料更經濟
• 集群中的MapReduce框架就會將Map程式就近地在HDFS資料所在的節點運行，即將計算節點和存盤節點放在一起運行，從而減少了節點間的資料移動開銷
• Master/Slave架構，一個Master多個Slave，Master上運行JobTracker，Slave上運行TaskTracker，用戶提交每個計算作業，會被劃分成若干任務
  • JobTracker，作業和任務調度，監控執行，并重新調度已經失敗的任務
  • TaskTracker，執行 JobTracker指派的任務
• 批量計算的框架，包括，從資料輸入、資料處理（map、reduce、combiner等 ）、資料輸出
• 組件：
  • 客戶端：提交MapReduce作業
  • yarn資源管理器：集群上計算資源的協調
  • yarn節點管理器：啟動和監控集群中集群上的計算容器（container）
  • application master：協調運行MapReduce作業
  • HDFS：分布式檔案系統，與其他物體共享作業檔案

2.3 MapReduce函式

簡單WordCount實體

• 統計一個文本檔案中每個單詞出現的次數
• Map函式自己寫:
  • 輸入： <某一行文本在檔案中的偏移位置，該行文本的內容>
  • 輸出：檔案中的一行讀取后決議出每個單詞，輸出一批中間結果<單詞，出現次數>
• Reduce函式自己寫:
  • 中間結果作為Reduce函式的輸入
  • 用戶可以將相同單詞的出現次數進行累加，輸出每個單詞出現的總次數

2.4作業流程

• 分布式存盤、分布式計算

  • 分布式檔案系統HDFS實作分布式資料存盤
  • MapReduce實作分布式計算
  • 輸入和輸出都需要借助于分布式檔案系統進行存盤，這些檔案被分布存盤到集群中的多個節點上
  • “分而治之”：把一個大的資料集拆分成多個小資料塊在多臺機器上并行處理

• 作業流程：

• Map：

  • MapReduce作業拆成多個Map任務，在多臺機器上并行執行
  • 運行在資料存盤節點（計算存盤一起沒有資料傳輸開銷）
  • 結果：生成<k,v>中間結果

• Reduce：

  • 中間結果被分到多個Reduce任務
  • 在多臺機器上并行執行，具有相同key的<k,v>被發送到同一個Reduce里
  • Reduce對中間結果進行匯總計算得到最后結果，并輸出到分布式檔案系統
    

• 不同Map任務間不通信，不資訊交換

• 所有資訊通過MapReduce框架自身實作

• Map任務的輸入檔案、Reduce任務的處理結果都是保存在分布式檔案系統（HDFS）中的，而Map任務處理得到的中間結果則保存在本地存盤（如磁盤）中，

2.5執行階段

1. MapReduce框架使用InputFormat模塊做Map前的預處理，比如，驗證輸入的格式是否符合輸入定義；然后，將輸入檔案切分為邏輯上的多個InputSplit，InputSplit是MapReduce對檔案進行處理和運算的輸入單位，只是一個邏輯概念，每個InputSplit并沒有對檔案進行實際切割，只是記錄了要處理的資料的位置和長度，
2. 因為InputSplit是邏輯切分而非物理切分，所以，還需要通過RecordReader（RR）并根據InputSplit中的資訊來處理InputSplit中的具體記錄，加載資料并轉換為適合Map任務讀取的鍵值對，輸入給Map任務，
3. Map任務會根據用戶自定義的映射規則，輸出一系列的<key,value>作為中間結果，
4. 為了讓Reduce可以并行處理Map的結果，需要對Map的輸出進行一定的磁區、排序（Sort）、合并（Combine）和歸并（Merge）等操作，得到<key,value-list>形式的中間結果，再交給對應的Reduce程式進行處理，這個程序稱為Shuffle，
5. Reduce以一系列<key,value-list>中間結果作為輸入，執行用戶定義的邏輯，輸出結果給OutputFormat模塊，
6. OutputFormat模塊會驗證輸出目錄是否已經存在，以及輸出結果型別是否符合組態檔中的配置型別，如果都滿足，就輸出Reduce的結果到分布式檔案系統，
   

2.6shuffle程序詳解

Shuffle是MapReduce的核心

2.6.1程序

• Shuffle，是指標對Map輸出結果進行磁區、排序和合并等處理，并交給Reduce的程序，
• Shuffle程序分為Map端的操作和Reduce端的操作，
  

1.Map

• Map結果寫入快取
• 快取滿時，就啟動溢寫操作，把快取中的資料寫入磁盤檔案，并清空快取
• 溢寫操作，把快取中的資料進行磁區，然后對每個磁區的資料進行排序（Sort）和合并（Combine），之后再寫入磁盤檔案
• 每次溢寫操作生成一個新磁盤檔案，Map全部結束錢，溢寫檔案被歸并（Merge）成一個大的磁盤檔案，通知相應的Reduce來領取屬于自己需要處理的資料
  

2.Reduce

• 從Map端不同Map機器領回屬于自己需要處理的資料
• 對資料進行歸并（Merge）后交給Reduce處理

2.6.2Map

1.資料資料和執行Map任務

• 檔案在檔案系統中（GFS、HDFS），格式任意（檔案、二進制）
• 接受<k,v>輸入后，按映射規則轉換成一批<k,v>輸出，

2.寫入快取

• Map結果先寫入快取：
• 積累一定數量后一次性批量入磁盤，減少對磁盤I/O影響，一次尋址連續寫入降低開銷
• k，v被序列化成位元組陣列

3.溢寫（磁區、排序和合并）

• 默認快取容量有限，100M
• 溢寫操作（spill），把快取中的內容一次性寫入磁盤，清空快取
  • 溢寫，另一個單獨的后臺執行緒完成，不影響Map結果寫入快取
  • 為了保證Map不停持續寫入，設定溢寫比例，0.8：當100MB大小的快取被填滿80MB資料時，就啟動溢寫程序
• 磁區（partition），快取中資料<k,v>交給不同的Reduce并行處理，
  • 通過Partitioner介面對這些鍵值對進行磁區
  • 默認磁區：Hash對Key哈希，用Reduce任務數量取模，hash(key) mod R
  • R：Reduce任務的數量
  • 允許用戶通過多載Partitioner介面來自定義磁區方式
• 記憶體排序（sort），根據key對每個磁區內的所有鍵值對排序
• 合并（combine）：事先定義了Combiner函式，則這個時候會執行合并操作，從而減少需要溢寫到磁盤的資料量
  • 將具有相同key的<k，v>的v加起來：<“xmu”,1>和<“xmu”,1>，經過合并操作以后就可以得到一個鍵值對<“xmu”,2>
  • 減少了鍵值對的數量
  • Map端的這種合并操作，其實和Reduce的功能相似，稱為“合并”
  • Combiner非必須
  • Combiner的輸出是Reduce任務的輸入，Combiner絕不能改變Reduce任務最終的計算結果
  • 一般而言，累加、最大值等場景可以使用合并操作
• 經過磁區、排序、可能的合并，快取中的鍵值對被寫入磁盤，并清空快取，
• 每次溢寫會在磁盤生成一個新的溢寫檔案，寫入溢寫檔案中的鍵值對都經過磁區，排序

4.檔案歸并（Merge）

• 對具有相同key的鍵值對，被歸并成一個新的鍵值對
• 具有相同key的鍵值對：<k1,v1>、<k1,v2>…，會被歸并成一個新的鍵值對<k1,<V1,V2,…vn>>
• 已生成的溢寫檔案的數量超過引數min.num.spills.for.combine的值時（默認值是3，用戶可以修改這個值），那么，就可以再次運行Combiner，對資料進行合并操作，從而減少寫入磁盤的資料量，
• 溢寫檔案少時不允許combiner

2.6.3Reduce

Map端讀取結果，歸并，送給Reduce處理
1.“領取”資料(Fetch)

• 從Map機本地磁盤“領取”資料，存在自己機器磁盤
• 每個reduce任務通過RPC（Remote Procedure Call），向JobTracker詢問Map任務是否完成
• JobTracker監測到一個Map任務完成后，就會通知相關的Reduce任務來“領取”資料
• Reduce任務收到JobTracker通知，它就會到該Map任務所在機器上把屬于自己處理的磁區資料領取到本地磁盤中
• 系統中會存在多個Map機器，因此，Reduce任務會使用多個執行緒同時從多個Map機器領回資料，

2.歸并資料

• Shuffle階段，Reduce任務未真正開始，記憶體大部分分給shuffle作為快取
  • 系統存在多個Map機器，Reduce任務從多個Map機器領回自己的資料
  • 一般會合并（combine）鍵值對
• 溢寫啟動，相同key歸并，
  • 定義了Combiner，歸并后的資料還可執行合并，減少寫入磁盤的資料量
  • 多個溢寫檔案歸并成一個大檔案，并對鍵值對排序
  • 快取足夠時，不溢寫到磁盤，直接記憶體中歸并，輸出Reduce任務
  • 多個溢寫檔案歸并成一個大檔案，需多輪歸并操作，每輪歸并操作可以歸并的檔案數量是由引數io.sort.factor的值來控制的（默認值是10，可以修改），歸并一輪一個大檔案

3.把資料輸入Reduce任務

• 多輪歸并后得到的若干個大檔案，不會繼續歸并成一個新的大檔案，而是直接輸入給Reduce任務（減少磁盤讀寫開銷）
• Reduce任務會執行Reduce函式中定義的各種映射，輸出最終結果，并保存到分布式檔案系統中，
  

2.7以WordCount例子，理解MapReduce

MapReduce使用前提：待處理的資料集可以分解成許多小的資料集，而且每一個小資料集都可以完全并行地進行處理，

WordCount任務：

1. 判斷是否可以MapReduce實作：不同單詞間的頻數不存在相關性，彼此獨立

   • 不同單詞分發給不同機器并行處理
   • MapReduce實作詞頻統計任務

2. 設計思路：檔案內容決議成許多個單詞，把所有相同的單詞聚集到一起

   • 計算出每個單詞出現的次數

3. 執行程序：把一個大檔案切分成許多個分片，每個分片輸入給不同機器上的Map任務，并行執行完成“從檔案中決議出所有單詞”任務，

   • Map的輸入采用Hadoop默認的<k,v>輸入
     • 檔案行號key
     • 檔案一行為valueMap的輸出以單詞為key，1為value，<word,1>，單詞出現了1次
   • Shuffle對中間結果排序、磁區，得到<key,value-list>，分給Reduce任務
   • Reduce接到分配的中間結果（鍵值對），執行匯總計算，并把頻數結果輸出到分布式檔案系統

2.7.1wordcount程序圖

2.7.2作業流程

2.7.3資料分片

Map：

• Inputformat
  • 加載、讀取HDFS檔案，對輸入格式驗證
  • 大檔案邏輯分成許多分片split，定義每個split起點和長度
• record reader
  • 記錄閱讀器，根據split位置和長度，從HDFS各個塊讀取相關分片，<k,v>

2.7.4程序詳解

1. 資料分片
   
2. split的map程序
   
3. reduce
   
4. wordcount的map
   
5. wordcount的reduce
   
6. shuffle
   

2.7.5MapReduce作業程序

2.7.6MapReduce體系結構

2.7.6.1client客戶端

• 負責提交作業，查看作業狀態
• 提交作業：MapReduce程式通過client提交到JobTracker端
• 查看作業狀態：通過Client提供的介面查看作業運行狀態

2.7.6.2JobTracker作業跟蹤器

• 負責資源監控、作業調度
• 資源監控：JobTracker監控所有TaskTracker與Job的健康狀況，一旦發現節點失效（通信失敗或節點故障），就將相應的任務轉移到其他節點，
• 作業調度：JobTracker會跟蹤任務的執行進度、資源使用量等資訊，并將這些資訊告訴任務調度器（TaskScheduler），而TaskScheduler會選擇合適的（比較空閑）節點資源來執行任務，

2.7.6.3TaskScheduler任務調度器

• 接受JobTracker發送命令，執行具體任務
• 發送heartbeat（資源使用情況，任務運行進度）給JobTracker

2.7.6.4TaskTracker（任務跟蹤器）

• 周期性根據“心跳”，將本節點資源使用情況和任務運行進度匯報給JobTracker，同時接受JobTracker發送命令，執行相應操作（啟動信任我，殺死任務等）
• 使用slot等量劃分本節點資源（cpu，記憶體）
  • 一個task獲取一個slot才有機會運行
  • TaskScheduler將各個TaskTracker上的空閑slot分配給Task使用
  • Map slot供MapTask使用
  • Reduce slot供ReduceTask使用

3參考

• https://shenhao-stu.github.io/Big-Data/#/