1，在介紹hadoop寫檔案的時候我們經常會說首先分割檔案為多個塊；那麼是怎麼分割的呢？

這裡其實不要有過的糾結，這裡的塊是block，是hdfs中切塊的大小，屬於物理劃分，預設64M，在hadoop-default.xml配置中有體現：

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

當然如果檔案沒有64M也不會佔據整塊空間。

將檔案分割成多個塊後，形成一個數據佇列，然後依次寫入datanode列表。

再者，如果寫入的是個資料夾，而且每個檔案的都不大，這樣在hdfs中是預設每個檔案一個塊的，即使沒有64m，當然也可做優化處理，不過hbase更便利於處理把小檔案合併到一個塊中，這個我會在其他博文中介紹。

2，下面我們說說split，並與block的關係

首先，split是mapreduce中的概念，而block是hdfs中切塊的大小。

如下：

//設定要處理的文字資料所存放的路徑  
        FileInputFormat.setInputPaths(wordCountJob, "hdfs://ubuntu:9000/input/aa.txt");  
        FileOutputFormat.setOutputPath(wordCountJob, new Path("hdfs://ubuntu:9000/output/"));  
 

我們設定要處理的檔案路徑時都會用到fileInputFormat類， 不過我們更多看到的是inputFormat，其實fileInputFormat這個類的也是實現inputFomat介面，

下面我們接著看原始碼，說明為什麼需要分片？

以hadoop自帶的wordCount的原始碼為例：

for (int i = 0; i < otherArgs.length - 1; ++i) {  
  FileInputFormat.addInputPath(job, new Path(otherArgs[i]));  
}  
FileOutputFormat.setOutputPath(job,  
  new Path(otherArgs[otherArgs.length - 1]));  
System.exit(job.waitForCompletion(true) ? 0 : 1);  
 

我們看到使用的InputFormat是FileOutputFormat，任務執行呼叫了Job的waitForCompletion方法。waitForCompletion方法中真正提交job的程式碼如下：

public boolean waitForCompletion(boolean verbose  
                                 ) throws IOException, InterruptedException,  
                                          ClassNotFoundException {  
  if (state == JobState.DEFINE) {  
    submit();  
  }  
  // 省略本文不關心的程式碼  
  return isSuccessful();  
}  

這裡的submit方法的實現如下：

public void submit()   
         throws IOException, InterruptedException, ClassNotFoundException {  
    // 省略本文不關心的程式碼</span>  
    final JobSubmitter submitter =   
        getJobSubmitter(cluster.getFileSystem(), cluster.getClient());  
    status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {  
      public JobStatus run() throws IOException, InterruptedException,   
      ClassNotFoundException {  
        return submitter.submitJobInternal(Job.this, cluster);  
      }  
    });  
    state = JobState.RUNNING;  
    LOG.info("The url to track the job: " + getTrackingURL());  
   }  

submit方法首先建立了JobSubmitter例項，然後非同步呼叫了JobSubmitter的submitJobInternal方法。JobSubmitter的submitJobInternal方法有關劃分任務的程式碼如下：

// Create the splits for the job  
LOG.debug("Creating splits at " + jtFs.makeQualified(submitJobDir));  
int maps = writeSplits(job, submitJobDir);  
conf.setInt(MRJobConfig.NUM_MAPS, maps);  
LOG.info("number of splits:" + maps); 

writeSplits方法的實現如下：

private int writeSplits(org.apache.hadoop.mapreduce.JobContext job,  
    Path jobSubmitDir) throws IOException,  
    InterruptedException, ClassNotFoundException {  
  JobConf jConf = (JobConf)job.getConfiguration();  
  int maps;  
  if (jConf.getUseNewMapper()) {  
    maps = writeNewSplits(job, jobSubmitDir);  
  } else {  
    maps = writeOldSplits(jConf, jobSubmitDir);  
  }  
  return maps;  
}  

由於WordCount使用的是新的mapreduce API，所以最終會呼叫writeNewSplits方法。writeNewSplits的實現如下：

private <T extends InputSplit>  
int writeNewSplits(JobContext job, Path jobSubmitDir) throws IOException,  
    InterruptedException, ClassNotFoundException {  
  Configuration conf = job.getConfiguration();  
  InputFormat<?, ?> input =  
    ReflectionUtils.newInstance(job.getInputFormatClass(), conf);  
  
  List<InputSplit> splits = input.getSplits(job);  
  T[] array = (T[]) splits.toArray(new InputSplit[splits.size()]);  
  
  // sort the splits into order based on size, so that the biggest  
  // go first  
  Arrays.sort(array, new SplitComparator());  
  JobSplitWriter.createSplitFiles(jobSubmitDir, conf,   
      jobSubmitDir.getFileSystem(conf), array);  
  return array.length;  
}  

