MapReduce是什麼

Hadoop MapReduce是一個軟體框架，基於此框架能夠容易地編寫應用程式，這些應用程式能夠執行在由上千個商用機器組成的大叢集上，並以一種可靠的，具有容錯能力的方式並行地處理上TB級別的海量資料集。簡單來說就是MapReduce就是一個計算框架。

MapReduce做什麼

MapReduce的思想就是“分而治之”。

• Mapper負責”分“
  把複雜的任務分解成若干個“簡單的任務”來處理。“簡單的任務“包含三層含義：
  1.資料或計算的規模相對原任務要大大縮小；
  2.就近計算原則，任務會分配到存放著所需要資料的節點上進行計算；
  3.這些小任務可以平行計算彼此間幾乎沒有依賴關係。

-Reducer負責對map階段的結果進行彙總
至於需要多個Reducer，可以根據具體問題，通過在mapred-site.xml配置檔案裡設定引數mapred.reduce.tasks的值，預設值為1。

MapReduce工作機制

MapReduce工作機制

mapreduce作業工作流程圖

• 步驟一
  客戶端編寫好mapreduce程式，配置好mapreduce的作業就是job，接下來提交job了，提交job是提交到JobTracker上的，這個時候JobTracker就會構建這個job，具體就是分配一個新的job任務的ID值。
  接下來它會做檢查操作，這個檢查就是確定輸出目錄是否存在，如果存在那麼job就不能正常執行下去，JobTracker就會丟擲錯誤給客戶端，接下來還要檢查輸入目錄是否存在，如果不存在也同樣丟擲錯誤，如果存在JobTracker會根據輸入計算輸入分片（Input Split），如果分片計算不出來也會丟擲錯誤。當這些都完成了JobTracker就會配置Job所需要的資源了。
  分配好資源後，JobTracker就會初始化作業，初始化主要做的是將job放在一個內部的佇列，讓配置好的作業排程器能夠排程到這個作業，作業排程器會初始化這個job，初始化就是建立一個正在執行的job物件（封裝任務和記錄資訊），以便JobTracker跟蹤job的狀態和程序。

• 步驟二
  初始化完畢後，作業排程器會獲取輸入分片資訊（input split），每個分片建立一個map任務。
  接下來就是任務分配了，這個時候tasktracker會執行一個簡單的迴圈機制定期傳送心跳給jobtracker，心跳間隔是5秒，程式設計師可以配置這個時間，心跳就是jobtracker和tasktracker溝通的橋樑，通過心跳，jobtracker可以監控tasktracker是否存活，也可以獲取tasktracker處理的狀態和問題，同時tasktracker也可以通過心跳裡的返回值獲取jobtracker給它的操作指令。
  任務分配好後就是執行任務了。在任務執行時候jobtracker可以通過心跳機制監控tasktracker的狀態和進度，同時也能計算出整個job的狀態和進度，而tasktracker也可以本地監控自己的狀態和進度。當jobtracker獲得了最後一個完成指定任務的tasktracker操作成功的通知時候，jobtracker會把整個job狀態置為成功，然後當客戶端查詢job執行狀態時候（注意：這個是非同步操作），客戶端會查到job完成的通知的。如果job中途失敗，mapreduce也會有相應機制處理，一般而言如果不是程式設計師程式本身有bug，mapreduce錯誤處理機制都能保證提交的job能正常完成。

MapReduce執行機制

MapReduce執行機制

• 在Hadoop中，一個MapReduce作業會把輸入的資料集切分為若干獨立的資料塊，由Map任務以完全並行的方式處理
• 框架會對Map的輸出先進行排序，然後把結果輸入給Reduce任務。
• 作業的輸入和輸出都會被儲存在檔案系統中，整個框架負責任務的排程和監控，以及重新執行已經關閉的任務
• MapReduce框架和分散式檔案系統是執行在一組相同的節點，計算節點和儲存節點都是在一起的

MapReduce框架的組成

MapReduce框架的組成

MapReduce作業處理流程

作業處理流程

按照時間順序包括：

