MapReduce詳解

MapReduce設計理念

map–>映射(key value)

reduce–>歸納

mapreduce必須構建在hdfs之上一種大資料離線計算框架

• 在線：實時資料處理
• 離線：資料處理時效性沒有在線那么強，但是相對也需要很快得到結果

mapreduce不會馬上得到結果,他會有一定的延時

• 如果資料量小,使用mapreduce反而不合適
• 殺雞用牛刀

原始資料–>map(Key,Value)–>Reduce

分布式計算

• 將大的資料切分成多個小資料，交給更多的節點參與運算

計算向資料靠攏

• 將計算傳遞給有資料的節點上進行作業

MR的計算流程

原始資料File

• 1T資料被切分成塊存放在HDFS上，每一個塊有128M大小

資料塊Block

• hdfs上資料存盤的一個單元,同一個檔案中塊的大小都是相同的
• 因為資料存盤到HDFS上不可變，所以有可能塊的數量和集群的計算能力不匹配
• 我們需要一個動態調整本次參與計算節點數量的一個單位
• 我們可以動態的改變這個單位–》參與的節點

切片Split

• 切片是一個邏輯概念

• 在不改變現在資料存盤的情況下，可以控制參與計算的節點數目

• 通過切片大小可以達到控制計算節點數量的目的

  • 有多少個切片就會執行多少個Map任務

• 一般切片大小為Block的整數倍(2 1/2)

  • 防止多余創建和很多的資料連接

• 如果Split>Block ,計算節點少了

• 如果Split<Block ,計算節點多了

• 默認情況下，Split切片的大小等于Block的大小 ,默認128M

• 一個切片對應一個MapTask

MapTask

• map默認從所屬切片讀取資料,每次讀取一行（默認讀取器）到記憶體中

• 我們可以根據自己書寫的分詞邏輯(空格分隔).計算每個單詞出現的次數

• 這是就會產生 (Map<String,Integer>)臨時資料，存放在記憶體中

• 但是記憶體大小是有限的，如果多個任務同時執行有可能記憶體溢位（OOM）

• 如果把資料都直接存放到硬碟，效率太低

• 我們需要在OOM和效率低之間提供一個有效方案

  • 可以現在記憶體中寫入一部分，然后寫出到硬碟

環形資料緩沖區

• 可以回圈利用這塊記憶體區域，減少資料溢寫時map的停止時間

  • 每一個Map可以獨享的一個記憶體區域
  • 在記憶體中構建一個環形資料緩沖區(kvBuffer),默認大小為100M
  • 設定緩沖區的閾值為80%,當緩沖區的資料達到80M開始向外溢寫到硬碟
  • 溢寫的時候還有20M的空間可以被使用效率并不會被級訓
  • 而且將資料回圈寫到硬碟，不用擔心OOM問題

磁區Partation

• 根據Key直接計算出對應的Reduce

• 磁區的數量和Reduce的數量是相等的

• hash(key) % partation = num

• 默認磁區的演算法是Hash然后取余

  • Object的hashCode()—equals()
  • 如果兩個物件equals,那么兩個物件的hashcode一定相等
  • 如果兩個物件的hashcode相等，但是物件不一定equlas

排序Sort

• 對要溢寫的資料進行排序（QuickSort）
• 按照先Partation后Key的順序排序–>相同磁區在一起,相同Key的在一起
• 我們將來溢寫出的小檔案也都是有序的

溢寫Spill

• 將記憶體中的資料回圈寫到硬碟，不用擔心OOM問題
• 每次會產生一個80M的檔案
• 如果本次Map產生的資料較多，可能會溢寫多個檔案

合并Merge

• 因為溢寫會產生很多有序(磁區 key)的小檔案，而且小檔案的數目不確定
• 后面向reduce傳遞資料帶來很大的問題
• 所以將小檔案合并成一個大檔案，將來拉取的資料直接從大檔案拉取即可
• 合并小檔案的時候同樣進行排序(歸并排序),最終產生一個有序的大檔案

