MapReduce工作原理

1、MapReduce工作原理1 MapReduce原理（一）1.1 MapReduce編程模型MapReduce采用"分而治之"的思想，把對大規(guī)模數(shù)據(jù)集的操作，分發(fā)給一個主節(jié)點管理下的各個分節(jié)點共同完成，然后通過整合各個節(jié)點的中間結(jié)果，得到最終結(jié)果。簡單地說，MapReduce就是"任務(wù)的分解與結(jié)果的匯總"。在Hadoop中，用于執(zhí)行MapReduce任務(wù)的機器角色有兩個：一個是JobTracker；另一個是TaskTracker，JobTracker是用于調(diào)度工作的，TaskTracker是用于執(zhí)行工作的。一個Hadoop集群中只有一臺JobTracke

2、r。在分布式計算中，MapReduce框架負責(zé)處理了并行編程中分布式存儲、工作調(diào)度、負載均衡、容錯均衡、容錯處理以及網(wǎng)絡(luò)通信等復(fù)雜問題，把處理過程高度抽象為兩個函數(shù)：map和reduce，map負責(zé)把任務(wù)分解成多個任務(wù)，reduce負責(zé)把分解后多任務(wù)處理的結(jié)果匯總起來。需要注意的是，用MapReduce來處理的數(shù)據(jù)集（或任務(wù)）必須具備這樣的特點：待處理的數(shù)據(jù)集可以分解成許多小的數(shù)據(jù)集，而且每一個小數(shù)據(jù)集都可以完全并行地進行處理。1.2 MapReduce處理過程在Hadoop中，每個MapReduce任務(wù)都被初始化為一個Job，每個Job又可以分為兩種階段：map階段和reduce階段。這兩個

3、階段分別用兩個函數(shù)表示，即map函數(shù)和reduce函數(shù)。map函數(shù)接收一個<key,value>形式的輸入，然后同樣產(chǎn)生一個<key,value>形式的中間輸出，Hadoop函數(shù)接收一個如<key,(list of values)>形式的輸入，然后對這個value集合進行處理，每個reduce產(chǎn)生0或1個輸出，reduce的輸出也是<key,value>形式的。一切都是從最上方的user program開始的，user program鏈接了MapReduce庫，實現(xiàn)了最基本的Map函數(shù)和Reduce函數(shù)。圖中執(zhí)行的順序都用數(shù)字標記了。1) Map

4、Reduce庫先把user program的輸入文件劃分為M份（M為用戶定義），每一份通常有16MB到64MB，如圖左方所示分成了split04；然后使用fork將用戶進程拷貝到集群內(nèi)其它機器上。2) user program的副本中有一個稱為master，其余稱為worker，master是負責(zé)調(diào)度的，為空閑worker分配作業(yè)（Map作業(yè)或者Reduce作業(yè)），worker的數(shù)量也是可以由用戶指定的。3) 被分配了Map作業(yè)的worker，開始讀取對應(yīng)分片的輸入數(shù)據(jù)，Map作業(yè)數(shù)量是由M決定的，和split一一對應(yīng)；Map作業(yè)從輸入數(shù)據(jù)中抽取出鍵值對，每一個鍵值對都作為參數(shù)傳遞給map函數(shù)

5、，map函數(shù)產(chǎn)生的中間鍵值對被緩存在內(nèi)存中。4) 緩存的中間鍵值對會被定期寫入本地磁盤，而且被分為R個區(qū)，R的大小是由用戶定義的，將來每個區(qū)會對應(yīng)一個Reduce作業(yè)；這些中間鍵值對的位置會被通報給master，master負責(zé)將信息轉(zhuǎn)發(fā)給Reduce worker。5) master通知分配了Reduce作業(yè)的worker它負責(zé)的分區(qū)在什么位置（肯定不止一個地方，每個Map作業(yè)產(chǎn)生的中間鍵值對都可能映射到所有R個不同分區(qū)），當(dāng)Reduce worker把所有它負責(zé)的中間鍵值對都讀過來后，先對它們進行排序，使得相同鍵的鍵值對聚集在一起。因為不同的鍵可能會映射到同一個分區(qū)也就是同一個Reduce

