簡介：本文對MapReduce物理、邏輯工作流以及shuffle過程做簡要整理

---

MapReduce采用的是分而治之的思想，把對大規模數據集的操作，分發給一個主節點管理下的各個從節點共同完成，然后通過整合各個節點的中間結果，得到最終結果。簡單來說，MapReduce就是”任務的分解與結果的匯總。

物理實體工作流

MapReduce體系結構主要由四個部分組成， 分別是：Client、JobTracker、TaskTracker以及Task

組成結構

Client

• 用戶編寫的MapReduce通過Client提交到JobTracker端

• 用戶通過Client提供的接口查看作業運行狀態

JobTracker

• JobTracker位于NameNode上

• JobTracker負責資源監控和作業調度，并與TaskTracker通信

• JobTracker監聽TaskTracker的健康狀況heartbeat，一旦失敗就將任務轉移至其他節點

• JobTracker通過任務調度器TaskScheduler選擇任務的執行次序，以便為基于隊列的FIFO調度器

TaskTracker

TaskTracker位于DataNode中，是JobTracker和Task的橋梁

匯報心跳：TaskTracker周期性將所有節點上各種信息通過心跳機制匯報給JobTracker。這些信息包括兩部分：

• 機器級別信息：節點健康情況、資源使用情況等。
• 任務級別信息：任務執行進度、任務運行狀態等。

執行命令：JobTracker會給TaskTracker下達各種命令，主要包括：啟動任(LaunchTaskAction)、提交任務(CommitTaskAction)、殺死任務(KillTaskAction)、殺死作業(KillJobAction)和重新初始化(TaskTrackerReinitAction)。

TaskTracker 使用“slot”等量劃分本節點上的資源量（CPU、內存等）。一個Task 獲 取到一個slot 后才有機會運行，而Hadoop調度器的作用就是將各個TaskTracker上的 空閑slot分配給Task使用。slot 分為Map slot 和Reduce slot 兩種，分別供MapTask 和 Reduce Task 使用

Task

Task分為Map Task和Reduce Task兩種， 均由TaskTracker 啟動
Map：對Map Task讀到的一行數據進行處理
Reduce：對Reduce Task獨到的同key數據進行處理

工作流程

• 提交作業。用戶將所有應該配置的參數根據需求配置好，并通過編寫的MapReduce程序通過客戶端（Client）向Hadoop集群的JobTracker發送作業提交請求。作業提交之后，就會進入自動化執行，用戶只能監控程序的執行情況和強制中斷作業，不能對作業的執行過程進行任何干預。
• 初始化作業。JobTracker 在 JobTracker 端開始初始化工作，包括在其內存里建立一系列數據結構，來記錄這個 Job 的運行情況。
• 分配任務。JobTracker 會向 HDFS 的 NameNode 詢問有關數據在哪些文件里面，這些文件分別散落在哪些結點里面。JobTracker 需要按照“就近運行”原則分配任務。
• 執行任務。TaskTracker 分配到一個任務后，通過 HDFS 把作業的 Jar 文件復制到 TaskTracker 所在的文件系統，同時，TaskTracker 將應用程序所需要的全部文件從分布式緩存復制到本地磁盤。TaskTracker 為任務新建一個本地工作目錄，并把 Jar 文件中的內容解壓到這個文件夾中。TaskTracker 啟動一個新的 JVM 來運行每個任務（包括 Map 任務和 Reduce 任務），這樣，JobClient 的 MapReduce 就不會影響 TaskTracker 守護進程。任務的子進程每隔幾秒便告知父進程它的進度，直到任務完成。
• 進程和狀態的更新。一個作業和它的每個任務都有一個狀態信息，包括作業或任務的運行狀態，Map 任務和 Reduce 任務的進度，計數器值，狀態消息或描述。任務在運行時，對其進度保持追蹤。
• 作業的完成。當 JobTracker 接收到的這次作業的最后一個任務已經完成時，它會將 Job 的狀態改為“successful”。當 JobClient 獲取到作業的狀態時，就知道該作業已經成功完成，然后 JobClient 打印信息告知用戶作業已成功結束。

---

邏輯實體工作流

MapReduce工作流程

