1. Hadoop MapReduce簡介

Hadoop MapReduce是一個使用簡便的軟件框架，是Google云計算模型MapReduce的Java開源實現(xiàn)，基于它寫出來的應用程序能夠運行在由上千萬臺普通機器注冊的大型集群系統(tǒng)中，并以一種可靠地、容錯的方式并行處理上T級別的數(shù)據(jù)集。

Hadoop MapReduce基本思想：一個MapReduce作業(yè)通常會把輸入的數(shù)據(jù)集合切分為若干獨立的數(shù)據(jù)塊，由Map任務并行的方式處理。該框架會對Map的輸出先進行排序，然后把結(jié)果輸出作為Reduce任務的輸入。通常作用的輸入和輸出都會存儲在文件系統(tǒng)中。

1.1 系統(tǒng)架構(gòu)

在系統(tǒng)架構(gòu)上，MapReduce框架是一種主從架構(gòu)，由一個單獨的JobTracker節(jié)點和多個TaskTracker節(jié)點共同組成。

JobTracker是MapReduce的Master，負責調(diào)度構(gòu)成一個作業(yè)的所有任務，這些任務分布在不同 的TaskTracker節(jié)點上，監(jiān)控它們的執(zhí)行，重新執(zhí)行已經(jīng)失敗的任務，同時提高狀態(tài)和診斷信息給作業(yè)客戶端。

TaskTracker是MapReduce的Slave，僅負責運行由Master指派的任務執(zhí)行。

Hadoop的作業(yè)客戶端提交作業(yè)（jar包和可執(zhí)行程序）和配置信息作為Master的JobTracker，JobTracker負責分發(fā)用戶程序和配置信息給集群中的TaskTracker，以及調(diào)度任務并監(jiān)控他們的執(zhí)行，同時提供狀態(tài)和診斷信息給作業(yè)客戶端。

2. MapReduce模型

2.1 MapReduce編程模型

對于Map函數(shù)，處理輸入的鍵值對，并且產(chǎn)生一組中間的鍵值對。MapReduce框架收集所有相同的中間鍵值的鍵值對，并且發(fā)送給Reduce函數(shù)進行處理。對于Reduce函數(shù)，它處理中間鍵的鍵值對，以及這個中間鍵值對相關(guān)的值集合。此函數(shù)合并這些值，最后形成一個相對較少的值集合。

2.2 MapReduce實現(xiàn)原理

用戶程序中的MapReduce函數(shù)庫首先把輸入文件分成M塊（Hadoop默認128M），接著在集群機器中執(zhí)行處理程序。

主控程序master分配Map任務和Reduce任務給工作執(zhí)行機器Worker。總該有M個任務和R個reduce任務需要分配。master會選擇空閑的Worker并且分配這些Map任務或者Reduce任務給Worker節(jié)點。

一個分配了Map任務的worker讀取并處理相關(guān)輸入的數(shù)據(jù)塊。從輸入的數(shù)據(jù)片段中解析key/value鍵值對，然后把key/value鍵值對傳遞給用戶自定義的map函數(shù)，map函數(shù)生成并輸出key/vaule鍵值對集合，這些集合會暫時緩存內(nèi)存中。

緩存中的key/value鍵值對通過分區(qū)函數(shù)分成R個區(qū)域，之后周期性地寫入本地磁盤上。同時緩存key/value鍵值對集合在本地磁盤上的存儲位置發(fā)送給master節(jié)點，由master再把這些記錄傳送給Reduce worker。

當Reduce worker程序接收到master程序發(fā)送過來的數(shù)據(jù)存儲位置信息之后，使用RPC從Map worker所在的主機磁盤上讀取這些緩存數(shù)據(jù)。在Reduce Worker讀取了所有的中間數(shù)據(jù)之后，通過key進行排序后使得具有相同key值的數(shù)據(jù)聚合在一起。由于許多不同的key值會映射到相同的Reduce任務上，因此必須進行排序。

Reduce Worker程序遍歷排序后的中間數(shù)據(jù)。對于每一個唯一的中間key值，Reduce Worker程序都會將這個key值和它相關(guān)的中間vaule值的集合傳遞給用戶自定義的Reduce函數(shù)。Reduce函數(shù)的輸出被追加到所屬分區(qū)的輸出文件中。

3. 計算流程與機制

3.1 Hadoop作業(yè)提交和初始化

MapReduce的Master接收到客戶端所提交的作業(yè)后首先要完成的就是將作業(yè)初始化為map任務和reduce任務，然后就是等待JobTracker調(diào)度執(zhí)行。

命令行提交。用戶使用Hadoop命令行腳本提交Mapreduce程序到集群中。

