mapred-site.xml

 <property>

   <name>mapred.local.dir</name>

   <value>/mnt/localdir1/local,/mnt/localdir2/local,/mnt/localdir3/local</value>

 </property>

    /** Get a path from the local FS. If size is known, we go

      *  round-robin over the set of disks (via the configured dirs) and return

      *  the first complete path which has enough space.

      *

      *  If size is not known, use roulette selection -- pick directories

      *  with probability proportional to their available space.

      */

     public synchronized

     Path getLocalPathForWrite(String pathStr, long size,

                               Configuration conf, boolean checkWrite

                               ) throws IOException {

         //检查task目录是否有变化

       confChanged(conf);

       int numDirs = localDirsPath.length;    //获取${mapred.local.dir}目录的个数

       int numDirsSearched = 0;    //表示已经搜索过的次数

       //remove the leading slash from the path (to make sure that the uri

       //resolution results in a valid path on the dir being checked)

       if (pathStr.startsWith("/")) {    //是指output/spill0.out文件

         pathStr = pathStr.substring(1);

       }

       Path returnPath = null;

       Path path = new Path(pathStr);

       //当要写入的数据量大小未知时

       if(size == SIZE_UNKNOWN) {  //do roulette selection: pick dir with probability

                     //proportional to available size

         long[] availableOnDisk = new long[dirDF.length];

         long totalAvailable = 0;

             //build the "roulette wheel"

         for(int i =0; i < dirDF.length; ++i) {

             //分别计算每一个${mapred.local.dir}目录可用大小，并计算总的可用大小

           availableOnDisk[i] = dirDF[i].getAvailable();

           totalAvailable += availableOnDisk[i];

         }

         // Keep rolling the wheel till we get a valid path

         Random r = new java.util.Random();

         while (numDirsSearched < numDirs && returnPath == null) {

           long randomPosition = Math.abs(r.nextLong()) % totalAvailable;

           int dir = 0;

           while (randomPosition > availableOnDisk[dir]) {

             randomPosition -= availableOnDisk[dir];

             dir++;

           }

           dirNumLastAccessed = dir;    //表示上次访问过的目录

           //从${mapred.local.dir}中选择一个目录，在其下创建output/spill0.out文件

           returnPath = createPath(path, checkWrite);

           if (returnPath == null) { //如果创建失败(可能存在disk read-only的情况)

             totalAvailable -= availableOnDisk[dir];

             availableOnDisk[dir] = 0; // skip this disk

             numDirsSearched++;

           }

         }

       } else {  //写入的数据量已知

         while (numDirsSearched < numDirs && returnPath == null) {

           long capacity = dirDF[dirNumLastAccessed].getAvailable();

           if (capacity > size) {

               returnPath = createPath(path, checkWrite);

           }

           //使用轮流的方式来选择${mapred.local.dir}

           dirNumLastAccessed++;

           dirNumLastAccessed = dirNumLastAccessed % numDirs;

           numDirsSearched++;

         }

       }

       if (returnPath != null) {

         return returnPath;

       }

       //no path found

       throw new DiskErrorException("Could not find any valid local " +

           "directory for " + pathStr);

     }

 /** This method gets called everytime before any read/write to make sure

      * that any change to localDirs is reflected immediately.

      */

     private synchronized void confChanged(Configuration conf

                                           ) throws IOException {

         //contextCfgItemName="mapred.local.dir"

       String newLocalDirs = conf.get(contextCfgItemName);

       if (!newLocalDirs.equals(savedLocalDirs)) { //savedLocalDirs代表上个task的工作目录

         String[] localDirs = conf.getStrings(contextCfgItemName);

         localFS = FileSystem.getLocal(conf);

         int numDirs = localDirs.length;  //${mapred.local.dir}目录的个数

         ArrayList<String> dirs = new ArrayList<String>(numDirs);

         ArrayList<DF> dfList = new ArrayList<DF>(numDirs);

         for (int i = 0; i < numDirs; i++) {

           try {

             // filter problematic directories

             Path tmpDir = new Path(localDirs[i]);

             //检查task的工作目录(attempt....)是否存在，如果不存在，则新建

             if(localFS.mkdirs(tmpDir)|| localFS.exists(tmpDir)) {

               try {

                 DiskChecker.checkDir(new File(localDirs[i]));

                 dirs.add(localDirs[i]);

                 dfList.add(new DF(new File(localDirs[i]), 30000));

               } catch (DiskErrorException de) {

                 LOG.warn( localDirs[i] + "is not writable\n" +

                     StringUtils.stringifyException(de));

               }

             } else {

               LOG.warn( "Failed to create " + localDirs[i]);

             }

           } catch (IOException ie) {

             LOG.warn( "Failed to create " + localDirs[i] + ": " +

                 ie.getMessage() + "\n" + StringUtils.stringifyException(ie));

           } //ignore

         }

         localDirsPath = new Path[dirs.size()];

         for(int i=0;i<localDirsPath.length;i++) {

           localDirsPath[i] = new Path(dirs.get(i));

         }

         dirDF = dfList.toArray(new DF[dirs.size()]);

         savedLocalDirs = newLocalDirs;  //保存此次的task工作目录

         // randomize the first disk picked in the round-robin selection

         //因为该task所有的工作目录都遍历过了，所以随机选择一个目录作为最后访问过的目录

         dirNumLastAccessed = dirIndexRandomizer.nextInt(dirs.size());

       }

     }

 /**

  * Filesystem disk space usage statistics. Uses the unix 'df' program to get

  * mount points, and java.io.File for space utilization. Tested on Linux,

  * FreeBSD, Cygwin.

  */

 public class DF extends Shell {

   private final String dirPath;

   private final File dirFile;

   private String filesystem;

   private String mount;