最近在看一些大數(shù)據(jù)的東西，發(fā)現(xiàn)對其中的shuffle過程很模糊，于是決定學習一下，深入之后又發(fā)現(xiàn)對整個mapreduce的數(shù)據(jù)完成處理過程也同樣模糊。

所以本文將從以下幾個角度來展開：

• mapreduce以及hadoop框架的一些認識
• mapreduce的核心思想是什么
• mapreduce數(shù)據(jù)處理過程推演
• mapreduce的shuffle是如何實現(xiàn)的

Hadoop三劍客

Hadoop是一個由Apache開發(fā)的大數(shù)據(jù)處理框架，它包括了HDFS（Hadoop分布式文件系統(tǒng)）、YARN（Yet Another Resource Negotiator，資源管理器）以及MapReduce計算框架。

HDFS是Hadoop的存儲組件，YARN則是用于資源管理和調(diào)度的組件，而MapReduce是Hadoop用于分布式計算的框架。

在Hadoop中，數(shù)據(jù)通常存儲在HDFS中，通過MapReduce框架進行分布式計算，YARN負責管理計算資源，并協(xié)調(diào)MapReduce等計算框架的運行。

MapReduce、Hadoop、HDFS和YARN之間是相互依存、協(xié)同工作的關(guān)系，它們共同構(gòu)成了一個完整的大數(shù)據(jù)處理系統(tǒng)。

MapReduce核心思想

分而治之

MapReduce的主要思想是將大規(guī)模數(shù)據(jù)處理任務(wù)分解成多個小任務(wù)，并在分布式計算集群上并行執(zhí)行，從而實現(xiàn)高效的數(shù)據(jù)處理和分析。

MapReduce數(shù)據(jù)處理任務(wù)分為兩個主要階段：Map階段和Reduce階段。

在Map階段中，MapReduce將輸入數(shù)據(jù)分割成若干個小塊，然后在分布式計算集群上同時執(zhí)行多個Map任務(wù)，每個任務(wù)都對一個小塊的數(shù)據(jù)進行處理，并將處理結(jié)果輸出為一系列鍵值對，Map任務(wù)的輸出結(jié)果會被臨時存儲在本地磁盤或內(nèi)存中，以供Reduce任務(wù)使用。

在Reduce階段中，它主要負責對Map任務(wù)的輸出結(jié)果進行整合和匯總，生成最終的輸出結(jié)果。Reduce任務(wù)會將所有Map任務(wù)輸出的鍵值對按照鍵進行排序，并將相同鍵的值合并在一起，Reduce任務(wù)的輸出結(jié)果通常會寫入到文件系統(tǒng)中

計算向數(shù)據(jù)靠攏

在傳統(tǒng)的分布式計算中，數(shù)據(jù)通常需要從存儲介質(zhì)中讀取到計算節(jié)點進行處理，這會造成大量的數(shù)據(jù)傳輸和網(wǎng)絡(luò)延遲，導(dǎo)致計算效率較低。

"計算向數(shù)據(jù)靠攏"是MapReduce的一個設(shè)計思想，旨在最大化利用數(shù)據(jù)本地性（data locality）來提高作業(yè)的性能。

在MapReduce中，數(shù)據(jù)通常分布在集群的不同節(jié)點上，而處理數(shù)據(jù)的任務(wù)（Map任務(wù)和Reduce任務(wù)）盡量在數(shù)據(jù)所在的節(jié)點上執(zhí)行，減少數(shù)據(jù)的網(wǎng)絡(luò)傳輸和節(jié)點間的通信開銷，提高作業(yè)的性能。

MapReduce過程推演

我們嘗試從頭來推演整個MapReduce任務(wù)的大致執(zhí)行過程：

• 怎么寫跨語言的MapReduce作業(yè)？
• JobClient做了些什么？
• JobClient和Yarn是如何交互的？
• Hadoop2.x中AppMaster、NodeManager是如何協(xié)作的？
• 輸入文件分割是怎么實現(xiàn)的？

Hadoop Streaming

如果使用非 Java 編程語言來實現(xiàn) MapReduce 任務(wù)，或者希望更靈活地定制 Map 和 Reduce 函數(shù)的實現(xiàn)方式，可以考慮使用 Hadoop Streaming。

