${mapred.local.dir}选择策略--Map Task存放中间结果

1 mapred-site.xml
2 <property>
3   <name>mapred.local.dir</name>
4   <value>/mnt/localdir1/local,/mnt/localdir2/local,/mnt/localdir3/local</value>
5 </property>

 1    /** Get a path from the local FS. If size is known, we go
 2      *  round-robin over the set of disks (via the configured dirs) and return
 3      *  the first complete path which has enough space.
 4      *  
 5      *  If size is not known, use roulette selection -- pick directories
 6      *  with probability proportional to their available space.
 7      */
 8     public synchronized 
 9     Path getLocalPathForWrite(String pathStr, long size, 
10                               Configuration conf, boolean checkWrite
11                               ) throws IOException {
12         //检查task目录是否有变化
13       confChanged(conf);
14       int numDirs = localDirsPath.length;    //获取${mapred.local.dir}目录的个数
15       int numDirsSearched = 0;    //表示已经搜索过的次数
16       //remove the leading slash from the path (to make sure that the uri
17       //resolution results in a valid path on the dir being checked)
18       if (pathStr.startsWith("/")) {    //是指output/spill0.out文件
19         pathStr = pathStr.substring(1);
20       }
21       Path returnPath = null;
22       Path path = new Path(pathStr);
23       
24       //当要写入的数据量大小未知时
25       if(size == SIZE_UNKNOWN) {  //do roulette selection: pick dir with probability 
26                     //proportional to available size
27         long[] availableOnDisk = new long[dirDF.length];
28         long totalAvailable = 0;
29         
30             //build the "roulette wheel"
31         for(int i =0; i < dirDF.length; ++i) { 
32             //分别计算每一个${mapred.local.dir}目录可用大小，并计算总的可用大小
33           availableOnDisk[i] = dirDF[i].getAvailable();
34           totalAvailable += availableOnDisk[i];
35         }
36 
37         // Keep rolling the wheel till we get a valid path
38         Random r = new java.util.Random();
39         while (numDirsSearched < numDirs && returnPath == null) {
40           long randomPosition = Math.abs(r.nextLong()) % totalAvailable;
41           int dir = 0;
42           while (randomPosition > availableOnDisk[dir]) {
43             randomPosition -= availableOnDisk[dir];
44             dir++;
45           }
46           dirNumLastAccessed = dir;    //表示上次访问过的目录
47           //从${mapred.local.dir}中选择一个目录，在其下创建output/spill0.out文件
48           returnPath = createPath(path, checkWrite); 
49           if (returnPath == null) { //如果创建失败(可能存在disk read-only的情况)
50             totalAvailable -= availableOnDisk[dir];
51             availableOnDisk[dir] = 0; // skip this disk
52             numDirsSearched++;
53           }
54         }
55       } else {  //写入的数据量已知
56         while (numDirsSearched < numDirs && returnPath == null) {
57           long capacity = dirDF[dirNumLastAccessed].getAvailable();
58           if (capacity > size) {
59               returnPath = createPath(path, checkWrite);
60           }
61           //使用轮流的方式来选择${mapred.local.dir}
62           dirNumLastAccessed++;
63           dirNumLastAccessed = dirNumLastAccessed % numDirs; 
64           numDirsSearched++;
65         } 
66       }
67       if (returnPath != null) {
68         return returnPath;
69       }
70       
71       //no path found
72       throw new DiskErrorException("Could not find any valid local " +
73           "directory for " + pathStr);
74     }