Merge -r 670085:670086 from trunk to branch-0.18 to fix HADOOP-3598

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.18@670087 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -88,6 +88,10 @@ Release 0.18.0 - Unreleased
 
     HADOOP-3512. Separate out the tools into a tools jar. (omalley)
 
+    HADOOP-3598. Ensure that temporary task-output directories are not created
+    if they are not necessary e.g. for Maps with no side-effect files.
+    (acmurthy)
+
   NEW FEATURES
 
     HADOOP-3074. Provides a UrlStreamHandler for DFS and other FS,

src/mapred/org/apache/hadoop/mapred/FileOutputFormat.java

@@ -195,5 +195,40 @@ public abstract class FileOutputFormat<K, V> implements OutputFormat<K, V> {
     return name == null ? null: new Path(name);
   }
 
+  /**
+   * Helper function to create the task's temporary output directory and 
+   * return the path to the task's output file.
+   * 
+   * @param conf job-configuration
+   * @param name temporary task-output filename
+   * @return path to the task's temporary output file
+   * @throws IOException
+   */
+  protected static Path getTaskOutputPath(JobConf conf, String name) 
+  throws IOException {
+    // ${mapred.job.dir}
+    Path outputPath = getOutputPath(conf);
+    if (outputPath == null) {
+      throw new IOException("Undefined job output-path");
+    }
+
+    // ${mapred.out.dir}/_temporary
+    Path jobTmpDir = new Path(outputPath, MRConstants.TEMP_DIR_NAME);
+    FileSystem fs = jobTmpDir.getFileSystem(conf);
+    if (!fs.exists(jobTmpDir)) {
+      throw new IOException("The temporary job-output directory " + 
+          jobTmpDir.toString() + " doesn't exist!"); 
+    }
+
+    // ${mapred.out.dir}/_temporary/_${taskid}
+    Path taskTmpDir = getWorkOutputPath(conf);
+    if (!fs.mkdirs(taskTmpDir)) {
+      throw new IOException("Mkdirs failed to create " 
+          + taskTmpDir.toString());
+    }
+    
+    // ${mapred.out.dir}/_temporary/_${taskid}/${name}
+    return new Path(taskTmpDir, name);
+  } 
 }
 

src/mapred/org/apache/hadoop/mapred/MapFileOutputFormat.java