拉取Fetch

• 我們需要將Map的臨時結果拉取到Reduce節點

• 原則

  • 相同的Key必須拉取到同一個Reduce節點
  • 但是一個Reduce節點可以有多個Key

• 未排序前拉取資料的時候必須對Map產生的最終的合并檔案做全序遍歷

  • 而且每一個reduce都要做一個全序遍歷

• 如果map產生的大檔案是有序的，每一個reduce只需要從檔案中讀取自己所需的即可

合并Merge

• 因為reduce拉取的時候，會從多個map拉取資料
• 那么每個map都會產生一個小檔案,這些小檔案（檔案與檔案之間無序，檔案內部有序）
• 為了方便計算（沒必要讀取N個小檔案）,需要合并檔案
• 歸并演算法合并成2個(qishishilia)
• 相同的key都在一起

歸并Reduce

• 將檔案中的資料讀取到記憶體中
• 一次性將相同的key全部讀取到記憶體中
• 直接將相同的key得到結果–>最終結果

寫出Output

• 每個reduce將自己計算的最終結果都會存放到HDFS上

MapReduce架構

MapReduce1.x

• client

  • 客戶端發送mr任務到集群

  • 客戶端的種類有很多種

    • hadoop jar wordcount.jar

• JobTracker

  • 接受客戶端的mr任務

    • 選擇一個資源豐富的，執行對應的任務
    • 并且給這個任務分配資源

  • 與TaskTracker保持心跳，接受匯報資訊

• TaskTracker

  • 保持心跳，匯報資源

    • 當前機器記憶體，當前機器任務數

  • 當分配資源之后，開始在本機分配對應的資源給Task

  • 并且實時監控任務的執行,并匯報

• Slot(槽)

  • 屬于JobTracker分配的資源
  • 計算能力，IO能力…
  • 不管任務大小，資源是恒定的，不靈活但是好管理

• Task(MapTask–ReduceTask)

  • 開始按照MR的流程執行業務
  • 當任務完成時，JobTracker告訴TaskTracker回收資源

• 缺點

  • 單點故障

  • 記憶體擴展

  • 業務瓶頸

  • 只能執行MR的操作

    • 如果其他框架需要運行在Hadoop上，需要獨立開發自己的資源調度框架

MapReduce2.x

• 2.x開始使用Yarn（Yet Another Resource Negotiator，另一種資源協調者）統一管理資源

• 以后其他的計算框架可以直接訪問yarn獲取當前集群的空閑節點

• client

  • 客戶端發送mr任務到集群

  • 客戶端的種類有很多種

    • hadoop jar wordcount.jar

• ResourceManager

  • 資源協調框架的管理者

  • 分為主節點和備用節點（防止單點故障）

    • 主備的切換基于Zookeeper進行管理

  • 時刻與NodeManager保持心跳，接受NodeManager的匯報

    • NodeManager匯報當前節點的資源情況

  • 當有外部框架要使用資源的時候直接訪問ResourceManager即可

  • 如果有MR任務，先去ResourceManager申請資源，ResourceManager根據匯報相對靈活分配資源

    • 資源在NodeManager1，NodeManager1要負責開辟資源

• NodeManager

  • 資源協調框架的執行者
  • 每一個DataNode上默認有一個NodeManager
  • NodeManager匯報自己的資訊到ResourceManager

• Container

  • 2.x資源的代名詞
  • Container動態分配的

• ApplicationMaster

  • 我們本次JOB任務的主導者
  • 負責調度本次被分配的資源Container
  • 當所有的節點任務全部完成，application告訴ResourceManager請求殺死當前ApplicationMaster執行緒
  • 本次任務所有的資源都會被釋放

• Task(MapTask–ReduceTask)

  • 開始按照MR的流程執行業務
  • 當任務完成時，ApplicationMaster接收到當前節點的回饋