Hadoop Streaming 是 Hadoop 提供的一個工具，可以讓用戶通過標準輸入和標準輸出來實現(xiàn)自定義 Map 和 Reduce 函數(shù)的功能。使用 Hadoop Streaming 可以使用任何語言來實現(xiàn) Map 和 Reduce 函數(shù)，而不僅僅局限于 Java。

當使用 Hadoop Streaming 時，客戶端會將 Map 和 Reduce 函數(shù)打包成可執(zhí)行文件，然后提交給 Hadoop 集群來執(zhí)行。這些可執(zhí)行文件可以用任何編程語言編寫，例如 Python、Perl、Ruby、C++ 等。

JobClient

在提交任務(wù)之前，客戶端需要將任務(wù)的輸入數(shù)據(jù)和輸出路徑等信息設(shè)置好，以便Hadoop集群能夠正確地執(zhí)行任務(wù)并將結(jié)果輸出到指定的路徑。

MapReduce的客戶端JobClient通常會將任務(wù)打包成JAR包，然后將該JAR包提交給Hadoop集群來執(zhí)行任務(wù)。

JAR包中包含了MapReduce任務(wù)所需的輸入數(shù)據(jù)、輸出路徑、 Map 和 Reduce 的數(shù)量，以及每個任務(wù)需要的內(nèi)存和 CPU 資源等參數(shù)，這樣可以保證任務(wù)在集群中的任何節(jié)點上都能夠正常運行。

hadoop 1.x和2.x

在Hadoop 1.x版本中，JobTracker和TaskTracker是Hadoop集群中的兩個重要組件，其中JobTracker負責整個集群中所有MapReduce任務(wù)的協(xié)調(diào)和管理，而TaskTracker負責具體的任務(wù)執(zhí)行。

在Hadoop 2.x版本中引入了YARN框架，將JobTracker的功能拆分成兩部分，一部分是ResourceManager，負責集群資源的管理和分配，另一部分是ApplicationMaster，負責具體任務(wù)的管理和協(xié)調(diào)。

在Hadoop 2.x版本及以后的版本中，ApplicationMaster扮演的角色類似于JobTracker。

AppMaster

在 Hadoop YARN 中，ApplicationMaster是一個關(guān)鍵的組件，它負責在集群中管理和監(jiān)控一個特定的應(yīng)用程序。

在 MapReduce 中，每個 MapReduce 作業(yè)都有一個對應(yīng)的 ApplicationMaster 實例，該實例負責協(xié)調(diào)整個作業(yè)的執(zhí)行過程，包括分配任務(wù)、監(jiān)控任務(wù)的進度和狀態(tài)、處理任務(wù)失敗等。

當客戶端提交 MapReduce 作業(yè)時，YARN 資源管理器會為該作業(yè)啟動一個 ApplicationMaster 實例。

ApplicationMaster 將向資源管理器請求分配資源，并與各個 NodeManager 協(xié)商任務(wù)的執(zhí)行。它還負責將 MapReduce 作業(yè)的邏輯劃分為多個 Map 和 Reduce 任務(wù)，并將任務(wù)分配給相應(yīng)的 NodeManager 執(zhí)行。

在任務(wù)執(zhí)行期間，ApplicationMaster 將持續(xù)監(jiān)控任務(wù)的進度和狀態(tài)，并在任務(wù)出現(xiàn)錯誤或失敗時進行相應(yīng)的處理。

NodeManager

AppMaster和NodeManager不同，它們是YARN框架中的兩個不同的組件，分別扮演著不同的角色。

AppMaster是一個應(yīng)用程序級別的組件，它運行在分布式集群中的一個節(jié)點上，負責協(xié)調(diào)和管理應(yīng)用程序的生命周期。它向ResourceManager請求資源，然后將這些資源分配給它的任務(wù)（如Map和Reduce任務(wù)）。在任務(wù)運行期間，AppMaster監(jiān)視任務(wù)的進度并與ResourceManager通信，以確保應(yīng)用程序在分布式集群上有效地運行。

NodeManager是一個節(jié)點級別的組件，它運行在每個節(jié)點上，負責管理該節(jié)點上的容器和資源。在應(yīng)用程序啟動時，AppMaster向ResourceManager請求節(jié)點資源，并指示NodeManager在該節(jié)點上啟動容器來執(zhí)行應(yīng)用程序的任務(wù)。

