code: private static void fillInMissingMapOutputs(FileSystem fs, 
                                                TaskAttemptID taskId,
                                                int numMaps,
                                                JobConf conf) throws IOException {
      Class<? extends WritableComparable> keyClass
          = conf.getMapOutputKeyClass().asSubclass(WritableComparable.class);
      Class<? extends Writable> valueClass
          = conf.getMapOutputValueClass().asSubclass(Writable.class);
      MapOutputFile namer = new MapOutputFile(taskId.getJobID());
      namer.setConf(conf);
      for(int i=0; i<numMaps; i++) {
        Path f = namer.getInputFile(i, taskId);
        if (!fs.exists(f)) {
          LOG.info("Create missing input: " + f);
          SequenceFile.Writer out =
            SequenceFile.createWriter(fs, conf, f, keyClass, valueClass);
          out.close();
        }
      }    
    }
docs: Create empty sequence files for any of the map outputs that we don't have.
  @param fs the filesystem to create the files in
  @param dir the directory name to create the files in
  @param conf the jobconf
  @throws IOException if something goes wrong writing