一  分塊（Block）

      HDFS儲存系統中，引入了檔案系統的分塊概念（block），塊是儲存的最小單位，HDFS定義其大小為64MB。與單磁碟檔案系統相似，儲存在 HDFS上的檔案均儲存為多個塊，不同的是，如果某檔案大小沒有到達64MB，該檔案也不會佔據整個塊空間。在分散式的HDFS叢集上，Hadoop系統保證一個塊儲存在一個datanode上。

      把File劃分成Block，這個是物理上真真實實的進行了劃分，資料檔案上傳到HDFS裡的時候，需要劃分成一塊一塊，每塊的大小由hadoop-default.xml裡配置選項進行劃分。一個大檔案可以把劃分後的所有塊儲存到同一個磁碟上，也可以在每個磁碟上都存在這個檔案的分塊。

      這個就是預設的每個塊64M：

<property>  
  <name>dfs.block.size</name>  
  <value>67108864</value>  
  <description>The default block size for new files.</description>  
</property>  

      資料劃分的時候有冗餘，即進行備份(預設是3個），個數是由以下配置指定的。具體的物理劃分步驟由Namenode決定。

<property>  
  <name>dfs.replication</name>  
  <value>3</value>  
  <description>Default block replication.   
  The actual number of replications can be specified when the file is created.  
  The default is used if replication is not specified in create time.  
  </description>  
</property>
 

二  分片（splits）

      由InputFormat這個介面來定義的，其中有個getSplits方法。這裡有一個新的概念：fileSplit。每個map處理一個fileSplit，所以有多少個fileSplit就有多少個map（map數並不是單純的由使用者設定決定的）。

      我們來看一下hadoop分配splits的原始碼：

long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
long minSize = Math.max(job.getLong("mapred.min.split.size", 1), minSplitSize);

for (FileStatus file: files) {
  Path path = file.getPath();
  FileSystem fs = path.getFileSystem(job);
  if ((length != 0) && isSplitable(fs, path)) { 
    long blockSize = file.getBlockSize();
    long splitSize = computeSplitSize(goalSize, minSize, blockSize);
    
    long bytesRemaining = length;
    while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
      String[] splitHosts = getSplitHosts(blkLocations,length-bytesRemaining, splitSize, clusterMap);
      splits.add(new FileSplit(path, length-bytesRemaining, splitSize, splitHosts));
      bytesRemaining -= splitSize;
    }

    if (bytesRemaining != 0) {
      splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining, blkLocations[blkLocations.length-1].getHosts()));
    }
  } else if (length != 0) {
    String[] splitHosts = getSplitHosts(blkLocations,0,length,clusterMap);
    splits.add(new FileSplit(path, 0, length, splitHosts));
  } else { 
    //Create empty hosts array for zero length files
    splits.add(new FileSplit(path, 0, length, new String[0]));
  }
}

return splits.toArray(new FileSplit[splits.size()]);

protected long computeSplitSize(long goalSize, long minSize, long blockSize) {
    return Math.max(minSize, Math.min(goalSize, blockSize));
}
 

totalSize：是整個Map-Reduce job所有輸入的總大小。

numSplits：來自job.getNumMapTasks()，即在job啟動時用org.apache.hadoop.mapred.JobConf.setNumMapTasks(int n)設定的值，給M-R框架的Map數量的提示。

goalSize：是輸入總大小與提示Map task數量的比值，即期望每個Mapper處理多少的資料，僅僅是期望，具體處理的資料數由下面的computeSplitSize決定。

minSplitSize：預設為1，可由子類複寫函式protected void setMinSplitSize(long minSplitSize) 重新設定。一般情況下，都為1，特殊情況除外。

minSize：取的1和mapred.min.split.size中較大的一個。

blockSize：HDFS的塊大小，預設為64M，一般大的HDFS都設定成128M。

splitSize：就是最終每個Split的大小，那麼Map的數量基本上就是totalSize/splitSize。

接下來看看computeSplitSize的邏輯：首先在goalSize（期望每個Mapper處理的資料量）和HDFS的block size中取較小的，然後與mapred.min.split.size相比取較大的。