6、作業(yè)（誰讓分區(qū)少呢），所以排序是必須的。6) reduce worker遍歷排序后的中間鍵值對，對于每個唯一的鍵，都將鍵與關(guān)聯(lián)的值傳遞給reduce函數(shù)，reduce函數(shù)產(chǎn)生的輸出會添加到這個分區(qū)的輸出文件中。7) 當(dāng)所有的Map和Reduce作業(yè)都完成了，master喚醒正版的user program，MapReduce函數(shù)調(diào)用返回user program的代碼。所有執(zhí)行完畢后，MapReduce輸出放在了R個分區(qū)的輸出文件中（分別對應(yīng)一個Reduce作業(yè)）。用戶通常并不需要合并這R個文件，而是將其作為輸入交給另一個MapReduce程序處理。整個過程中，輸入數(shù)據(jù)是來自底層分布式文件系統(tǒng)（G

7、FS）的，中間數(shù)據(jù)是放在本地文件系統(tǒng)的，最終輸出數(shù)據(jù)是寫入底層分布式文件系統(tǒng)（GFS）的。而且我們要注意Map/Reduce作業(yè)和map/reduce函數(shù)的區(qū)別：Map作業(yè)處理一個輸入數(shù)據(jù)的分片，可能需要調(diào)用多次map函數(shù)來處理每個輸入鍵值對；Reduce作業(yè)處理一個分區(qū)的中間鍵值對，期間要對每個不同的鍵調(diào)用一次reduce函數(shù)，Reduce作業(yè)最終也對應(yīng)一個輸出文件。2 MapReduce原理（二）2.1 MapReduce作業(yè)運行流程流程分析：1) 在客戶端啟動一個作業(yè)。2) 向JobTracker請求一個Job ID。3) 將運行作業(yè)所需要的資源文件復(fù)制到HDFS上，包括MapReduc

8、e程序打包的JAR文件、配置文件和客戶端計算所得的輸入劃分信息。這些文件都存放在JobTracker專門為該作業(yè)創(chuàng)建的文件夾中。文件夾名為該作業(yè)的Job ID。JAR文件默認會有10個副本（屬性控制）；輸入劃分信息告訴了JobTracker應(yīng)該為這個作業(yè)啟動多少個map任務(wù)等信息。4) JobTracker接收到作業(yè)后，將其放在一個作業(yè)隊列里，等待作業(yè)調(diào)度器對其進行調(diào)度，當(dāng)作業(yè)調(diào)度器根據(jù)自己的調(diào)度算法調(diào)度到該作業(yè)時，會根據(jù)輸入劃分信息為每個劃分創(chuàng)建一個map任務(wù)，并將map任務(wù)分配給TaskTracker執(zhí)行。對于map和reduce任務(wù)，TaskTracker根據(jù)主機核的數(shù)量和內(nèi)存的大小有固

9、定數(shù)量的map槽和reduce槽。這里需要強調(diào)的是：map任務(wù)不是隨隨便便地分配給某個TaskTracker的，這里有個概念叫：數(shù)據(jù)本地化（Data-Local）。意思是：將map任務(wù)分配給含有該map處理的數(shù)據(jù)塊的TaskTracker上，同時將程序JAR包復(fù)制到該TaskTracker上來運行，這叫“運算移動，數(shù)據(jù)不移動”。而分配reduce任務(wù)時并不考慮數(shù)據(jù)本地化。5) TaskTracker每隔一段時間會給JobTracker發(fā)送一個心跳，告訴JobTracker它依然在運行，同時心跳中還攜帶著很多的信息，比如當(dāng)前map任務(wù)完成的進度等信息。當(dāng)JobTracker收到作業(yè)的最后一個任務(wù)