InputFormat預處理。驗證輸入格式是否要求；對輸入文件使用InputSplit進行邏輯切片，大多數情況下，理想的分片大小是一個HDFS塊（block），每個邏輯切片將由一個Map Task進行處理；通過RecordReader(RR)將切片數據轉換為適合Map任務處理的鍵值對，輸入給Map任務

執行Map操作。根據用戶定義的Map()函數，處理數據，輸出<key,value>

Map端的Shuffle。對Map輸出結果進行分區（Partition）、排序（Sort）、合并（Combine）、歸并（Merge）等操作，得到<key,value-list>的中間結果

Reduce端的Shuffle。Reduce以<key,value-list>為輸入，執行用戶自定義的邏輯，輸出<key,value>到OutputFormat模塊

OutputFormat模塊會驗證輸出目錄是否已經存在、輸出結果類型是否符合配置文件中的配置類型，若滿足則輸出值HDFS中

Shuffle

Shuffle是對Map輸出結果進行分區Partition、排序Sort、合并Combiner等處理并交給Reduce的過程。Shuffle分為Map端和Reduce端兩部分操作

Map端的Shuffle過程

寫入緩存。每個Map任務會被分配一個緩存，在執行Map的過程中會將執行結果寫入緩存，為提高效率不直接寫入磁盤

溢寫。當緩存已用容量達到一定比例后，就會啟動溢寫（Spill）操作，將緩存中的內容寫入磁盤，清空緩存。 在寫入磁盤前，緩存中的數據會被分區（Partition），以便均勻分配給Reduce任務出并行處理；對每個分區內的鍵值對，會根據key進行排序（Sort），并選擇是否對數據進行合并（Combine）操作，以減少溢寫到磁盤的數據量。每次溢寫操作都會在磁盤中生成一個新的溢寫文件

隨著溢寫文件的增多，系統會對所有溢寫文件進行歸并（Merge）操作以生成一個大的溢寫文件。如：將<k1,v1>，<k1,v2>，<k1,v3>歸并成<k1,<v1,v2,v3>>。Map端的Shuffle最終會生成一個大文件存儲在本地磁盤上，這個大文件中的數據是被分區的，不同分區會被發送到不同的Reduce任務進行并行處理。

Reduce端的Shuffle過程

從Map端讀取Map結果，進行歸并操作，輸出給Reduce任務

領取（Fetch）數據。Reduce任務會不斷通過RPC輪詢JobTracker關于Map任務是否執行完成，若有Map任務執行完畢，那么就會有相關Reduce任務來Fetch數據到本地磁盤中

歸并（Merge）數據。從Map端取回的數據會放入Reduce任務所在節點的緩存中，當緩存區滿了，那么會溢寫到磁盤中形成多個磁盤文件，若磁盤文件超出一定數量則會被歸并成一個大文件

將歸并的大文件輸入給Reduce任務。接下來，Reduce任務會執行用戶自定義的Reduce函數，輸出結果并保存到DFS中

注意：

不同的Map任務之間不會進行通信

不同的Reduce任務之間也不會發生任何信息交換

用戶不能顯式地從一臺機器向另一臺機器發送消息

所有的數據交換都是通過MapReduce框架自身去實現的

其他細節知識

split與block

split（切片）是MapReduce里的概念；而block（塊）是hdfs中物理切塊的大小
HDFS以固定大小的block為基本單位存儲數據，而對于MapReduce而言，其處理單位是split。split只包含一些元數據信息，比如數據起始位置、數據長度、數據所在節點等。它的劃分方法完全由用戶自己決定。

hadoop在默認的情況下，split和hdfs的block的大小是一樣的，但也可以不同。可以通過配置文件進行設置：

minsize默認大小為1，可以通過屬性mapreduce.input.fileinputformat.split.minsize進行設置。

maxsize默認大小為Long.MAXValue，可以通過屬性mapreduce.input.fileinputformat.split.maxsize進行設置。

在mapreduce的FileInputFormat類中有個getSplits()方法對文件進行spli

---

參考：
2022最新黑馬程序員大數據Hadoop入門視頻教程_Map_Reduce_bilibili
InputFormat&InputSplit&RecordReader解析-CSDN博客

總結

以上是生活随笔為你收集整理的MapReduce工作流理解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。