@@ -0,0 +1,1531 @@
+package flambo.hadoop;%0A%0Aimport java.io.IOException;%0A%0Aimport org.apache.commons.logging.Log;%0Aimport org.apache.commons.logging.LogFactory;%0Aimport org.apache.hadoop.fs.Path;%0Aimport org.apache.hadoop.fs.FileSystem;%0Aimport org.apache.hadoop.mapred.*;%0A%0Apublic class DirectOutputCommitter extends OutputCommitter %7B%0A%0A  public static final Log LOG = LogFactory.getLog(DirectOutputCommitter.class);%0A%0A  @Override%0A  public void setupJob(JobContext jobContext) %7B%0A    LOG.warn(%22setupJob or nah%22);%0A  %7D%0A%0A  @Override%0A  public void setupTask(TaskAttemptContext taskContext) %7B%0A    LOG.warn(%22setupTask or nah%22);%0A  %7D%0A%0A  @Override%0A  public boolean needsTaskCommit(TaskAttemptContext taskContext) %7B%0A    return true;%0A  %7D%0A%0A  @Override%0A  public void commitTask(TaskAttemptContext taskContext) %7B%0A    LOG.warn(%22commitTask or nah%22);%0A  %7D%0A%0A  @Override%0A  public void abortTask(TaskAttemptContext taskContext) %7B%0A    LOG.warn(%22abortTask or nah%22);%0A  %7D%0A%0A  @Override%0A  public void commitJob(JobContext jobContext) throws IOException %7B%0A    JobConf conf = jobContext.getJobConf();%0A    if (shouldCreateSuccessFile(conf)) %7B%0A      Path outputPath = FileOutputFormat.getOutputPath(conf);%0A      if (outputPath != null) %7B%0A        FileSystem fs = outputPath.getFileSystem(conf);%0A        Path path = new Path(outputPath, FileOutputCommitter.SUCCEEDED_FILE_NAME);%0A        fs.create(path);%0A        fs.close();%0A      %7D%0A    %7D%0A  %7D%0A%0A  private boolean shouldCreateSuccessFile(JobConf conf) %7B%0A    return conf.getBoolean(%22mapreduce.fileoutputcommitter.marksuccessfuljobs%22, true);%0A  %7D%0A%7D%0A

add a directoutputcommitter for writing to s3