10、完成信息時，便把該作業(yè)設(shè)置成“成功”。當(dāng)JobClient查詢狀態(tài)時，它將得知任務(wù)已完成，便顯示一條消息給用戶。以上是在客戶端、JobTracker、TaskTracker的層次來分析MapReduce的工作原理的，下面我們再細致一點，從map任務(wù)和reduce任務(wù)的層次來分析分析吧。2.2 Map、Reduce任務(wù)中Shuffle和排序的過程流程分析： Map端： 1) 每個輸入分片會讓一個map任務(wù)來處理，默認情況下，以HDFS的一個塊的大?。J為64M）為一個分片，當(dāng)然我們也可以設(shè)置塊的大小。map輸出的結(jié)果會暫且放在一個環(huán)形內(nèi)存緩沖區(qū)中（該緩沖區(qū)的大小默認為100M，由屬性控制），當(dāng)

11、該緩沖區(qū)快要溢出時（默認為緩沖區(qū)大小的80%，由屬性控制），會在本地文件系統(tǒng)中創(chuàng)建一個溢出文件，將該緩沖區(qū)中的數(shù)據(jù)寫入這個文件。2) 在寫入磁盤之前，線程首先根據(jù)reduce任務(wù)的數(shù)目將數(shù)據(jù)劃分為相同數(shù)目的分區(qū)，也就是一個reduce任務(wù)對應(yīng)一個分區(qū)的數(shù)據(jù)。這樣做是為了避免有些reduce任務(wù)分配到大量數(shù)據(jù)，而有些reduce任務(wù)卻分到很少數(shù)據(jù)，甚至沒有分到數(shù)據(jù)的尷尬局面。其實分區(qū)就是對數(shù)據(jù)進行hash的過程。然后對每個分區(qū)中的數(shù)據(jù)進行排序，如果此時設(shè)置了Combiner，將排序后的結(jié)果進行Combia操作，這樣做的目的是讓盡可能少的數(shù)據(jù)寫入到磁盤。3) 當(dāng)map任務(wù)輸出最后一個記錄時，可能會

12、有很多的溢出文件，這時需要將這些文件合并。合并的過程中會不斷地進行排序和combia操作，目的有兩個：（1）盡量減少每次寫入磁盤的數(shù)據(jù)量；（2）盡量減少下一復(fù)制階段網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)量。最后合并成了一個已分區(qū)且已排序的文件。為了減少網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)量，這里可以將數(shù)據(jù)壓縮，只要將設(shè)置為true就可以了。4) 將分區(qū)中的數(shù)據(jù)拷貝給相對應(yīng)的reduce任務(wù)。有人可能會問：分區(qū)中的數(shù)據(jù)怎么知道它對應(yīng)的reduce是哪個呢？其實map任務(wù)一直和其父TaskTracker保持聯(lián)系，而TaskTracker又一直和JobTracker保持心跳。所以JobTracker中保存了整個集群中的宏觀信息。只要reduce

13、任務(wù)向JobTracker獲取對應(yīng)的map輸出位置就ok了哦。到這里，map端就分析完了。那到底什么是Shuffle呢？Shuffle的中文意思是“洗牌”，如果我們這樣看：一個map產(chǎn)生的數(shù)據(jù)，結(jié)果通過hash過程分區(qū)卻分配給了不同的reduce任務(wù)，是不是一個對數(shù)據(jù)洗牌的過程呢？Reduce端： 1) Reduce會接收到不同map任務(wù)傳來的數(shù)據(jù)，并且每個map傳來的數(shù)據(jù)都是有序的。如果reduce端接受的數(shù)據(jù)量相當(dāng)小，則直接存儲在內(nèi)存中（緩沖區(qū)大小由屬性控制，表示用作此用途的堆空間的百分比），如果數(shù)據(jù)量超過了該緩沖區(qū)大小的一定比例（由決定），則對數(shù)據(jù)合并后溢寫到磁盤中。2) 隨著溢寫文件的

14、增多，后臺線程會將它們合并成一個更大的有序的文件，這樣做是為了給后面的合并節(jié)省時間。其實不管在map端還是reduce端，MapReduce都是反復(fù)地執(zhí)行排序，合并操作，現(xiàn)在終于明白了有些人為什么會說：排序是hadoop的靈魂。3) 合并的過程中會產(chǎn)生許多的中間文件（寫入磁盤了），但MapReduce會讓寫入磁盤的數(shù)據(jù)盡可能地少，并且最后一次合并的結(jié)果并沒有寫入磁盤，而是直接輸入到reduce函數(shù)。3 MapReduce原理（三）3.1 物理實體談mapreduce運行機制，可以從很多不同的角度來描述，比如說從mapreduce運行流程來講解，也可以從計算模型的邏輯流程來進行講解，也許有些深入

