一、MapReduce完整運行流程
解析：
1 在客戶端啟動一個作業。
2 向JobTracker請求一個Job ID。
3 將運行作業所需要的資源文件復制到HDFS上，包括MapReduce程序打包的jar文件、配置文件和客戶端計算所得的計算划分信息。這些文件都存放在JobTracker專門為該作業創建的文件夾中。文件夾名為該作業的Job ID。jar文件默認會有10個副本（mapred.submit.replication屬性控制）；輸入划分信息告訴了JobTracker應該為這個作業啟動多少個map任務等信息。
4 JobTracker接收到作業后，將其放在一個作業隊列里，等待作業調度器對其進行調度（這里是不是很像微機中的進程調度呢），當作業調度器根據自己的調度算法調度到該作業時，會根據輸入划分信息為每個划分創建一個map任務，並將map任務分配給TaskTracker執行。對於map和reduce任務，TaskTracker根據主機核的數量和內存的大小有固定數量的map槽和reduce槽。這里需強調的是：map任務不是隨隨便便地分配給某個TaskTracker的，這里有個概念叫：數據本地化（Data-Local）。意思是：將map任務分配給含有該map處理的數據塊的TaskTracker上，同事將程序jar包復制到該TaskTracker上來運行，這叫“運算移動，數據不移動”。而分配reduce任務時並不考慮數據本地化。
5 TaskTracker每隔一段時間會給JobTracker發送一個心跳，告訴JobTracker它依然在運行，同時心跳中還攜帶者很多信息，比如當前map任務完成的進度等信息。當JobTracker收到作業的最后一個任務完成信息時，便把該作業設置成“成功”。當JobTracker查詢狀態時，它將得知任務已完成，便顯示一條消息給用戶。
二、MapReduce任務的Shuffle和排序過程
Map端流程分析
1 每個輸入分片會讓一個map任務來處理，默認情況下，以HDFS的一個塊的大小（默認64M）為一個分片，當然我們也可以設置塊的大小。map輸出的結果會暫且放在一個環形內存緩沖區中（該緩沖區的大小默認為100M，由io.sort.mb屬性控制），當該緩沖區快要溢出時（默認為緩沖區大小的80%，由io.sort.spill.percent屬性控制），會在本地文件系統中創建一個溢出文件，將該緩沖區中的數據寫入這個文件。
2 在寫入磁盤之前，線程首先根據reduce任務的數目將數據划分為相同數目的分區，也就是一個reduce任務對應一個分區的數據。這樣做是為了避免有些reduce任務分配到大量數據，而有些reduce任務卻分到很少數據，甚至沒有分到數據的尷尬局面。其實分區就是對數據進行hash的過程。然后對每個分區中的數據進行排序，如果此時設置了Combiner，將排序后的結果進行Combianer操作，這樣做的目的是讓盡可能少的數據寫入到磁盤。
3 當map任務輸出最后一個記錄時，可能會有很多的溢出文件，這時需要將這些文件合並。合並的過程中會不斷地進行排序和combiner操作，目的有兩個：1、盡量減少每次寫入磁盤的數據量；2、盡量減少下一復制階段網絡傳輸的數據量。最后合並成了一個已分區且已排序的文件。為了減少網絡傳輸的數據量，這里可以將數據壓縮，只要將mapred.compress.map.out設置為true就可以。
數據壓縮：Gzip、Lzo、snappy。
4 將分區中的數據拷貝給相對應的reduce任務。有人可能會問：分區中的數據怎么知道它對應的reduce是哪個呢？其實map任務一直和其父TaskTracker保持聯系，而TaskTracker又一直和obTracker保持心跳。所以JobTracker中保存了整個集群中的宏觀信息。只要reduce任務向JobTracker獲取對應的map輸出位置就OK了。
Shuffle分析
Shuffle的中文意思是“洗牌”，如果我們這樣看：一個map產生的數據，結果通過hash過程分區缺分配給了不同的reduce任務，是不是一個對數據洗牌的過程呢？
shuffle的概念：
Collections.shuffle(List list)：隨機打亂list里的元素順序。
MapReduce里的Shuffle：描述着數據從map task輸出到reduce task輸入的這段過程。
Map端shuffle的過程：
1 每個map task都有一個內存緩沖區，存儲着map的輸出結果，當緩沖區快滿的時候需要將緩沖區的數據以一個臨時文件的方式存放到磁盤，當整個map task結束后在對磁盤中這個map task產生的所有臨時文件做一個合並，生成最終的正式輸出文件，然后等待reduce task來拉數據。
2 在map task執行時，它的輸入數據來源於HDFS的block，當然在MapReduce概念中，map task只讀取split。split與block對應關系可能是多對一，默認是一對一。在wordcount例子里，假設map的輸入數據都是是像“aaa”這樣的字符串。
3 在經過mapper的運行后，我們得知mapper的輸出是這樣一個key/value對：key是“aaa”，value是數值1。因為當前map端只做加1的操作，在reduce task里采取合並結果集。前面我們知道這個job有3個reduce task。那到底當前的“aaa”究竟該丟給哪個reduce去處理呢？是需要現在做決定的。
4 MapReduce提供Partitioner接口，作用就是根據key或value及reduce的數量來決定當前的輸出數據最終應該交由哪個reduce task處理。默認對key hash后再以reduce task數據取模。默認的取模方式只是為了平均reduce的處理能力，如果用戶自己對Partitioner有需求，可以定制並設置到job上。
5 在例子中，“aaa”經過Partition后返回0，也就是這對值應當交由第一個reduce來處理。接下來，需要將數據寫入內存緩沖區中，緩沖區的作用是批量收集map結果，減少磁盤IO的影響。我們的key/value對以及Partition的結果都會被寫入緩沖區。當然，寫入之前，key與value值都會被序列化成字節數組。
6 內存緩沖區是有大小限制的，默認是100MB。當map task的輸出結果很多時，就可能會撐爆內存，所以需要在一定條件下將緩沖區中的數據臨時寫入磁盤，然后重新利用這塊緩沖區。這個從內存往磁盤寫數據的過程被稱為spill，中文可理解為溢寫。溢寫是由單獨線程來完成，不影響往緩沖區寫map結果的線程。溢寫線程啟動時不應該阻止map的結果輸出，所以整個緩沖區有個溢寫的比例spill.percent。比例默認是0.8，也就是當緩沖區的數據值已經達到閾值（buffer size * spill percent = 100MB * 0.8 = 80MB），溢寫線程啟動，鎖定這80MB的內存，執行溢寫過程。map task的輸出結果還可以往剩下的20MB內存中寫，互不影響。
7 當溢寫線程啟動后，需要對這80MB空間內的key做排序（sort）。排序是MapReduce模型默認的行為，這里的排序也是對序列化的字節做的排序。
8 因為map task的輸出是需要發送到不同的reduce端去，而內存緩沖區沒有對將發送到相同reduce端的數據做合並，那么這種合並應該是體現在磁盤文件中的。從官方圖上也可以看到寫到磁盤中的一些文件是對不同的reduce端的數值做過合並。所以溢寫過程一個很重要的細節在於，如果有很多個key/value對需要發送到某個reduce端去，那么需要將這些key/value值拼接到一塊，減少與partition相關的索引記錄。
在針對每個reduce端而合並數據時，有些數據可能像這樣：“aaa”/1，“aaa”/1。對於wordcount例子，只是簡單地統計單詞出現的次數，如果在同一個map task的結果中有很多像“aaa”一樣出現多次的key，我們就應該把它們的值合並到一塊，這個過程叫reduce也叫combine。但MapReduce的術語中，reduce只值reduce端執行從多個map task取數據做計算的過程。除reduce外，非正式地合並數據只能算作combine了。其實大家知道的，MapReduce中將Combiner等同於Reducer。
如果client設置過Combiner，那么現在就是使用Combiner的時候了。將有相同key的key/value對的value加起來，減少溢寫到磁盤的數據量。Combiner會優化MapReduce的中間結果，所以它在整個模型中會多次使用。那哪些場景才能使用Combiner呢？從這里分析，Combiner的輸出是Reducer的輸入，Combiner絕不能改變最終的計算結果。所以從我的想法來看，Combiner只應該用於那種Reduce的輸入key/value與輸出key/value類型完全一致，且不影響最終結果的場景。比如累加，最大值等。Combiner的使用一定得慎重，如果用好，它對job執行效率有幫助，反之會影響reduce的最終結果。
9 每次溢寫會在磁盤上生成一個溢寫文件，如果map的輸出結果真的很大，有多次這樣的溢寫發生，磁盤上相應的就會有多個溢寫文件存在。當map task真正完成時，內存緩沖區中的數據也全部溢寫到磁盤中形成一個溢寫文件。最終磁盤中會至少有一個這樣的溢寫文件存在（如果map的輸出結果很少，當map執行完成時，只會產生一個溢寫文件），因為最終的文件只有一個，所以需要將這些溢寫文件歸並到一起，這個過程就叫Merge。Merge是怎樣的？如前面的例子，“aaa”從某個map task讀取過來時值是5，從另外一個map讀取時值是8，因為他們有相同的key，所以要merge成group。
什么是group：對於“aaa”就是像真陽的：{“aaa”，[5,8,2,...]}，數組中的值就是從不同的溢寫文件中讀取出來的，然后再把這些值加起來。請注意，因為merge是將多個溢寫文件合並到一個文件，所以可能也有相同的key存在，在這個過程中，如果client設置過Combiner，也會使用Combiner來合並相同的key。
至此，map端的所有工作都已經結束，最終生成的這個文件也存放在TaskTracker夠得到的某個本地目錄中。每個reduce task不斷地通過RPC從JobTRacker那獲取map task是否完成的信息，如果reduce task得到通知，獲知某台TaskTracker上的map task執行完成，Shuffle的后半段過程開始啟動。
Reduce端的shuffle過程：
1 copy過程，簡單地拉取數據。Reduce進程啟動一些數據copy線程（Fetcher），通過http方式請求map task所在的TaskTracker獲取map task的輸出文件。因為map task早已結束，這些文件就歸TaskTracker管理在本地磁盤中。
2 Merge階段。這里的merge和map端的merge動作相同，只是數組中存放的是不同map端copy來的數值。copy過來的數據會先放入內存緩沖區中，這里的緩沖區大小要比map端更為靈活，它基於JVM的heap size設置，因為Shuffle階段Reducer不運行，所以應該把絕大部分的內存都給Shuffle使用。
3 Merge有三種形式：1、內存到內存；2、內存到磁盤；3、磁盤到磁盤。默認情況下第一種形式不啟用，讓人比較困惑。當內存中的數據量到達一定閾值，就啟動內存到磁盤的merge。與map端類似，這也是溢寫的過程，在這個過程中如果你設置有Combiner，也是會啟用的，然后在磁盤中生成了眾多溢寫文件。第二種merge方式一直在運行，直到沒有map端的數據時才結束，然后啟動第三種磁盤到磁盤的merge方式生成最終的那個文件。
reduce端流程分析
1 reduce會接收到不同map任務傳來的數據，並且每個map傳來的數據都是有序的。如果reduce端接收的數據量相當小，則直接存儲在內存中（緩沖區大小由mapred.job.shuffle.input.buffer.percent屬性控制，表示用作此用途的堆空間百分比），如果數據量超過了該緩沖區大小的一定比例（由mapred.job.shuffle.merg.percent決定），則對數據合並后溢寫到磁盤中。
2 隨着溢寫文件的增多，后台線程會將它們合並成一個更大的有序的文件，這樣做是為了給后面的合並節省空間。其實不管在map端還是在reduce端，MapReduce都是反復地執行排序，合並操作，現在終於明白了有些人為什么會說：排序是hadoop的靈魂。
3 合並的過程中會產生許多的中間文件（寫入磁盤了），但MapReduce會讓寫入磁盤的數據盡可能地少，並且最后一次合並的結果並沒有寫入磁盤，而是直接輸入到reduce函數。
4 Reducer的輸入文件。不斷地merge后，最后會生成一個“最終文件”。為什么加引號？因為這個文件可能存在於磁盤上，也可能存在於內存中。對我們來說，希望它存放於內存中，直接作為Reducer的輸入，但默認情況下，這個文件是存放於磁盤中的。當Reducer的輸入文件已定，整個Shuffle才最終結束。然后就是Reducer執行，把結果放到HDSF上。
注意：對MapReduce的調優在很大程度上就是對MapReduce Shuffle的性能的調優。
三、內存緩沖區：MapOutputBuffer
兩級索引結構：
環形緩沖區：
1 kvoffsets緩沖區：也叫偏移量索引數組，用於保存key/value信息在位置索引kvindices中的偏移量。當kvoffsets的使用率超過io.sort.spill.percent（默認為80%）后，便會觸發一次SpillThread線程的“溢寫”操作，也就是開始一次spill階段的操作。
2 kvindices緩沖區：也叫位置索引數組，用於保存key/value在數據緩沖區kvbuffer中的起始位置。
3 kvbuffer數據緩沖區：用於保存實際的key/value的值。默認情況下該緩沖區最多可以使用io.sort.mb的95%，當kvbuffer使用率超過io.sort.spill.percent（默認80%）后，便會觸發一次SpillThread線程的“溢寫”操作，也就是開始一次spill階段的操作。