• 輸入分片（input split）：
  在進行map計算之前，mapreduce會根據輸入檔案計算輸入分片（input split），每個輸入分片（input split）針對一個map任務
  輸入分片（input split）儲存的並非資料本身，而是一個分片長度和一個記錄資料的位置的陣列，輸入分片（input split）往往和hdfs的block（塊）關係很密切

  假如我們設定hdfs的塊的大小是64mb，如果我們輸入有三個檔案，大小分別是3mb、65mb和127mb，那麼mapreduce會把3mb檔案分為一個輸入分片（input split），65mb則是兩個輸入分片（input split）而127mb也是兩個輸入分片（input split）
  即我們如果在map計算前做輸入分片調整，例如合併小檔案，那麼就會有5個map任務將執行，而且每個map執行的資料大小不均，這個也是mapreduce優化計算的一個關鍵點。

• map階段：
  程式設計師編寫好的map函數了，因此map函式效率相對好控制，而且一般map操作都是本地化操作也就是在資料儲存節點上進行；

• combiner階段：
  combiner階段是程式設計師可以選擇的，combiner其實也是一種reduce操作，因此我們看見WordCount類裡是用reduce進行載入的。
  Combiner是一個本地化的reduce操作，它是map運算的後續操作，主要是在map計算出中間檔案前做一個簡單的合併重複key值的操作，例如我們對檔案裡的單詞頻率做統計，map計算時候如果碰到一個hadoop的單詞就會記錄為1，但是這篇文章裡hadoop可能會出現n多次，那麼map輸出檔案冗餘就會很多，因此在reduce計算前對相同的key做一個合併操作，那麼檔案會變小，這樣就提高了寬頻的傳輸效率，畢竟hadoop計算力寬頻資源往往是計算的瓶頸也是最為寶貴的資源，但是combiner操作是有風險的，使用它的原則是combiner的輸入不會影響到reduce計算的最終輸入，
  例如：如果計算只是求總數，最大值，最小值可以使用combiner，但是做平均值計算使用combiner的話，最終的reduce計算結果就會出錯。

• shuffle階段：
  將map的輸出作為reduce的輸入的過程就是shuffle了

• reduce階段：
  和map函式一樣也是程式設計師編寫的，最終結果是儲存在hdfs上的。

MapReduce相關問題

• jobtracker的單點故障：
  jobtracker和hdfs的namenode一樣也存在單點故障，
  單點故障一直是hadoop被人詬病的大問題，
  為什麼hadoop的做的檔案系統和mapreduce計算框架都是高容錯的，但是最重要的管理節點的故障機制卻如此不好，我認為主要是namenode和jobtracker在實際執行中都是在記憶體操作，而做到記憶體的容錯就比較複雜了，只有當記憶體資料被持久化後容錯才好做，namenode和jobtracker都可以備份自己持久化的檔案，但是這個持久化都會有延遲，因此真的出故障，任然不能整體恢復，另外hadoop框架裡包含zookeeper框架，zookeeper可以結合jobtracker，用幾臺機器同時部署jobtracker，保證一臺出故障，有一臺馬上能補充上，不過這種方式也沒法恢復正在跑的mapreduce任務。

• 不能存在output資料夾
  做mapreduce計算時候，輸出一般是一個資料夾，而且該資料夾是不能存在，我在出面試題時候提到了這個問題，而且這個檢查做的很早，當我們提交job時候就會進行，mapreduce之所以這麼設計是保證資料可靠性，如果輸出目錄存在reduce就搞不清楚你到底是要追加還是覆蓋，不管是追加和覆蓋操作都會有可能導致最終結果出問題，mapreduce是做海量資料計算，一個生產計算的成本很高，例如一個job完全執行完可能要幾個小時，因此一切影響錯誤的情況mapreduce是零容忍的。

• Mapreduce還有一個InputFormat和OutputFormat。
  我們在編寫map函式時候發現map方法的引數是之間操作行資料，沒有牽涉到InputFormat，這些事情在我們new Path時候mapreduce計算框架幫我們做好了，而OutputFormat也是reduce幫我們做好了，我們使用什麼樣的輸入檔案，就要呼叫什麼樣的InputFormat，InputFormat是和我們輸入的檔案型別相關的，mapreduce裡常用的InputFormat有FileInputFormat普通文字檔案，SequenceFileInputFormat是指hadoop的序列化檔案，另外還有KeyValueTextInputFormat。OutputFormat就是我們想最終儲存到hdfs系統上的檔案格式了，這個根據你需要定義了，hadoop有支援很多檔案格式，這裡不一一列舉，想知道百度下就看到了

連結：https://www.jianshu.com/p/196d490a6f44