NodeManager負責啟動容器并為它們分配資源，同時監(jiān)視它們的進度，并向ResourceManager報告資源使用情況。

HDFS輸入文件邏輯分割

客戶端程序中的Job對象會設(shè)置輸入文件的路徑和InputFormat類，在提交作業(yè)之前，Job會調(diào)用InputFormat的getSplits()方法來獲取輸入文件的切片信息。

InputFormat 是針對輸入文件進行邏輯分割，將一個或多個輸入文件劃分為一組輸入切片InputSplit，以便于 MapReduce 作業(yè)的并行處理。

對于大多數(shù)常見的數(shù)據(jù)類型，Hadoop 提供了一些內(nèi)置的 InputFormat 實現(xiàn)，如下：

• TextInputFormat：按行讀取文本文件，并將每行作為一個輸入記錄。
• KeyValueTextInputFormat：按行讀取鍵值對形式的文本文件，并將每個鍵值對作為一個輸入記錄。
• SequenceFileInputFormat：讀取 Hadoop 序列文件（SequenceFile），并將其中的每個 key-value 對作為一個輸入記錄。
• CombineFileInputFormat：支持讀取多個小文件或者多個小數(shù)據(jù)塊，并將它們合并成一個或多個輸入切片，以便于提高作業(yè)的并行度和執(zhí)行效率。

輸入文件的切片由InputFormat的getSplits()方法生成，這個方法會計算輸入文件的切片，并返回一個切片數(shù)組。每個切片都包含了一個起始偏移量和一個長度，這些信息告訴了Map任務(wù)它需要處理的輸入數(shù)據(jù)的范圍，ResourceManager會根據(jù)Split對象數(shù)組來計算作業(yè)需要的資源，并分配相應(yīng)的資源來運行作業(yè)。

當Map任務(wù)啟動時，它會使用InputFormat提供的InputSplit信息來創(chuàng)建一個RecordReader實例，用于讀取該Map任務(wù)需要處理的輸入數(shù)據(jù)。

RecordReader從HDFS或其他存儲系統(tǒng)中讀取數(shù)據(jù)，將數(shù)據(jù)劃分成適當大小的記錄，然后將它們轉(zhuǎn)換為鍵值對(key-value pairs)，再將它們傳遞給Mapper進行處理。

shuffle簡介

先看一個完整的圖，接下來會進行展開：

什么是shuffle

在 MapReduce 中，Map 和 Reduce 任務(wù)都是并行執(zhí)行的，Map 任務(wù)負責對輸入數(shù)據(jù)進行處理，將其轉(zhuǎn)換為鍵值對的形式，而 Reduce 任務(wù)負責對 Map 任務(wù)的輸出結(jié)果進行聚合和計算。

在 MapReduce 中，Shuffle 過程的主要作用是將 Map 任務(wù)的輸出結(jié)果傳遞給 Reduce 任務(wù)，并為 Reduce 任務(wù)提供輸入數(shù)據(jù)，它是 MapReduce 中非常重要的一個步驟，可以提高 MapReduce 作業(yè)效率。

Shuffle 過程的作用包括以下幾點：

• 合并相同 Key 的 Value：Map 任務(wù)輸出的鍵值對可能會包含相同的 Key，Shuffle 過程會將相同 Key 的 Value 合并在一起，減少 Reduce 任務(wù)需要處理的數(shù)據(jù)量。
• 按照 Key 進行排序：Shuffle 過程會將 Map 任務(wù)的輸出結(jié)果按照 Key 進行排序，這樣 Reduce 任務(wù)可以順序地處理鍵值對序列，避免在處理數(shù)據(jù)時需要進行額外的排序操作。
• 劃分數(shù)據(jù)并傳輸：Shuffle 過程會將 Map 任務(wù)的輸出結(jié)果按照 Key 劃分成多個分區(qū)，并將每個分區(qū)的數(shù)據(jù)傳輸?shù)綄?yīng)的 Reduce 任務(wù)中。這樣，Reduce 任務(wù)可以從不同的 Map 任務(wù)中獲取到數(shù)據(jù)，從而實現(xiàn)更好的并行化處理。

Map端的Shuffle

