HADOOP-71. Allow Text and SequenceFile Map/Reduce inputs from non-default

filesystems.


git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@564741 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -39,6 +39,9 @@ Trunk (unreleased changes)
     HADOOP-1463.  HDFS report correct usage statistics for disk space
     used by HDFS.  (Hairong Kuang via dhruba)
 
+    HADOOP-71.  Allow Text and SequenceFile Map/Reduce inputs from non-default 
+    filesystems. (omalley)
+
   IMPROVEMENTS
 
     HADOOP-1651.  Improve progress reporting.

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

@@ -69,7 +69,7 @@ public class LineRecordReader implements RecordReader {
     final CompressionCodec codec = compressionCodecs.getCodec(file);
 
     // open the file and seek to the start of the split
-    FileSystem fs = FileSystem.get(job);
+    FileSystem fs = file.getFileSystem(job);
     FSDataInputStream fileIn = fs.open(split.getPath());
     InputStream in = fileIn;
     boolean skipFirstLine = false;

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

@@ -210,14 +210,11 @@ class MapTask extends Task {
 
     private Reporter reporter = null;
 
-    private JobConf job;
-
     public DirectMapOutputCollector(TaskUmbilicalProtocol umbilical,
         JobConf job, Reporter reporter) throws IOException {
-      this.job = job;
       this.reporter = reporter;
       String finalName = getOutputName(getPartition());
-      FileSystem fs = FileSystem.get(this.job);
+      FileSystem fs = FileSystem.get(job);
 
       out = job.getOutputFormat().getRecordWriter(fs, job, finalName, reporter);
     }

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

@@ -20,10 +20,11 @@ package org.apache.hadoop.mapred;
 
 import java.io.IOException;
 
-import org.apache.hadoop.fs.FileSystem;
 
-import org.apache.hadoop.io.*;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.*;
 import org.apache.hadoop.util.ReflectionUtils;
 
 /** An {@link RecordReader} for {@link SequenceFile}s. */
@@ -36,8 +37,9 @@ public class SequenceFileRecordReader implements RecordReader {
 
   public SequenceFileRecordReader(Configuration conf, FileSplit split)
     throws IOException {
-    FileSystem fs = FileSystem.get(conf);
-    this.in = new SequenceFile.Reader(fs, split.getPath(), conf);
+    Path path = split.getPath();
+    FileSystem fs = path.getFileSystem(conf);
+    this.in = new SequenceFile.Reader(fs, path, conf);
     this.end = split.getStart() + split.getLength();
     this.conf = conf;
 

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

@@ -180,7 +180,8 @@ abstract class Task implements Writable, Configurable {
   private Path getTaskOutputPath(JobConf conf) {
     Path p = new Path(conf.getOutputPath(), ("_" + taskId));
     try {
-      return p.makeQualified(FileSystem.get(conf));
+      FileSystem fs = p.getFileSystem(conf);
+      return p.makeQualified(fs);
     } catch (IOException ie) {
       LOG.warn(StringUtils.stringifyException(ie));
       return p;
@@ -398,21 +399,23 @@ abstract class Task implements Writable, Configurable {
    * @throws IOException
    */
   void saveTaskOutput() throws IOException {
-    FileSystem fs = FileSystem.get(conf);
 
-    if (taskOutputPath != null && fs.exists(taskOutputPath)) {
-      Path jobOutputPath = taskOutputPath.getParent();
+    if (taskOutputPath != null) {
+      FileSystem fs = taskOutputPath.getFileSystem(conf);
+      if (fs.exists(taskOutputPath)) {
+        Path jobOutputPath = taskOutputPath.getParent();
 
-      // Move the task outputs to their final place
-      moveTaskOutputs(fs, jobOutputPath, taskOutputPath);
+        // Move the task outputs to their final place
+        moveTaskOutputs(fs, jobOutputPath, taskOutputPath);
 
-      // Delete the temporary task-specific output directory
-      if (!fs.delete(taskOutputPath)) {
-        LOG.info("Failed to delete the temporary output directory of task: " + 
-                getTaskId() + " - " + taskOutputPath);
+        // Delete the temporary task-specific output directory
+        if (!fs.delete(taskOutputPath)) {
+          LOG.info("Failed to delete the temporary output directory of task: " + 
+                  getTaskId() + " - " + taskOutputPath);
+        }
+        
+        LOG.info("Saved output of task '" + getTaskId() + "' to " + jobOutputPath);
       }
-      
-      LOG.info("Saved output of task '" + getTaskId() + "' to " + jobOutputPath);
     }
   }
   
@@ -447,13 +450,14 @@ abstract class Task implements Writable, Configurable {
    * @throws IOException
    */
   void discardTaskOutput() throws IOException {
-    FileSystem fs = FileSystem.get(conf);
-
-    if (taskOutputPath != null && fs.exists(taskOutputPath)) {
-      // Delete the temporary task-specific output directory
-      FileUtil.fullyDelete(fs, taskOutputPath);
-      LOG.info("Discarded output of task '" + getTaskId() + "' - " 
-              + taskOutputPath);
+    if (taskOutputPath != null) {
+      FileSystem fs = taskOutputPath.getFileSystem(conf);
+      if (fs.exists(taskOutputPath)) {
+        // Delete the temporary task-specific output directory
+        FileUtil.fullyDelete(fs, taskOutputPath);
+        LOG.info("Discarded output of task '" + getTaskId() + "' - " 
+                + taskOutputPath);
+      }
     }
   }
 

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

@@ -51,18 +51,19 @@ public class TestMiniMRWithDFS extends TestCase {
     }
   }
   public static TestResult launchWordCount(JobConf conf,
+                                           Path inDir,
+                                           Path outDir,
                                            String input,
                                            int numMaps,
                                            int numReduces) throws IOException {
-    final Path inDir = new Path("/testing/wc/input");
-    final Path outDir = new Path("/testing/wc/output");
-    FileSystem fs = FileSystem.get(conf);
-    fs.delete(outDir);
-    if (!fs.mkdirs(inDir)) {
+    FileSystem inFs = inDir.getFileSystem(conf);
+    FileSystem outFs = outDir.getFileSystem(conf);
+    outFs.delete(outDir);
+    if (!inFs.mkdirs(inDir)) {
       throw new IOException("Mkdirs failed to create " + inDir.toString());
     }
     {
-      DataOutputStream file = fs.create(new Path(inDir, "part-0"));
+      DataOutputStream file = inFs.create(new Path(inDir, "part-0"));
       file.writeBytes(input);
       file.close();
     }
@@ -177,7 +178,9 @@ public class TestMiniMRWithDFS extends TestCase {
       // Keeping tasks that match this pattern
       jobConf.setKeepTaskFilesPattern("task_[^_]*_[0-9]*_m_000001_.*");
       TestResult result;
-      result = launchWordCount(jobConf, 
+      final Path inDir = new Path("/testing/wc/input");
+      final Path outDir = new Path("/testing/wc/output");
+      result = launchWordCount(jobConf, inDir, outDir,
                                "The quick brown fox\nhas many silly\n" + 
                                "red fox sox\n",
                                3, 1);
@@ -188,8 +191,25 @@ public class TestMiniMRWithDFS extends TestCase {
       checkTaskDirectories(mr, new String[]{jobid}, new String[]{taskid});
       // test with maps=0
       jobConf = mr.createJobConf();
-      result = launchWordCount(jobConf, "owen is oom", 0, 1);
+      result = launchWordCount(jobConf, inDir, outDir, "owen is oom", 0, 1);
       assertEquals("is\t1\noom\t1\nowen\t1\n", result.output);
+      // Run a job with input and output going to localfs even though the 
+      // default fs is hdfs.
+      {
+        FileSystem localfs = FileSystem.getLocal(jobConf);
+        String TEST_ROOT_DIR =
+          new File(System.getProperty("test.build.data","/tmp"))
+          .toString().replace(' ', '+');
+        Path localIn = localfs.makeQualified
+                          (new Path(TEST_ROOT_DIR + "/local/in"));
+        Path localOut = localfs.makeQualified
+                          (new Path(TEST_ROOT_DIR + "/local/out"));
+        result = launchWordCount(jobConf, localIn, localOut,
+                                 "all your base belong to us", 1, 1);
+        assertEquals("all\t1\nbase\t1\nbelong\t1\nto\t1\nus\t1\nyour\t1\n", 
+                     result.output);
+        assertTrue("outputs on localfs", localfs.exists(localOut));
+      }
     } finally {
       if (fileSys != null) { fileSys.close(); }
       if (dfs != null) { dfs.shutdown(); }