作業(yè)上次。在提交作業(yè)到JobTracker之前還需要完成相關(guān)的初始化工作。這些工作包括用戶作業(yè)的JobId，創(chuàng)建HDFS目錄，上傳作業(yè)、相關(guān)依賴庫、需要分發(fā)的文件到HDFS中，同時還包括用戶輸入數(shù)據(jù)的所有分片信息。

產(chǎn)生分片文件。在作業(yè)提交之后，JobClient調(diào)用InputFormat的getSplit()方法產(chǎn)生用戶數(shù)據(jù)的split分片信息，這些信息包括Input元數(shù)據(jù)信息，原始切分數(shù)據(jù)信息，其中元數(shù)據(jù)信息會被JobTracker使用，原始切分信息會在Map任務初始化的時候獲取自己需要處理的數(shù)據(jù)信息。這兩部分數(shù)據(jù)被保存在job.split文件和job.splitmetainfo文件中

提交作業(yè)到JobTracker。JobClient通過遠程調(diào)用RPC將作業(yè)提交到JobTracker作業(yè)調(diào)度器中，首先為作業(yè)創(chuàng)建JobInprogress對象。JobTracker會為用戶提交的每一個作業(yè)創(chuàng)建JobInprogress對象，這個對象維護了作業(yè)的運行時的信息，主要用于跟蹤正在運行的任務的狀態(tài)和進度。其次，檢查用戶是否具有指定隊列的作業(yè)提交權(quán)限。

通過上面的操作就完成了MapReduce作業(yè)的提交工作了，那么接下來就開始作業(yè)的初始化操作：作業(yè)的初始化操作主要指的就是構(gòu)造MapTask和ReduceTask并且對他們進行初始化操作，這一步操作主要是調(diào)度器JobTracker.initJob()方法來進行的。具體情況是Hadoop將每個作業(yè)分成4個不同類型的任務：Setup Task、Map Task、Reduce Task、Cleanup Task。

3.2 Mapper

Mapper是MapReduce框架給用戶暴露的Map編程接口，用戶在實現(xiàn)自己的Mapper類時需要繼承這個基類。執(zhí)行Map Task任務：將輸入鍵值對（key/value pair）映射到一組中間格式的鍵值對集合。

處理流程如下：

通過InputFormat接口獲得InputSplit的實現(xiàn)，然后對輸入的數(shù)據(jù)切分。每一個Split分塊對應一個Mapper任務。

通過RecordReader對象讀取生成<k,v>鍵值對。Map函數(shù)接受數(shù)據(jù)并處理后輸出<k1,v1>鍵值對。

通過context.collect方法寫入context對象中。當鍵值對集中被收集后，會被Partition類中的partition()函數(shù)以指定方式區(qū)分并寫入輸出緩沖區(qū)(系統(tǒng)默認的是HashPartitioner)，同時調(diào)用sort()進行排序。

如果用戶指定了Combiner，則會將鍵值對進行combine合并(相當于map端的reduce)，輸出到reduce寫入文件。

3.3 Reducer

Reducer將與一個key關(guān)聯(lián)的一組中間數(shù)值集歸約為一個更小的數(shù)值集。

Shuffle階段。框架通過HTTP協(xié)議為每個Reducer獲得所有Mapper輸出中與之相關(guān)的分塊，這一階段也稱混洗階段，所做的大量操作就是數(shù)據(jù)復制，因此也可以稱為數(shù)據(jù)復制階段。

Sort階段。 框架按照key的值對Reducer的輸入進行分組(因為不同的Mapper輸出可能會有相同的key)。 Shuffle和Sort是同時進行的，Map的輸出也是一邊被取回一邊被合并的。 如果需要改變分組方式，則需要指定一個Compartor，實現(xiàn)二次排序（后面會介紹）。

Reduce階段。 調(diào)用Reduce()函數(shù)，對Shuffle和sort得到的<key,(list of values)>進行處理，輸出結(jié)果到DFS中。

3.4 Reporter和OutputCollector

Report是用于MapReduce應用程序的報告進度，設定應用級別的狀態(tài)信息，更新Counters(計數(shù)器)的機制。Master和Reducer的處理情況可以利用Reporter來報告進度或者表明自己是運行正常的

OutputCollector是一個由Map/Reduce框架提供的，用于收集Mapper或者Reducer輸出數(shù)據(jù)的通用機制。

4. MapReduce的輸入/輸出格式

MapReduce計算框架本質(zhì)上是一種基于磁盤的批處理并行計算系統(tǒng)，每一輪MapReduce作業(yè)都需要從分布式文件系統(tǒng)中讀取數(shù)據(jù)，處理之后再寫入分布式文件系統(tǒng)。其他涉及到很多I/O操作，這些操作包括內(nèi)存到磁盤、磁盤到內(nèi)存，以及節(jié)點之間的數(shù)據(jù)交換。