• Map任務(wù)在執(zhí)行過程中會將輸出的鍵值對寫入本地磁盤上的環(huán)形緩沖區(qū)中，按照Partitioner函數(shù)的規(guī)則將鍵值對映射到不同的區(qū)域Partition中。
• 當環(huán)形緩沖區(qū)滿或者Map任務(wù)執(zhí)行完畢時，Map任務(wù)將環(huán)形緩沖區(qū)中的數(shù)據(jù)進行排序，并按照Partition和排序結(jié)果寫入到本地磁盤上的Spill文件中。
• 當所有Map任務(wù)都執(zhí)行完畢后，MapReduce會按照Partition對所有Spill文件進行歸并排序，形成一個有序的大文件。

環(huán)形緩沖區(qū)的作用

環(huán)形緩沖區(qū)是一種特殊的緩沖區(qū)，它將數(shù)據(jù)存儲在一個固定大小的、循環(huán)的緩沖區(qū)中，當緩沖區(qū)滿時，新的數(shù)據(jù)將覆蓋最老的數(shù)據(jù)，于鏈表和數(shù)組等數(shù)據(jù)結(jié)構(gòu)，環(huán)形緩沖區(qū)具有以下優(yōu)勢：

• 高效的讀寫性能：由于環(huán)形緩沖區(qū)具有固定的大小，可以預(yù)先分配一定的空間，因此讀寫操作非常高效。而且由于緩沖區(qū)是循環(huán)的，可以通過簡單的指針操作來實現(xiàn)數(shù)據(jù)的讀寫操作，而不需要復(fù)雜的數(shù)據(jù)移動操作。
• 空間利用率高：由于環(huán)形緩沖區(qū)可以循環(huán)利用空間，因此空間利用率非常高。相比于鏈表和數(shù)組等數(shù)據(jù)結(jié)構(gòu)，它不需要額外的空間來維護節(jié)點或索引等信息，可以更好地利用內(nèi)存空間。
• 易于實現(xiàn)和管理：由于環(huán)形緩沖區(qū)的結(jié)構(gòu)非常簡單，因此易于實現(xiàn)和管理。可以通過一些簡單的技術(shù)來解決緩沖區(qū)溢出等問題，從而保證 Shuffle 過程的正確性。

分區(qū)&排序&溢寫

在 Map 階段中，Map 任務(wù)會將輸出數(shù)據(jù)寫入環(huán)形緩沖區(qū)，而這些數(shù)據(jù)會被分為多個分區(qū)，并且每個分區(qū)內(nèi)的數(shù)據(jù)是按照鍵（Key）進行排序的。

分區(qū)的劃分是由 Partitioner 類完成的，默認情況下，Partitioner 類會根據(jù)鍵值對的鍵（Key）來計算分區(qū)編號（Partition ID），從而將數(shù)據(jù)分配到對應(yīng)的分區(qū)中。

不同的鍵（Key）可能會被分到同一個分區(qū)中，而相同的鍵（Key）則一定會被分到同一個分區(qū)中，這是為了保證相同鍵（Key）的數(shù)據(jù)能夠被發(fā)送到同一個 Reduce 任務(wù)中進行處理。

Map 任務(wù)向環(huán)形緩沖區(qū)中寫入數(shù)據(jù)時，先將數(shù)據(jù)插入到分區(qū)內(nèi)，然后對該分區(qū)內(nèi)的所有數(shù)據(jù)進行快速排序。

當環(huán)形緩沖區(qū)中的數(shù)據(jù)量達到一定閾值（MapReduce 1.x 中默認為環(huán)形緩沖區(qū)大小的 0.8 倍），或者某個分區(qū)內(nèi)存放的數(shù)據(jù)大小達到一定閾值（MapReduce 1.x 中默認為 100MB），就會觸發(fā)溢寫操作，將數(shù)據(jù)按照分區(qū)寫到磁盤上的臨時文件中。

Reduce端的Shuffle

• map的shuffle階段會生成包含不同分區(qū)的大文件
• reduce任務(wù)可能會多個不同的分區(qū)，但是同一個分區(qū)的數(shù)據(jù)一定在同一個reduce中
• reduce任務(wù)收集來自多個map輸出的同一個分區(qū)的數(shù)據(jù)，在內(nèi)部再針對同一分區(qū)的多個文件做歸并成一個大文件

小結(jié)

很多時候明確問題比知道答案更重要，多想為什么、多在腦海里去推演過程、最終才能自洽吸收外界知識，化為自己的經(jīng)驗。?