一個片為一個splits，即一個map，只要搞清楚片的大小，就能計算出執行時的map數。而一個split的大小是由goalSize, minSize, blockSize這三個值決定的。computeSplitSize的邏輯是，先從goalSize和blockSize兩個值中選出最小的那個（比如一般不設定map數，這時blockSize為當前檔案的塊size，而goalSize是檔案大小除以使用者設定的map數得到的，如果沒設定的話，預設是1），在預設的大多數情況下，blockSize比較小。然後再取blockSize和minSize中最大的那個。而minSize如果不通過”mapred.min.split.size”設定的話（”mapred.min.split.size”預設為0），minSize為1，這樣得出的一個splits的size就是blockSize，即一個塊一個map，有多少塊就有多少map。

input_file_num : 輸入檔案的個數
（1）預設map個數
如果不進行任何設定，預設的map個數是和blcok_size相關的。
default_num = total_size / block_size;
（2）期望map數量
可以通過引數mapred.map.tasks來設定程式設計師期望的map個數，但是這個個數只有在大於default_num的時候，才會生效。
goal_num =mapred.map.tasks;
（3）設定處理的檔案大小
可以通過mapred.min.split.size 設定每個task處理的檔案大小，但是這個大小隻有在大於
block_size的時候才會生效。
split_size = max(
mapred.min.split.size,
block_size);split_num = total_size / split_size;
（4）計算的map個數
compute_map_num = min(split_num, max(default_num, goal_num))
除了這些配置以外，mapreduce還要遵循一些原則。 mapreduce的每一個map處理的資料是不能跨越檔案的，也就是說max_map_num <= input_file_num。 所以，最終的map個數應該為：
final_map_num = min(compute_map_num, input_file_num)
經過以上的分析，在設定map個數的時候，可以簡單的總結為以下幾點：
  i）如果想增加map個數，則設定mapred.map.tasks 為一個較大的值。
  ii）如果想減小map個數，則設定mapred.min.split.size 為一個較大的值。

Map數量的調整

有了上述分析，如何調整map的數量就顯而易見了。

減小Map-Reduce job 啟動時建立的Mapper數量

當處理大批量的大資料時，一種常見的情況是job啟動的mapper數量太多而超出了系統限制，導致Hadoop丟擲異常終止執行。解決這種異常的思路是減少mapper的數量。具體如下：

輸入檔案size巨大，但不是小檔案

　　這種情況可以通過增大每個mapper的input size，即增大minSize或者增大blockSize來減少所需的mapper的數量。增大blockSize通常不可行，因為當HDFS被hadoop namenode -format之後，blockSize就已經確定了（由格式化時dfs.block.size決定），如果要更改blockSize，需要重新格式化HDFS，這樣當然會丟失已有的資料。所以通常情況下只能通過增大minSize，即增大mapred.min.split.size的值。

輸入檔案數量巨大，且都是小檔案

　　所謂小檔案，就是單個檔案的size小於blockSize。這種情況通過增大mapred.min.split.size不可行，需要使用FileInputFormat衍生的CombineFileInputFormat將多個input path合併成一個InputSplit送給mapper處理，從而減少mapper的數量。

增加Map-Reduce job 啟動時建立的Mapper數量

增加mapper的數量，可以通過減小每個mapper的輸入做到，即減小blockSize或者減小mapred.min.split.size的值。通常情況下都是通過增大minSize，即增大mapred.min.split.size的值。

三  Shuffle機制

Shuffle過程是MapReduce的核心，描述著資料從map task輸出到reduce task輸入的這段過程。

Hadoop的叢集環境，大部分的map task和reduce task是執行在不同的節點上的，那麼reduce就要取map的輸出結果。那麼叢集中執行多個Job時，task的正常執行會對叢集內部的網路資源消耗嚴重。雖說這種消耗是正常的，是不可避免的，但是，我們可以採取措施儘可能的減少不必要的網路資源消耗。另一方面，每個節點的內部，相比於記憶體，磁碟IO對Job完成時間的影響相當的大。

所以：從以上分析，shuffle過程的基本要求：

　　1.完整地從map task端拉取資料到reduce task端

　　2.在拉取資料的過程中，儘可能地減少網路資源的消耗

　　3.儘可能地減少磁碟IO對task執行效率的影響