4.1 輸入格式

檢查作業(yè)輸入的有效性。

把輸入文件切分成多個邏輯InputSplit實例，并把每個實例分發(fā)給一個Mapper（一對一）；FileSplit是默認的InputSplit，通過write（DataOutput out）和readFields（DataInput in）兩種方法進行序列化和反序列化。

提供RecordReader實現(xiàn)。這個RecordReader從邏輯InputSplit中獲取輸入記錄，這些記錄將由Mapper處理，Mapper利用該實現(xiàn)從InputSplit中讀取<k,v>鍵值對。

TextInputFormat

TextInputFormat 用于讀取純文本文件。

KeyValueTextInputFormat

KeyValueTextInputFormat同樣是用于讀取文本文件。

NLineInputForamt

NLineInputForamt可以將文件以行為單位進行split切分，比如文件中的每一行對應的一個Map。

SequenceFileInputFormat

SequenceFileInputFormat用于讀取SequenceFile。

4.2 輸出格式

Hadoop中的OutputFormat用來描述MapReduce作業(yè)的輸出格式：

檢驗作業(yè)的輸出。

驗證輸出結(jié)果類型是否如在Config中所配置的。默認的OutputFormat是TextOutputFormat

提供一個RecordWriter的實現(xiàn)，用來輸出作業(yè)結(jié)果。RecordWriter生成的<key,value>鍵值對輸出到文件。

TextOutputFormat

TextOutputFormat是Hadoop默認的輸出格式。

SequenceFileOutputFormat

SequenceFileOutputFormat用于就是輸出到Hadoop中的SequenceFile文件格式。

MapFileOutputFormat

指定MapFileOutputFormat輸出類型可以將數(shù)據(jù)輸出為Hadoop中的MapFile文件格式。

MultipleOutputFormat

MultipleOutputFormat是Hadoop中的多路輸出處理類，通過這個類可以實現(xiàn)根據(jù)key將記錄控制輸出到不同的文件。

5. 核心問題

5.1 Map和Reduce數(shù)量問題

MapTask數(shù)量

Max.split(100M) Min.split(10M) Block(64M)InputSize=Max(min.split,min(max.split,block))

• Max.split指的是最大InputSplit文件大小
• Min.split指的是最小InputSplit文件大小
• Block指的是Block文件大小

其中InputSize的大小是InputSize=Max(min.split,min(max.split,block))

ReduceTask數(shù)量

job.setNumReduceTasks(numReduceTask);

通過job設置ReduceTask數(shù)量個數(shù)。

單個Reduce：

多個Reduce

數(shù)量為0(適應于不需要歸約和處理的作業(yè))

5.2 作業(yè)配置

• 作業(yè)配置的相關(guān)設置方法

作業(yè)配置方法功能說明

setNumReduceTasks	設置reduce數(shù)目
setNumMapTasks	設置Map數(shù)目
setInputFormatClass	設置輸入文件格式類
setOutputFormatClass	設置輸出文件格式類
setMapperClass	輸出Map類
setCombiner	設置Combiner類
setReducerClass	設置Reduce類
setPartitionerClass	設置Partitioner類
setMapOutputKeyClass	設置Map輸出的Key類
setMapOutputValueClass	設置Map輸出的Value類
setCompressMapOutput	設置Map輸出是否壓縮
setOutputValueClass	設置輸出value類
setJobName	設置作業(yè)名字
setSpeculativeExecution	設置是否開啟預防性執(zhí)行
setMapSpeculativeExecution	設置是否開啟Map任務的預防性執(zhí)行
setReduceSpeculativeExecution	設置是否開啟Reduce任務的預防性執(zhí)行

5.3 作業(yè)的容錯機制

MapReduce作為一個通用的并行計算框架，有著非常健壯的容錯機制，在不同的粒度上均有考慮。

再執(zhí)行

用戶的一個MapReduce作業(yè)往往是由很多任務組成的，只有所有的任務執(zhí)行處理完畢之后才算整個作業(yè)成功。對于任務的容錯機制，MapReduce采用的最簡單的方法進行處理，即“再執(zhí)行”，也就是對于失敗的任務重新調(diào)度執(zhí)行一次。一般有以下兩種情況需要再執(zhí)行：

如果是Map任務或者Reduce任務失敗了，那么調(diào)度器就會將這個失敗的任務分配到其他節(jié)點重新執(zhí)行。

如果是一個節(jié)點死掉了，那么在這臺死機的節(jié)點上已經(jīng)完成了運行的map
任務以及正在運行中的Map和Reduce任務都將調(diào)度重新執(zhí)行，同時在其他機器上正在運行的Reduce任務也將被重新執(zhí)行。

