很多文件中描述，Mapper的數量在預設情況下不可直接控制干預，因為Mapper的數量由輸入的大小和個數決定。在預設情況下，最終input佔據了多少block，就應該啟動多少個Mapper。如果輸入的檔案數量巨大，但是每個檔案的size都小於HDFS的blockSize，那麼會造成啟動的Mapper等於檔案的數量（即每個檔案都佔據了一個block），那麼很可能造成啟動的Mapper數量超出限制而導致崩潰。這些邏輯確實是正確的，但都是在預設情況下的邏輯。其實如果進行一些客戶化的設定，就可以控制了。

在Hadoop中，設定Map task的數量不像設定Reduce task數量那樣直接，即：不能夠通過API直接精確的告訴Hadoop應該啟動多少個Map task。

你也許奇怪了，在API中不是提供了介面org.apache.hadoop.mapred.JobConf.setNumMapTasks(int n)嗎？這個值難道不可以設定Map task的數量嗎？這個API的確沒錯，在文件上解釋”Note: This is only a hint to the framework.“，即這個值對Hadoop的框架來說僅僅是個提示，不起決定性的作用。也就是說，即便你設定了，也不一定得到你想要的效果。

1. InputFormat介紹

在具體設定Map task數量之前，非常有必要了解一下與Map-Reduce輸入相關的基礎知識。

這個介面（org.apache.hadoop.mapred.InputFormat）描述了Map-Reduce job的輸入規格說明

InputFormat有多種具體實現，諸如FileInputFormat（處理基於檔案的輸入的基礎抽象類）, DBInputFormat（處理基於資料庫的輸入，資料來自於一個能用SQL查詢的表）,KeyValueTextInputFormat（特殊的FineInputFormat，處理Plain Text File，檔案由回車或者回車換行符分割成行，每一行由key.value.separator.in.input.line分割成Key和Value），CompositeInputFormat，DelegatingInputFormat等。在絕大多數應用場景中都會使用FileInputFormat及其子型別。

通過以上的簡單介紹，我們知道InputFormat決定著InputSplit，每個InputSplit會分配給一個單獨的Mapper，因此InputFormat決定了具體的Map task數量。

2. FileInputFormat中影響Map數量的因素

在日常使用中，FileInputFormat是最常用的InputFormat，它有很多具體的實現。以下分析的影響Map數量的因素僅對FileInputFormat及其子類有效，其他非FileInputFormat可以去檢視相應的 getSplits(JobConf job, int numSplits) 具體實現即可。

請看如下程式碼段（摘抄自org.apache.hadoop.mapred.FileInputFormat.getSplits，hadoop-0.20.205.0原始碼）：

1. long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);  
2. long minSize = Math.max(job.getLong("mapred.min.split.size", 1), minSplitSize);  
3. for (FileStatus file: files) {  
4.   Path path = file.getPath();  
5.   FileSystem fs = path.getFileSystem(job);  
6.   if ((length != 0) && isSplitable(fs, path)) {   
7.     long blockSize = file.getBlockSize();  
8.     long splitSize = computeSplitSize(goalSize, minSize, blockSize);  
9.     long bytesRemaining = length;  
10.     while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {  
11.       String[] splitHosts = getSplitHosts(blkLocations,length-bytesRemaining, splitSize, clusterMap);  
12.       splits.add(new FileSplit(path, length-bytesRemaining, splitSize, splitHosts));  
13.       bytesRemaining -= splitSize;  
14.     }  
15.     if (bytesRemaining != 0) {  
16.       splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining, blkLocations[blkLocations.length-1].getHosts()));  
17.     }  
18.   } elseif (length != 0) {  
19.     String[] splitHosts = getSplitHosts(blkLocations,0,length,clusterMap);  
20.     splits.add(new FileSplit(path, 0, length, splitHosts));  
21.   } else {   
22.     //Create empty hosts array for zero length files
23.     splits.add(new FileSplit(path, 0, length, new String[0]));  
24.   }  
25. }  
26. return splits.toArray(new FileSplit[splits.size()]);  
27. protectedlong computeSplitSize(long goalSize, long minSize, long blockSize) {  
28.     return Math.max(minSize, Math.min(goalSize, blockSize));  
29. }  

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相比取較大的。

3. 如何調整Map的數量

有了2的分析，下面調整Map的數量就很容易了。

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

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

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

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

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

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

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

增加mapper的數量，可以通過減小每個mapper的輸入做到，即減小blockSize或者減小mapred.min.split.size的值。

• map task的數量即mapred.map.tasks的引數值，使用者不能直接設定這個引數。Input Split的大小，決定了一個Job擁有多少個map。預設input split的大小是64M（與dfs.block.size的預設值相同）。然而，如果輸入的資料量巨大，那麼預設的64M的block會有幾萬甚至幾十萬的Map Task，叢集的網路傳輸會很大，最嚴重的是給Job Tracker的排程、佇列、記憶體都會帶來很大壓力。mapred.min.split.size這個配置項決定了每個 Input Split的最小值，使用者可以修改這個引數，從而改變map task的數量。
• 一個恰當的map並行度是大約每個節點10-100個map，且最好每個map的執行時間至少一分鐘。
•  reduce task的數量由mapred.reduce.tasks這個引數設定，預設值是1。
• 合適的reduce task數量是0.95或者0.75*( nodes * mapred.tasktracker.reduce.tasks.maximum), mapred.tasktracker.tasks.reduce.maximum的數量一般設定為各節點cpu core數量，即能同時計算的slot數量。對於0.95，當map結束時，所有的reduce能夠立即啟動；對於1.75，較快的節點結束第一輪reduce後，可以開始第二輪的reduce任務，從而提高負載均衡。

• 對一 個job的map數和reduce數的設定對一個job的執行是非常重要的，並且非常簡單。以下是一些設 置這幾個值的經驗總結：

  • 如果job的每個map或者 reduce task的執行時間都只有30-40秒鐘，那麼就減少該job的map或者reduce數，每一個task(map|reduce)的setup和加入到 排程器中進行排程，這個中間的過程可能都要花費幾秒鐘，所以如果每個task都非常快就跑完了，就會在task的開始和結束的時候浪費太多的時間。JVM 的reuse方式也可以解決 這個問題。
  • 如 果某個input的檔案 非常的大，比如 1TB，可以考慮將hdfs上的每個block size設大，比如設成256MB或者512MB，這樣map和reduce的資料 可以減小。而且使用者還可以通過命令 ：hadoop distcp -Ddfs.block.size=$[256*1024*1024] /path/to/inputdata /path/to/inputdata-with-largeblocks的方式來將已經存在咋hdfs上的資料進行大塊化。然後刪除掉原先的檔案。
  • 只 要每個task都執行至少30-40秒鐘，就可以考慮將mapper數擴大，比如叢集的map slots為100個，那麼就不要將一個job的mapper設成101，這樣前100個map能夠並行完成，而最後一個map要在前100個 mapper結束後才開始，因此在reduce開始執行前，map階段的時間幾乎就要翻倍。
  • 儘量不要執行太多的reduce task。對大多數job來說，最好rduce的個數最多和叢集中的reduce持平，或者比叢集的 reduce slots小。這個對於小叢集而言，尤其重要。