HADOOP-619. Extend InputFormatBase to accept individual files and glob patterns as MapReduce inputs. Contributed by Sanjay.

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

CHANGES.txt

@@ -200,6 +200,10 @@ Trunk (unreleased changes)
     hadoop-${version}-core.jar so that it can be more easily
     identified.  (Nigel Daley via cutting)
 
+57. HADOOP-619.  Extend InputFormatBase to accept individual files and
+    glob patterns as MapReduce inputs, not just directories.  Also
+    change contrib/streaming to use this.  (Sanjay Dahia via cutting)
+
 
 Release 0.9.2 - 2006-12-15
 

src/contrib/streaming/src/java/org/apache/hadoop/streaming/StreamInputFormat.java

@@ -21,20 +21,13 @@ package org.apache.hadoop.streaming;
 import java.io.*;
 import java.lang.reflect.*;
 import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 
 import org.apache.commons.logging.*;
 
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.PathFilter;
 import org.apache.hadoop.fs.FSDataInputStream;
 
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
-
 import org.apache.hadoop.mapred.*;
 
 /** An input format that performs globbing on DFS paths and
@@ -48,10 +41,6 @@ public class StreamInputFormat extends InputFormatBase {
 
   protected static final Log LOG = LogFactory.getLog(StreamInputFormat.class.getName());
 
-  /** This implementation always returns true. */
-  public void validateInput(JobConf job) throws IOException {
-  }
-
   static boolean isGzippedInput(JobConf job) {
     String val = job.get(StreamBaseRecordReader.CONF_NS + "compression");
     return "gzip".equals(val);
@@ -79,55 +68,6 @@ public class StreamInputFormat extends InputFormatBase {
     return (FileSplit[]) splits.toArray(new FileSplit[splits.size()]);
   }
 
-  protected Path[] listPaths(JobConf job) throws IOException {
-    Path[] globs = job.getInputPaths();
-    ArrayList list = new ArrayList();
-    int dsup = globs.length;
-    for (int d = 0; d < dsup; d++) {
-      String leafName = globs[d].getName();
-      LOG.info("StreamInputFormat: globs[" + d + "] leafName = " + leafName);
-      Path[] paths;
-      Path dir;
-      FileSystem fs = globs[d].getFileSystem(job);
-      PathFilter filter = new GlobFilter(fs, leafName);
-      dir = new Path(globs[d].getParent().toString());
-      if (dir == null) dir = new Path(".");
-      paths = fs.listPaths(dir, filter);
-      list.addAll(Arrays.asList(paths));
-    }
-    return (Path[]) list.toArray(new Path[] {});
-  }
-
-  class GlobFilter implements PathFilter {
-
-    public GlobFilter(FileSystem fs, String glob) {
-      fs_ = fs;
-      pat_ = Pattern.compile(globToRegexp(glob));
-    }
-
-    String globToRegexp(String glob) {
-      String re = glob;
-      re = re.replaceAll("\\.", "\\\\.");
-      re = re.replaceAll("\\+", "\\\\+");
-      re = re.replaceAll("\\*", ".*");
-      re = re.replaceAll("\\?", ".");
-      LOG.info("globToRegexp: |" + glob + "|  ->  |" + re + "|");
-      return re;
-    }
-
-    public boolean accept(Path pathname) {
-      boolean acc = !fs_.isChecksumFile(pathname);
-      if (acc) {
-        acc = pat_.matcher(pathname.getName()).matches();
-      }
-      LOG.info("matches " + pat_ + ", " + pathname + " = " + acc);
-      return acc;
-    }
-
-    Pattern pat_;
-    FileSystem fs_;
-  }
-
   public RecordReader getRecordReader(final InputSplit genericSplit, 
                                       JobConf job,
                                       Reporter reporter) throws IOException {

src/java/org/apache/hadoop/fs/FileSystem.java

@@ -652,8 +652,7 @@ public abstract class FileSystem extends Configured {
         throws IOException {
       Path [] parents = new Path[1];
       int level = 0;
-      
-      String filename = filePattern.toString();
+      String filename = filePattern.toUri().getPath();
       if("".equals(filename) || Path.SEPARATOR.equals(filename)) {
         parents[0] = filePattern;
         return parents;

src/java/org/apache/hadoop/fs/Path.java

@@ -32,6 +32,7 @@ public class Path implements Comparable {
 
   /** The directory separator, a slash. */
   public static final String SEPARATOR = "/";
+  public static final char SEPARATOR_CHAR = '/';
   
   static final boolean WINDOWS
     = System.getProperty("os.name").startsWith("Windows");

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

@@ -22,12 +22,14 @@ import java.io.FileNotFoundException;
 import java.io.IOException;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import org.apache.commons.logging.*;
 
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
 
 /** A base class for {@link InputFormat}. */
 public abstract class InputFormatBase implements InputFormat {
@@ -38,7 +40,12 @@ public abstract class InputFormatBase implements InputFormat {
   private static final double SPLIT_SLOP = 1.1;   // 10% slop
 
   private long minSplitSize = 1;
-
+  private static final PathFilter hiddenFileFilter = new PathFilter(){
+    public boolean accept( Path p ){
+      String name = p.getName(); 
+      return !name.startsWith("_") && !name.startsWith("."); 
+    }
+  }; 
   protected void setMinSplitSize(long minSplitSize) {
     this.minSplitSize = minSplitSize;
   }
@@ -63,39 +70,23 @@ public abstract class InputFormatBase implements InputFormat {
    * Subclasses may override to, e.g., select only files matching a regular
    * expression. 
    * 
-   * <p>Property <code>mapred.input.subdir</code>, if set, names a subdirectory
-   * that is appended to all input dirs specified by job, and if the given fs
-   * lists those too, each is added to the returned array of Path.
-   *
    * @param job the job to list input paths for
    * @return array of Path objects
    * @throws IOException if zero items.
    */
   protected Path[] listPaths(JobConf job)
     throws IOException {
-    String subdir = job.get("mapred.input.subdir");
     Path[] dirs = job.getInputPaths();
     if (dirs.length == 0) {
-      throw new IOException("No input directories specified in job");
+      throw new IOException("No input paths specified in job");
     }
-    ArrayList result = new ArrayList();
-    for (int i = 0; i < dirs.length; i++) {
-      FileSystem fs = dirs[i].getFileSystem(job);
-      Path[] dir = fs.listPaths(dirs[i]);
-      if (dir != null) {
-        for (int j = 0; j < dir.length; j++) {
-          Path file = dir[j];
-          if (subdir != null) {
-            Path[] subFiles = fs.listPaths(new Path(file, subdir));
-            if (subFiles != null) {
-              for (int k = 0; k < subFiles.length; k++) {
-                result.add(fs.makeQualified(subFiles[k]));
-              }
-            }
-          } else {
-            result.add(fs.makeQualified(file));
-          }
-        }
+    List<Path> result = new ArrayList(); 
+    for (Path p: dirs) {
+      FileSystem fs = p.getFileSystem(job); 
+      Path[] matches =
+        fs.listPaths(fs.globPaths(p, hiddenFileFilter),hiddenFileFilter);
+      for (Path match: matches) {
+        result.add(fs.makeQualified(match));
       }
     }
 
@@ -104,22 +95,50 @@ public abstract class InputFormatBase implements InputFormat {
 
   public void validateInput(JobConf job) throws IOException {
     Path[] inputDirs = job.getInputPaths();
+    if (inputDirs.length == 0) {
+      throw new IOException("No input paths specified in input"); 
+    }
+    
     List<IOException> result = new ArrayList();
-    for(int i=0; i < inputDirs.length; ++i) {
-      FileSystem fs = inputDirs[i].getFileSystem(job);
-      if (!fs.exists(inputDirs[i])) {
-        result.add(new FileNotFoundException("Input directory " + 
-                                             inputDirs[i] + 
-                                             " doesn't exist."));
-      } else if (!fs.isDirectory(inputDirs[i])) {
-        result.add(new InvalidFileTypeException
-                     ("Invalid input path, expecting directory : " + 
-                      inputDirs[i]));
+    int totalFiles = 0; 
+    for (Path p: inputDirs) {
+      FileSystem fs = p.getFileSystem(job);
+      if (fs.exists(p)) {
+        // make sure all paths are files to avoid exception
+        // while generating splits
+        for (Path subPath : fs.listPaths(p, hiddenFileFilter)) {
+          FileSystem subFS = subPath.getFileSystem(job); 
+          if (!subFS.isFile(subPath)) {
+            result.add(new IOException(
+                "Input path is not a file : " + subPath)); 
+          } else {
+            totalFiles++; 
+          }
+        }
+      } else {
+        Path [] paths = fs.globPaths(p, hiddenFileFilter); 
+        if (paths.length == 0) {
+          result.add(
+            new IOException("Input Pattern " + p + " matches 0 files")); 
+        } else {
+          // validate globbed paths 
+          for (Path gPath : paths) {
+            FileSystem gPathFS = gPath.getFileSystem(job); 
+            if (!gPathFS.exists(gPath)) {
+              result.add(
+                new FileNotFoundException(
+                    "Input path doesnt exist : " + gPath)); 
+            }
+          }
+          totalFiles += paths.length ; 
+        }
       }
     }
     if (!result.isEmpty()) {
       throw new InvalidInputException(result);
     }
+    // send output to client. 
+    LOG.info("Total input paths to process : " + totalFiles); 
   }
 
   /** Splits files returned by {@link #listPaths(JobConf)} when
@@ -146,7 +165,7 @@ public abstract class InputFormatBase implements InputFormat {
       Path file = files[i];
       FileSystem fs = file.getFileSystem(job);
       long length = fs.getLength(file);
-      if (isSplitable(fs, file)) {
+      if (isSplitable(fs, file)) { 
         long blockSize = fs.getBlockSize(file);
         long splitSize = computeSplitSize(goalSize, minSize, blockSize);