那麼，Shuffle的設計目的就要滿足以下條件：

　　1.保證拉取資料的完整性

　　2.儘可能地減少拉取資料的資料量

　　3.儘可能地使用節點的記憶體而不是磁碟

一、map階段
map節點執行map task任務生成map的輸出結果
shuffle的工作內容：
從運算效率的出發點，map輸出結果優先儲存在map節點的記憶體中。每個map task都有一個記憶體緩衝區，儲存著map的輸出結果，預設大小100M（由io.sort.mb屬性控制），一旦記憶體緩衝達到閾值0.8（io.sort.spill.percent），一個後臺執行緒就會將將緩衝區中的資料以一個臨時檔案的方式存（spill)到磁碟的指定目錄（mapred.local.dir）。同時，在寫磁碟前，會進行partition、sort操作, 如果有combiner, combine排序後資料。 當整個map task結束後再對磁碟中這個map task所產生的所有臨時檔案做合併，生成最終的輸出檔案。最後，等待reduce task來拉取資料。當然，如果map task的結果不大，能夠完全儲存到記憶體緩衝區，且未達到記憶體緩衝區的閥值，那麼就不會有寫臨時檔案到磁碟的操作，也不會有後面的合併。

圖解如下：

環形緩衝區：是使用指標機制把記憶體中的地址首尾相接形成一個儲存中間資料的快取區域，預設100MB;80M閾值,20M緩衝區,是為了解決寫入環形緩衝區資料的速度大於寫出到spill檔案的速度是資料的不丟失;

Spill檔案：spill檔案是環形緩衝區到達閾值後寫入到磁碟的單個檔案.這些檔案在map階段計算結束時,會合成分好區的一個merge檔案供給給reduce任務抓取;spill檔案過小的時候,就不會浪費io資源合併merge;預設情況下3個以下spill檔案不合並;對於在環形緩衝區中的資料,最終達不到80M但是資料已經計算完畢的情況,map任務將會呼叫flush將緩衝區中的資料強行寫出spill檔案。

二、reduce階段

當mapreduce任務提交後，reduce task就不斷通過RPC從JobTracker那裡獲取map task是否完成的資訊，如果獲知某臺TaskTracker上的map task執行完成，Shuffle的後半段過程就開始啟動。其實呢，reduce task在執行之前的工作就是：不斷地拉取當前job裡每個map task的最終結果，並對不同地方拉取過來的資料不斷地做merge，過程如下：

reduce階段分三個步驟：
抓取，合併，排序
1 reduce 任務會建立並行的抓取執行緒（fetcher）負責從完成的map任務中獲取結果檔案，是否完成是通過rpc心跳監聽，通過http協議抓取；預設是5個抓取執行緒，可調，為了使整體並行，在map任務量大，分割槽多的時候，抓取執行緒調大；
2 抓取過來的資料會先儲存在記憶體中，如果記憶體過大也溢位，不可見，不可調，但是單位是每個merge檔案，不會切分資料；每個merge檔案都會被封裝成一個segment的物件，這個物件控制著這個merge檔案的讀取記錄操作，有兩種情況出現：
      在記憶體中有merge資料
      在溢寫之後存到磁碟上的資料
通過建構函式的區分，來分別建立對應的segment物件
3 這種segment物件會放到一個記憶體佇列中MergerQueue，對記憶體和磁碟上的資料分別進行合併，記憶體中的merge對應的segment直接合並，磁碟中的合併與一個叫做合併因子的factor有關（預設是10）
4 排序問題

MergerQueue繼承輪換排序的介面，每一個segment 是排好序的，而且按照key的值大小邏輯（和真的大小沒關係）；每一個segment的第一個key都是邏輯最小，而所有的segment的排序是按照第一個key大小排序的，最小的在前面，這種邏輯總能保證第一個segment的第一個key值是所有key的邏輯最小檔案，合併之後，最終交給reduce函式計算的，是MergeQueue佇列，每次計算的提取資料邏輯都是提取第一個segment的第一個key和value資料，一旦segment被呼叫了提取key的方法，MergeQueue佇列將會整體重新按照最小key對segment排序，最終形成整體有序的計算結果；

四  執行流程