MapReduce工作原理

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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