writeNewSplits方法中，劃分任務數量最關鍵的程式碼即為InputFormat的getSplits方法（提示：大家可以直接通過此處的呼叫，檢視不同InputFormat的劃分任務實現）。根據前面的分析我們知道此時的InputFormat即為FileOutputFormat，其getSplits方法的實現如下：

public List<InputSplit> getSplits(JobContext job) throws IOException {  
  Stopwatch sw = new Stopwatch().start();  
  long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));  
  long maxSize = getMaxSplitSize(job);  
  
  // generate splits  
  List<InputSplit> splits = new ArrayList<InputSplit>();  
  List<FileStatus> files = listStatus(job);  
  for (FileStatus file: files) {  
    Path path = file.getPath();  
    long length = file.getLen();  
    if (length != 0) {  
      BlockLocation[] blkLocations;  
      if (file instanceof LocatedFileStatus) {  
        blkLocations = ((LocatedFileStatus) file).getBlockLocations();  
      } else {  
        FileSystem fs = path.getFileSystem(job.getConfiguration());  
        blkLocations = fs.getFileBlockLocations(file, 0, length);  
      }  
      if (isSplitable(job, path)) {  
        long blockSize = file.getBlockSize();  
        long splitSize = computeSplitSize(blockSize, minSize, maxSize);  
  
        long bytesRemaining = length;  
        while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {  
          int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);  
          splits.add(makeSplit(path, length-bytesRemaining, splitSize,  
                      blkLocations[blkIndex].getHosts(),  
                      blkLocations[blkIndex].getCachedHosts()));  
          bytesRemaining -= splitSize;  
        }  
  
        if (bytesRemaining != 0) {  
          int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);  
          splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,  
                     blkLocations[blkIndex].getHosts(),  
                     blkLocations[blkIndex].getCachedHosts()));  
        }  
      } else { // not splitable  
        splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),  
                    blkLocations[0].getCachedHosts()));  
      }  
    } else {   
      //Create empty hosts array for zero length files  
      splits.add(makeSplit(path, 0, length, new String[0]));  
    }  
  }  
  // Save the number of input files for metrics/loadgen  
  job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());  
  sw.stop();  
  if (LOG.isDebugEnabled()) {  
    LOG.debug("Total # of splits generated by getSplits: " + splits.size()  
        + ", TimeTaken: " + sw.elapsedMillis());  
  }  
  return splits;  
}  

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，可理解為一個block塊，這樣得出的一個splits的size就是blockSize，即一個塊一個map，有多少塊就有多少map。

split的大小時預設和hdfs的block塊大小一致，但是可以通過配置檔案自己設定： 
其中有倆個配置檔案（如下）：

--minsize   預設大小為1
mapreduce.input.fileinputformat.split.minsize  

--maxsize   預設大小為Long.MAXValue 
mapreduce.input.fileinputformat.split.maxsize

舉例：

比如說我問寫入一個資料夾，裡面有10個只有幾k的檔案，所以List<FileStatus> files = listStatus(job);方法返回的files列表的大小為10。在遍歷files列表的過程中，會獲取每個檔案的blockSize，最終呼叫computeSplitSize方法計算每個輸入檔案應當劃分的任務數。computeSplitSize方法的實現如下：

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

然後根據

  long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));  

很明顯取split預設值，也就是一個塊，那麼10個就要分為10塊，這也說明為什麼處理小檔案時，block的大小小於split的 大小。同時我們看到了一個split分配一個map任務。

這裡我們可以總結下split大小與block的關係：

（1）block塊的小於split分片的最小值，那split的值就是split分片的大小

（2）block塊的小大介於split分片配置的最小值和最大值之間，block的大小就是split的大小。

（3）block塊的大小大於split分片的最大值，split的大小就是split配置的最大值。但會增加map執行的併發度，但是會造成在節點之間拉取資料

也有公式可以計算split也就是map任務數，這裡就不做討論了。

一個map對應一個split分片嗎？

經過上面的討論，答案是顯而易見的：

map個數：由任務切片spilt決定的，預設情況下一個split的大小就是block
由參與任務的檔案個數決定的