推測式執(zhí)行

在mapreduce中，影響一個作業(yè)的總執(zhí)行時間最通常的因素是“落伍者”：在運算過程中，如果有一臺機器花了很長時間才完成最后幾個Map或者Reduce任務，導致mapReduce任務執(zhí)行時間超過預期。出現(xiàn)“落伍者”的原因很多，CPU、內(nèi)存、本地磁盤和網(wǎng)絡帶寬等因素都會導致某些Map或者reduce任務執(zhí)行效率更加緩慢。

所謂推測式執(zhí)行策略就是MapReduce對每一個任務都計算它的進度，如果一個任務的進度遠遠慢于其他的任務時，那么這個任務便可以被認為是一個“落伍者”。在發(fā)現(xiàn)一個“落伍者”之后，調(diào)度器就會在其他節(jié)點上重新調(diào)度這任務以便重新執(zhí)行。在這個時候，一般會有兩個相同的任務在同時執(zhí)行，最終先完成的那個任務就算成功了，而沒有完成的那個任務就會被殺死。

5.4 作業(yè)調(diào)度

調(diào)度的功能是將各種類型的作業(yè)在調(diào)度算法作用下分配給Hadoop集群中的計算節(jié)點，從而達到 分布式和并行計算 的目的。
調(diào)度算法模塊中至少涉及兩個重要流程：1.作業(yè)的選擇 2.任務的分配。

調(diào)度過程 ：

1）MapReduce框架中作業(yè)通常是通過JobClient.runJob(job)方法提交到JobTracker，JobTracker接收到JobClient的請求后將其加入作業(yè)調(diào)度隊列中。

2）然后JobTracker一直等待JobClient通過RPC向其提交作業(yè)，而TaskTracker則一直通過RPC向JobTracker發(fā)送心跳信號詢問是否有任務可執(zhí)行，有則請求JobTracker派發(fā)任務給它執(zhí)行。

3）如果JobTracker的作業(yè)隊列不為空，則TaskTracker發(fā)送的心跳將會獲得JobTracker向它派發(fā)的任務。
這是一個主動請求的任務：slave的TaskTracker主動向master的JobTracker請求任務。

4）當TaskTracker接到任務后，通過自身調(diào)度在本slave建立起Task，執(zhí)行任務。

常用調(diào)度器 主要包括：JobQueueTaskScheduler(FIFO調(diào)度器)，CapacityScheduler(容量調(diào)度器)，Fair Scheduler(公平調(diào)度器)等。

• FIFO調(diào)度器：基本思想是作業(yè)按照先后順序統(tǒng)一放入一個隊列中，然后根據(jù)優(yōu)先級按照時間先后順序依次執(zhí)行，總體遵循先進先出的基本調(diào)度策略
• 容量調(diào)度器：計算能力調(diào)度器，是雅虎結(jié)合自己的集群業(yè)務類型提出的一種調(diào)度策略。這種調(diào)度策略支持多種隊列，每個隊列可以單獨配置一定的資源量，每個隊列采取FIFO策略
• 公平調(diào)度器：FaceBook開發(fā)的貢獻給開源社區(qū)的，可以是多種作業(yè)并行執(zhí)行并共享資源池。公平調(diào)度器的目的就是為了保證在多用戶、多作業(yè)類型的情況下保證整個集群的資源利用率，同時可以讓所有用戶公平地共享整個集群資源。

6. MapReduce特性

6.1 計數(shù)器（Counters）

MapReduce Counter可以為我們提供一個觀察MapReduce Job運行期中的各個細節(jié)數(shù)據(jù)視圖。通過這些Counter計數(shù)器我們可以從全局視角來審查程序的運行情況，以及做出錯誤診斷進行相應處理。

6.2 DistributedCache

DistributedCache是MapReduce計算框架提供的功能，能夠緩存應用程序所需要的文件（包括文本、檔案文件、jar文件等）。可以將具體應用相關(guān)的、大尺寸的、只讀的文件有效地分發(fā)到各個計算機中，應用程序只需要在JobConf中通過url（hdfs://）指定需要緩存的文件。

MapReduce框架在作業(yè)的所有任務執(zhí)行之前會把必要的文件復制到slave節(jié)點上。它運行高效的原因是因為每個作業(yè)的文件只復制一次并且為那些沒有文檔的slave文件緩存文檔。

6.3 Tool

6.4 Profiling

6.5 數(shù)據(jù)壓縮

總結(jié)

以上是生活随笔為你收集整理的Hadoop详解（六）：MapReduce计算框架详解的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。