public static void main(String[] args) throws Exception {
02
  Configuration conf = new Configuration();
03
  String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
04
  if (otherArgs.length != 2) {
05
    System.err.println("Usage: wordcount <in> <out>");
06
    System.exit(2);
07
  }
08
  Job job = new Job(conf, "word count");
09
  job.setJarByClass(WordCount.class);
10
  job.setMapperClass(TokenizerMapper.class);
11
  job.setCombinerClass(IntSumReducer.class);
12
  job.setReducerClass(IntSumReducer.class);
13
  job.setOutputKeyClass(Text.class);
14
  job.setOutputValueClass(IntWritable.class);
15
  FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
16
  FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
17
  System.exit(job.waitForCompletion(true) ? 0 : 1);
18
}

JobConf jobCopy = job;
Path jobStagingArea = JobSubmissionFiles.getStagingDir(JobClient.this, jobCopy); // »ñÈ¡µ½StagingAreaĿ¼
JobID jobId = jobSubmitClient.getNewJobId();
Path submitJobDir = new Path(jobStagingArea, jobId.toString());

private String getStagingAreaDirInternal(String user) throws IOException {
2
  final Path stagingRootDir = new Path(conf.get("mapreduce.jobtracker.staging.root.dir", "/tmp/hadoop/mapred/staging"));
3
  final FileSystem fs = stagingRootDir.getFileSystem(conf);
4
  return fs.makeQualified(new Path(stagingRootDir, user+"/.staging")).toString();
5
}

Path filesDir = JobSubmissionFiles.getJobDistCacheFiles(submitJobDir);
2
    Path archivesDir = JobSubmissionFiles.getJobDistCacheArchives(submitJobDir);
3
    Path libjarsDir = JobSubmissionFiles.getJobDistCacheLibjars(submitJobDir);
4
...
5
    Path submitJarFile = JobSubmissionFiles.getJobJar(submitJobDir);
6
    job.setJar(submitJarFile.toString());
7
    fs.copyFromLocalFile(originalJarFile, submitJarFile);

Configuration conf = job.getConfiguration();
2
InputFormat<?, ?> input = ReflectionUtils.newInstance(job.getInputFormatClass(), conf);
3
List<InputSplit> splits = input.getSplits(job);

T[] array = (T[]) splits.toArray(new InputSplit[splits.size()]);
2
// sort the splits into order based on size, so that the biggest
3
// go first
4
Arrays.sort(array, new SplitComparator()); // ¸ù¾ÝInputSplitµÄ³¤¶È×öÁËÒ»¸öÄæÐòÅÅÐò
5
JobSplitWriter.createSplitFiles(jobSubmitDir, conf, jobSubmitDir.getFileSystem(conf), array); // ½«split¼°ÆäÔªÊý¾ÝÐÅϢдÈëHDFS

SerializationFactory factory = new SerializationFactory(conf);
02
int i = 0;
03
long offset = out.getPos();
04
for(T split: array) {
05
  long prevCount = out.getPos();
06
  Text.writeString(out, split.getClass().getName());
07
  Serializer<T> serializer = factory.getSerializer((Class<T>) split.getClass());
08
  serializer.open(out);
09
  serializer.serialize(split); // ½«splitÐòÁл¯Ð´Èëµ½HDFSÎļþÖÐ
10
  long currCount = out.getPos();
11
  String[] locations = split.getLocations();
12
  final int max_loc = conf.getInt(MAX_SPLIT_LOCATIONS, 10);
13
  if (locations.length > max_loc) {
14
    LOG.warn("Max block location exceeded for split: "+ split + " splitsize: " + locations.length + " maxsize: " + max_loc);
15
    locations = Arrays.copyOf(locations, max_loc);
16
  }
17
  info[i++] = new JobSplit.SplitMetaInfo(locations, offset, split.getLength()); // ´´½¨SplitMetaInfoʵÀý
18
  offset += currCount - prevCount;
19
}

private Path file;
2
private long start;
3
private long length;
4
private String[] hosts;

@Override
2
public void write(DataOutput out) throws IOException {
3
  Text.writeString(out, file.toString());
4
  out.writeLong(start);
5
  out.writeLong(length);
6
}

private static void writeJobSplitMetaInfo(FileSystem fs, Path filename,
02
    FsPermission p, int splitMetaInfoVersion,
03
    JobSplit.SplitMetaInfo[] allSplitMetaInfo) throws IOException {
04
  // write the splits meta-info to a file for the job tracker
05
  FSDataOutputStream out = FileSystem.create(fs, filename, p);
06
  out.write(JobSplit.META_SPLIT_FILE_HEADER); 
// дÈëMETAÍ·ÐÅÏ¢£ºMETA_SPLIT_FILE_HEADER = "META-SPL".getBytes("UTF-8");
07
  WritableUtils.writeVInt(out, splitMetaInfoVersion); // META°æ±¾ÐÅÏ¢£º1
08
  WritableUtils.writeVInt(out, allSplitMetaInfo.length); // META¶ÔÏóµÄÊýÁ¿£ºÃ¿¸öInputSplit¶ÔÓ¦Ò»¸öSplitMetaInfo
09
  for (JobSplit.SplitMetaInfo splitMetaInfo : allSplitMetaInfo) {
10
    splitMetaInfo.write(out); // ÿ¸ö¶¼½øÐд洢
11
  }
12
  out.close();
13
}

public void write(DataOutput out) throws IOException {
2
  WritableUtils.writeVInt(out, locations.length); // location¸öÊý
3
  for (int i = 0; i < locations.length; i++) {
4
    Text.writeString(out, locations[i]); // дÈëÿһ¸ölocationλÖÃÐÅÏ¢
5
  }
6
  WritableUtils.writeVLong(out, startOffset); // Æ«ÒÆÁ¿
7
  WritableUtils.writeVLong(out, inputDataLength); // Êý¾Ý³¤¶È
8
}

FSDataOutputStream out = createFile(fs, JobSubmissionFiles.getJobSplitFile(jobSubmitDir), conf);
2
SplitMetaInfo[] info = writeNewSplits(conf, splits, out);

writeJobSplitMetaInfo(fs,JobSubmissionFiles.getJobSplitMetaFile(jobSubmitDir),
2
    new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION), splitVersion, info);

...
02
        Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
03
        jobCopy.set("mapreduce.job.dir", submitJobDir.toString());
04
...
05
        // Write job file to JobTracker's fs      
06
        FSDataOutputStream out = FileSystem.create(fs, submitJobFile, 
new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION));
07
...
08
          try {
09
            jobCopy.writeXml(out);
10
          } finally {
11
            out.close();
12
          }