@@ -42,14 +42,10 @@ extends FileOutputFormat<WritableComparable, Writable> {
   public RecordWriter<WritableComparable, Writable> getRecordWriter(FileSystem ignored, JobConf job,
                                       String name, Progressable progress)
     throws IOException {
-
-    Path outputPath = getWorkOutputPath(job);
-    FileSystem fs = outputPath.getFileSystem(job);
-    if (!fs.exists(outputPath)) {
-      throw new IOException("Output directory doesnt exist");
-    }
-    Path file = new Path(outputPath, name);
+    // get the path of the temporary output file 
+    Path file = FileOutputFormat.getTaskOutputPath(job, name);
     
+    FileSystem fs = file.getFileSystem(job);
     CompressionCodec codec = null;
     CompressionType compressionType = CompressionType.NONE;
     if (getCompressOutput(job)) {

src/mapred/org/apache/hadoop/mapred/SequenceFileAsBinaryOutputFormat.java

@@ -130,13 +130,10 @@ public class SequenceFileAsBinaryOutputFormat
              getRecordWriter(FileSystem ignored, JobConf job,
                              String name, Progressable progress)
     throws IOException {
-
-    Path outputPath = getWorkOutputPath(job);
-    FileSystem fs = outputPath.getFileSystem(job);
-    if (!fs.exists(outputPath)) {
-      throw new IOException("Output directory doesnt exist");
-    }
-    Path file = new Path(outputPath, name);
+    // get the path of the temporary output file 
+    Path file = FileOutputFormat.getTaskOutputPath(job, name);
+    
+    FileSystem fs = file.getFileSystem(job);
     CompressionCodec codec = null;
     CompressionType compressionType = CompressionType.NONE;
     if (getCompressOutput(job)) {

src/mapred/org/apache/hadoop/mapred/SequenceFileOutputFormat.java

@@ -39,13 +39,10 @@ public class SequenceFileOutputFormat <K,V> extends FileOutputFormat<K, V> {
                                           FileSystem ignored, JobConf job,
                                           String name, Progressable progress)
     throws IOException {
-
-    Path outputPath = getWorkOutputPath(job);
-    FileSystem fs = outputPath.getFileSystem(job);
-    if (!fs.exists(outputPath)) {
-      throw new IOException("Output directory doesnt exist");
-    }
-    Path file = new Path(outputPath, name);
+    // get the path of the temporary output file 
+    Path file = FileOutputFormat.getTaskOutputPath(job, name);
+    
+    FileSystem fs = file.getFileSystem(job);
     CompressionCodec codec = null;
     CompressionType compressionType = CompressionType.NONE;
     if (getCompressOutput(job)) {

src/mapred/org/apache/hadoop/mapred/Task.java

@@ -493,6 +493,9 @@ abstract class Task implements Writable, Configurable {
                       > 0) {
                 shouldBePromoted = true;
               }
+            } else {
+              LOG.info(getTaskID() + ": No outputs to promote from " + 
+                       taskOutputPath);
             }
           }
         } catch (IOException ioe) {

src/mapred/org/apache/hadoop/mapred/TaskTracker.java

@@ -1458,22 +1458,6 @@ public class TaskTracker
       localJobConf.set("mapred.task.id", task.getTaskID().toString());
       keepFailedTaskFiles = localJobConf.getKeepFailedTaskFiles();
 
-      // create _taskid directory in output path temporary directory.
-      Path outputPath = FileOutputFormat.getOutputPath(localJobConf);
-      if (outputPath != null) {
-        Path jobTmpDir = new Path(outputPath, MRConstants.TEMP_DIR_NAME);
-        FileSystem fs = jobTmpDir.getFileSystem(localJobConf);
-        if (fs.exists(jobTmpDir)) {
-          Path taskTmpDir = new Path(jobTmpDir, "_" + task.getTaskID());
-          if (!fs.mkdirs(taskTmpDir)) {
-            throw new IOException("Mkdirs failed to create " 
-                                 + taskTmpDir.toString());
-          }
-        } else {
-          throw new IOException("The directory " + jobTmpDir.toString()
-                                 + " doesnt exist "); 
-        }
-      }
       task.localizeConfiguration(localJobConf);
       
       List<String[]> staticResolutions = NetUtils.getAllStaticResolutions();

src/mapred/org/apache/hadoop/mapred/TextOutputFormat.java

@@ -108,16 +108,13 @@ public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {
                                                   String name,
                                                   Progressable progress)
     throws IOException {
-
-    String keyValueSeparator = job.get("mapred.textoutputformat.separator", "\t");
-    Path dir = getWorkOutputPath(job);
-    FileSystem fs = dir.getFileSystem(job);
-    if (!fs.exists(dir)) {
-      throw new IOException("Output directory doesnt exist");
-    }
     boolean isCompressed = getCompressOutput(job);
+    String keyValueSeparator = job.get("mapred.textoutputformat.separator", 
+                                       "\t");
     if (!isCompressed) {
-      FSDataOutputStream fileOut = fs.create(new Path(dir, name), progress);
+      Path file = FileOutputFormat.getTaskOutputPath(job, name);
+      FileSystem fs = file.getFileSystem(job);
+      FSDataOutputStream fileOut = fs.create(file, progress);
       return new LineRecordWriter<K, V>(fileOut, keyValueSeparator);
     } else {
       Class<? extends CompressionCodec> codecClass =
@@ -126,8 +123,11 @@ public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {
       CompressionCodec codec = (CompressionCodec)
         ReflectionUtils.newInstance(codecClass, job);
       // build the filename including the extension
-      Path filename = new Path(dir, name + codec.getDefaultExtension());
-      FSDataOutputStream fileOut = fs.create(filename, progress);
+      Path file = 
+        FileOutputFormat.getTaskOutputPath(job, 
+                                           name + codec.getDefaultExtension());
+      FileSystem fs = file.getFileSystem(job);
+      FSDataOutputStream fileOut = fs.create(file, progress);
       return new LineRecordWriter<K, V>(new DataOutputStream
                                         (codec.createOutputStream(fileOut)),
                                         keyValueSeparator);

src/test/org/apache/hadoop/mapred/TestMultipleTextOutputFormat.java

@@ -39,7 +39,10 @@ public class TestMultipleTextOutputFormat extends TestCase {
   }
 
   private static Path workDir = 
-    new Path(new Path(System.getProperty("test.build.data", "."), "data"), 
+    new Path(new Path(
+                      new Path(System.getProperty("test.build.data", "."), 
+                               "data"), 
+                      MRConstants.TEMP_DIR_NAME), 
              "TestMultipleTextOutputFormat");
 
   private static void writeData(RecordWriter<Text, Text> rw) throws IOException {
@@ -81,6 +84,7 @@ public class TestMultipleTextOutputFormat extends TestCase {
   
   public void testFormat() throws Exception {
     JobConf job = new JobConf();
+    FileOutputFormat.setOutputPath(job, workDir.getParent().getParent());
     FileOutputFormat.setWorkOutputPath(job, workDir);
     FileSystem fs = workDir.getFileSystem(job);
     if (!fs.mkdirs(workDir)) {

src/test/org/apache/hadoop/mapred/TestSequenceFileAsBinaryOutputFormat.java

@@ -38,7 +38,11 @@ public class TestSequenceFileAsBinaryOutputFormat extends TestCase {
   public void testBinary() throws IOException {
     JobConf job = new JobConf();
     FileSystem fs = FileSystem.getLocal(job);
-    Path dir = new Path(System.getProperty("test.build.data",".") + "/mapred");
+    
+    Path dir = 
+      new Path(new Path(new Path(System.getProperty("test.build.data",".")), 
+                        MRConstants.TEMP_DIR_NAME),
+               "mapred");
     Path file = new Path(dir, "testbinary.seq");
     Random r = new Random();
     long seed = r.nextLong();
@@ -49,6 +53,7 @@ public class TestSequenceFileAsBinaryOutputFormat extends TestCase {
       fail("Failed to create output directory");
     }
 
+    FileOutputFormat.setOutputPath(job, dir.getParent().getParent());
     FileOutputFormat.setWorkOutputPath(job, dir);
 
     SequenceFileAsBinaryOutputFormat.setSequenceFileOutputKeyClass(job, 

src/test/org/apache/hadoop/mapred/TestTextOutputFormat.java

@@ -36,13 +36,17 @@ public class TestTextOutputFormat extends TestCase {
     }
   }
 
-  private static Path workDir =
-    new Path(new Path(System.getProperty("test.build.data", "."), "data"),
+  private static Path workDir = 
+    new Path(new Path(
+                      new Path(System.getProperty("test.build.data", "."), 
+                               "data"), 
+                      MRConstants.TEMP_DIR_NAME), 
              "TestTextOutputFormat");
 
   @SuppressWarnings("unchecked")
   public void testFormat() throws Exception {
     JobConf job = new JobConf();
+    FileOutputFormat.setOutputPath(job, workDir.getParent().getParent());
     FileOutputFormat.setWorkOutputPath(job, workDir);
     FileSystem fs = workDir.getFileSystem(job);
     if (!fs.mkdirs(workDir)) {
@@ -94,6 +98,7 @@ public class TestTextOutputFormat extends TestCase {
     JobConf job = new JobConf();
     String separator = "\u0001";
     job.set("mapred.textoutputformat.separator", separator);
+    FileOutputFormat.setOutputPath(job, workDir.getParent().getParent());
     FileOutputFormat.setWorkOutputPath(job, workDir);
     FileSystem fs = workDir.getFileSystem(job);
     if (!fs.mkdirs(workDir)) {