15、理解了mapreduce運行機制還會從更好的角度來描述，但是將mapreduce運行機制有些東西是避免不了的，就是一個個參入的實例對象，一個就是計算模型的邏輯定義階段。首先講講物理實體，參入mapreduce作業(yè)執(zhí)行涉及4個獨立的實體：1) 客戶端（client）：編寫mapreduce程序，配置作業(yè)，提交作業(yè)，這就是程序員完成的工作；2) JobTracker：初始化作業(yè)，分配作業(yè)，與TaskTracker通信，協(xié)調(diào)整個作業(yè)的執(zhí)行；3) TaskTracker：保持與JobTracker的通信，在分配的數(shù)據(jù)片段上執(zhí)行Map或Reduce任務(wù)，TaskTracker和JobTracker的不同

16、有個很重要的方面，就是在執(zhí)行任務(wù)時候TaskTracker可以有n多個，JobTracker則只會有一個（JobTracker只能有一個就和hdfs里namenode一樣存在單點故障，我會在后面的mapreduce的相關(guān)問題里講到這個問題的）4) Hdfs：保存作業(yè)的數(shù)據(jù)、配置信息等等，最后的結(jié)果也是保存在hdfs上面。3.2 運行原理首先是客戶端要編寫好mapreduce程序，配置好mapreduce的作業(yè)也就是job，接下來就是提交job了，提交job是提交到JobTracker上的，這個時候JobTracker就會構(gòu)建這個job，具體就是分配一個新的job任務(wù)的ID值，接下來它會做檢查操

17、作，這個檢查就是確定輸出目錄是否存在，如果存在那么job就不能正常運行下去，JobTracker會拋出錯誤給客戶端，接下來還要檢查輸入目錄是否存在，如果不存在同樣拋出錯誤，如果存在JobTracker會根據(jù)輸入計算輸入分片（Input Split），如果分片計算不出來也會拋出錯誤，至于輸入分片我后面會做講解的，這些都做好了JobTracker就會配置Job需要的資源了。分配好資源后，JobTracker就會初始化作業(yè)，初始化主要做的是將Job放入一個內(nèi)部的隊列，讓配置好的作業(yè)調(diào)度器能調(diào)度到這個作業(yè)，作業(yè)調(diào)度器會初始化這個job，初始化就是創(chuàng)建一個正在運行的job對象（封裝任務(wù)和記錄信息），以便

18、JobTracker跟蹤job的狀態(tài)和進程。初始化完畢后，作業(yè)調(diào)度器會獲取輸入分片信息（input split），每個分片創(chuàng)建一個map任務(wù)。接下來就是任務(wù)分配了，這個時候tasktracker會運行一個簡單的循環(huán)機制定期發(fā)送心跳給jobtracker，心跳間隔是5秒，程序員可以配置這個時間，心跳就是jobtracker和tasktracker溝通的橋梁，通過心跳，jobtracker可以監(jiān)控tasktracker是否存活，也可以獲取tasktracker處理的狀態(tài)和問題，同時tasktracker也可以通過心跳里的返回值獲取jobtracker給它的操作指令。任務(wù)分配好后就是執(zhí)行任務(wù)了。在任

19、務(wù)執(zhí)行時候jobtracker可以通過心跳機制監(jiān)控tasktracker的狀態(tài)和進度，同時也能計算出整個job的狀態(tài)和進度，而tasktracker也可以本地監(jiān)控自己的狀態(tài)和進度。當(dāng)jobtracker獲得了最后一個完成指定任務(wù)的tasktracker操作成功的通知時候，jobtracker會把整個job狀態(tài)置為成功，然后當(dāng)客戶端查詢job運行狀態(tài)時候（注意：這個是異步操作），客戶端會查到j(luò)ob完成的通知的。如果job中途失敗，mapreduce也會有相應(yīng)機制處理，一般而言如果不是程序員程序本身有bug，mapreduce錯誤處理機制都能保證提交的job能正常完成。下面我從邏輯實體的角度講解m

20、apreduce運行機制，這些按照時間順序包括：輸入分片（input split）、map階段、combiner階段、shuffle階段和reduce階段。1) 輸入分片（input split）：在進行map計算之前，mapreduce會根據(jù)輸入文件計算輸入分片（input split），每個輸入分片（input split）針對一個map任務(wù)，輸入分片（input split）存儲的并非數(shù)據(jù)本身，而是一個分片長度和一個記錄數(shù)據(jù)的位置的數(shù)組，輸入分片（input split）往往和hdfs的block（塊）關(guān)系很密切，假如我們設(shè)定hdfs的塊的大小是64mb，如果我們輸入有三個文件，大小分別

21、是3mb、65mb和127mb，那么mapreduce會把3mb文件分為一個輸入分片（input split），65mb則是兩個輸入分片（input split）而127mb也是兩個輸入分片（input split），換句話說我們?nèi)绻趍ap計算前做輸入分片調(diào)整，例如合并小文件，那么就會有5個map任務(wù)將執(zhí)行，而且每個map執(zhí)行的數(shù)據(jù)大小不均，這個也是mapreduce優(yōu)化計算的一個關(guān)鍵點。2) map階段：就是程序員編寫好的map函數(shù)了，因此map函數(shù)效率相對好控制，而且一般map操作都是本地化操作也就是在數(shù)據(jù)存儲節(jié)點上進行；3) combiner階段：combiner階段是程序員可以選擇的

22、，combiner其實也是一種reduce操作，因此我們看見WordCount類里是用reduce進行加載的。Combiner是一個本地化的reduce操作，它是map運算的后續(xù)操作，主要是在map計算出中間文件前做一個簡單的合并重復(fù)key值的操作，例如我們對文件里的單詞頻率做統(tǒng)計，map計算時候如果碰到一個hadoop的單詞就會記錄為1，但是這篇文章里hadoop可能會出現(xiàn)n多次，那么map輸出文件冗余就會很多，因此在reduce計算前對相同的key做一個合并操作，那么文件會變小，這樣就提高了寬帶的傳輸效率，畢竟hadoop計算力寬帶資源往往是計算的瓶頸也是最為寶貴的資源，但是combine

23、r操作是有風(fēng)險的，使用它的原則是combiner的輸入不會影響到reduce計算的最終輸入，例如：如果計算只是求總數(shù)，最大值，最小值可以使用combiner，但是做平均值計算使用combiner的話，最終的reduce計算結(jié)果就會出錯。4) shuffle階段：將map的輸出作為reduce的輸入的過程就是shuffle了，這個是mapreduce優(yōu)化的重點地方。這里我不講怎么優(yōu)化shuffle階段，講講shuffle階段的原理，因為大部分的書籍里都沒講清楚shuffle階段。Shuffle一開始就是map階段做輸出操作，一般mapreduce計算的都是海量數(shù)據(jù)，map輸出時候不可能把所有文件都放到內(nèi)存操作，因此map寫入磁盤的過程十分的復(fù)雜，更何況map輸出時候要對結(jié)果進行排序，內(nèi)存開銷是很大的，map在做輸出時候會在內(nèi)存里開啟一個環(huán)形內(nèi)存緩沖區(qū)，這個緩沖區(qū)專門用來輸出的，默認大小是100mb，并且在配置文件里為這個緩沖區(qū)設(shè)定了一個閥值，默認是0.80（這個大小和閥值都是可以在配置文件里進行配置的），同時map還會為輸出操作啟動一個守護線程，如果緩沖區(qū)的內(nèi)存達到了閥值的80%時候，這個守護線程就會把內(nèi)容寫到磁盤上，這個過程叫spill，另外的20%內(nèi)存可以繼續(xù)寫入要寫進磁盤的數(shù)據(jù)，寫入磁盤和寫入內(nèi)存操作是互不干擾的，如