hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cutt...@apache.org
Subject svn commit: r534970 - in /lucene/hadoop/trunk: ./ src/contrib/streaming/src/java/org/apache/hadoop/streaming/ src/contrib/streaming/src/test/org/apache/hadoop/streaming/
Date Thu, 03 May 2007 19:33:28 GMT
Author: cutting
Date: Thu May  3 12:33:27 2007
New Revision: 534970

URL: http://svn.apache.org/viewvc?view=rev&rev=534970
Log:
HADOOP-1315.  Clean up contrib/streaming, switching it to use more core classes and removing
unused classes.  Contributed by Runping.

Removed:
    lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/CompoundDirSpec.java
    lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/MergerInputFormat.java
    lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/MuxOutputFormat.java
    lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeCombiner.java
    lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/TupleInputFormat.java
Modified:
    lucene/hadoop/trunk/CHANGES.txt
    lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeMapRed.java
    lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeMapper.java
    lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeReducer.java
    lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/StreamJob.java
    lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/StreamUtil.java
    lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/UTF8ByteArrayUtils.java
    lucene/hadoop/trunk/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreaming.java
    lucene/hadoop/trunk/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TrApp.java

Modified: lucene/hadoop/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/CHANGES.txt?view=diff&rev=534970&r1=534969&r2=534970
==============================================================================
--- lucene/hadoop/trunk/CHANGES.txt (original)
+++ lucene/hadoop/trunk/CHANGES.txt Thu May  3 12:33:27 2007
@@ -316,6 +316,9 @@
     that killed the heartbeat monitoring thread.
     (Dhruba Borthakur via cutting)
 
+94. HADOOP-1315.  Clean up contrib/streaming, switching it to use core
+    classes more and removing unused code.  (Runping Qi via cutting)
+
 
 Release 0.12.3 - 2007-04-06
 

Modified: lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeMapRed.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeMapRed.java?view=diff&rev=534970&r1=534969&r2=534970
==============================================================================
--- lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeMapRed.java
(original)
+++ lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeMapRed.java
Thu May  3 12:33:27 2007
@@ -19,8 +19,6 @@
 package org.apache.hadoop.streaming;
 
 import java.io.*;
-import java.net.Socket;
-import java.net.URI;
 import java.nio.charset.CharacterCodingException;
 import java.io.IOException;
 import java.util.Date;
@@ -29,14 +27,11 @@
 import java.util.Arrays;
 import java.util.ArrayList;
 import java.util.Properties;
-import java.util.regex.*;
 
 import org.apache.commons.logging.*;
 
 import org.apache.hadoop.fs.FileUtil;
-import org.apache.hadoop.mapred.FileSplit;
 import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.PhasedFileSystem;
 import org.apache.hadoop.mapred.Reporter;
 import org.apache.hadoop.mapred.OutputCollector;
 import org.apache.hadoop.util.StringUtils;
@@ -45,7 +40,6 @@
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.io.Writable;
 
-import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.FileSystem;
 
 /** Shared functionality for PipeMapper, PipeReducer.
@@ -60,60 +54,12 @@
    */
   abstract String getPipeCommand(JobConf job);
 
-  /*
-   */
-  abstract String getKeyColPropName();
-
   abstract char getFieldSeparator();
   
   abstract int getNumOfKeyFields();
-  
-  /** Write output as side-effect files rather than as map outputs.
-      This is useful to do "Map" tasks rather than "MapReduce" tasks. */
-  boolean getUseSideEffect() {
-    return false;
-  }
 
   abstract boolean getDoPipe();
 
-  /**
-   * @returns how many TABS before the end of the key part
-   * usually: 1 or "ALL"
-   * used for tool output of both Map and Reduce
-   * configured via tool's argv: splitKeyVal=ALL or 1..
-   * although it is interpreted here, not by tool
-   */
-  int getKeyColsFromPipeCommand(String cmd) {
-    String key = getKeyColPropName();
-    Pattern kcPat = Pattern.compile(".*" + key + "=([^\\s]*).*");
-    Matcher match = kcPat.matcher(cmd);
-    String kc;
-    if (!match.matches()) {
-      kc = null;
-    } else {
-      kc = match.group(1);
-    }
-
-    int cols;
-    if (kc == null) {
-      // default value is 1 and the Stream applications could instead
-      // add/remove the \t separator on lines to get the same effect as value 0, 1, ALL
-      cols = 1;
-    } else if (kc.equals("ALL")) {
-      cols = ALL_COLS;
-    } else {
-      try {
-        cols = Integer.parseInt(kc);
-      } catch (NumberFormatException nf) {
-        cols = Integer.MAX_VALUE;
-      }
-    }
-
-    System.out.println("getKeyColsFromPipeCommand:" + key + " parse:" + cols + " from cmd="
+ cmd);
-
-    return cols;
-  }
-
   final static int OUTSIDE = 1;
   final static int SINGLEQ = 2;
   final static int DOUBLEQ = 3;
@@ -164,54 +110,15 @@
     return (String[]) argList.toArray(new String[0]);
   }
 
-  OutputStream getURIOutputStream(URI uri, boolean allowSocket) throws IOException {
-    final String SOCKET = "socket";
-    if (uri.getScheme().equals(SOCKET)) {
-      if (!allowSocket) {
-        throw new IOException(SOCKET + " not allowed on outputstream " + uri);
-      }
-      final Socket sock = new Socket(uri.getHost(), uri.getPort());
-      OutputStream out = new FilterOutputStream(sock.getOutputStream()) {
-          public void close() throws IOException {
-            sock.close();
-            super.close();
-          }
-        };
-      return out;
-    } else {
-      // a FSDataOutputStreamm, localFS or HDFS.
-      // localFS file may be set up as a FIFO.
-      return sideFs_.create(new Path(uri.getSchemeSpecificPart()));
-    }
-  }
-
-  String getSideEffectFileName() {
-    FileSplit split = StreamUtil.getCurrentSplit(job_);
-    return new String(split.getPath().getName() + "-" + split.getStart() + 
-                      "-" + split.getLength());
-  }
-
   public void configure(JobConf job) {
     try {
       String argv = getPipeCommand(job);
 
-      keyCols_ = getKeyColsFromPipeCommand(argv);
-
-      debug_ = (job.get("stream.debug") != null);
-      if (debug_) {
-        System.out.println("PipeMapRed: stream.debug=true");
-      }
-
       joinDelay_ = job.getLong("stream.joindelay.milli", 0);
 
       job_ = job;
       fs_ = FileSystem.get(job_);
-      if (job_.getBoolean("stream.sideoutput.localfs", false)) {
-        //sideFs_ = new LocalFileSystem(job_);
-        sideFs_ = FileSystem.getLocal(job_);
-      } else {
-        sideFs_ = fs_;
-      }
+
       String mapOutputFieldSeparator = job_.get("stream.map.output.field.separator", "\t");
       String reduceOutputFieldSeparator = job_.get("stream.reduce.output.field.separator",
"\t");
       this.mapOutputFieldSeparator = mapOutputFieldSeparator.charAt(0);
@@ -219,22 +126,12 @@
       this.numOfMapOutputKeyFields = job_.getInt("stream.num.map.output.key.fields", 1);
       this.numOfReduceOutputKeyFields = job_.getInt("stream.num.reduce.output.key.fields",
1);
       
-      if (debug_) {
-        System.out.println("kind   :" + this.getClass());
-        System.out.println("split  :" + StreamUtil.getCurrentSplit(job_));
-        System.out.println("fs     :" + fs_.toString());
-        System.out.println("sideFs :" + sideFs_.toString());
-      }
 
       doPipe_ = getDoPipe();
       if (!doPipe_) return;
 
       setStreamJobDetails(job);
-      setStreamProperties();
-
-      if (debugFailEarly_) {
-        throw new RuntimeException("debugFailEarly_");
-      }
+      
       String[] argvSplit = splitArgs(argv);
       String prog = argvSplit[0];
       File currentDir = new File(".").getAbsoluteFile();
@@ -245,39 +142,6 @@
         FileUtil.chmod(new File(jobCacheDir, prog).toString(), "a+x");
       }
 
-      if (job_.getInputValueClass().equals(BytesWritable.class)) {
-        // TODO expose as separate config:
-        // job or semistandard inputformat property
-        optUseKey_ = false;
-      }
-
-      optSideEffect_ = getUseSideEffect();
-
-      if (optSideEffect_) {
-        // during work: use a completely unique filename to avoid HDFS namespace conflicts
-        // after work: rename to a filename that depends only on the workload (the FileSplit)
-        //   it's a friendly name and in case of reexecution it will clobber. 
-        // reexecution can be due to: other job, failed task and speculative task
-        // See StreamJob.setOutputSpec(): if reducerNone_ aka optSideEffect then: 
-        // client has renamed outputPath and saved the argv's original output path as:
-        if (useSingleSideOutputURI_) {
-          finalOutputURI = new URI(sideOutputURI_);
-          sideEffectPathFinal_ = null; // in-place, no renaming to final
-        } else {
-          sideFs_ = new PhasedFileSystem(sideFs_, job);
-          String sideOutputPath = job_.get("stream.sideoutput.dir"); // was: job_.getOutputPath()

-          String fileName = getSideEffectFileName(); // see HADOOP-444 for rationale
-          sideEffectPathFinal_ = new Path(sideOutputPath, fileName);
-          finalOutputURI = new URI(sideEffectPathFinal_.toString()); // implicit dfs: 
-        }
-        // apply default scheme
-        if (finalOutputURI.getScheme() == null) {
-          finalOutputURI = new URI("file", finalOutputURI.getSchemeSpecificPart(), null);
-        }
-        boolean allowSocket = useSingleSideOutputURI_;
-        sideEffectOut_ = getURIOutputStream(finalOutputURI, allowSocket);
-      }
-
       // 
       // argvSplit[0]:
       // An absolute path should be a preexisting valid path on all TaskTrackers
@@ -295,8 +159,6 @@
         f = null;
       }
       logprintln("PipeMapRed exec " + Arrays.asList(argvSplit));
-      logprintln("sideEffectURI_=" + finalOutputURI);
-
       Environment childEnv = (Environment) StreamUtil.env().clone();
       addJobConfToEnvironment(job_, childEnv);
       addEnvironment(childEnv, job_.get("stream.addenvironment"));
@@ -327,34 +189,6 @@
       logprintln("JobConf set minRecWrittenToEnableSkip_ =" + minRecWrittenToEnableSkip_);
     }
     taskId_ = StreamUtil.getTaskInfo(job_);
-    debugFailEarly_ = isDebugFail("early");
-    debugFailDuring_ = isDebugFail("during");
-    debugFailLate_ = isDebugFail("late");
-
-    sideOutputURI_ = job_.get("stream.sideoutput.uri");
-    useSingleSideOutputURI_ = (sideOutputURI_ != null);
-  }
-
-  boolean isDebugFail(String kind) {
-    String execidlist = job_.get("stream.debugfail.reexec." + kind);
-    if (execidlist == null) {
-      return false;
-    }
-    String[] e = execidlist.split(",");
-    for (int i = 0; i < e.length; i++) {
-      int ei = Integer.parseInt(e[i]);
-      if (taskId_.execid == ei) {
-        return true;
-      }
-    }
-    return false;
-  }
-
-  void setStreamProperties() {
-    String s = System.getProperty("stream.port");
-    if (s != null) {
-      reportPortPlusOne_ = Integer.parseInt(s);
-    }
   }
 
   void logStackTrace(Exception e) {
@@ -442,7 +276,6 @@
     if (log_ != null) {
       StreamUtil.exec("/bin/rm " + LOGNAME, log_);
     }
-    // TODO socket-based aggregator (in JobTrackerInfoServer)
   }
 
   void startOutputThreads(OutputCollector output, Reporter reporter) {
@@ -474,14 +307,7 @@
    * @throws IOException
    */
   void splitKeyVal(byte[] line, Text key, Text val) throws IOException {
-    int pos = -1;
-    if (keyCols_ != ALL_COLS) {
-      pos = UTF8ByteArrayUtils.findNthByte(line, (byte)this.getFieldSeparator(), this.getNumOfKeyFields());
-    }
-    LOG.info("FieldSeparator: " + this.getFieldSeparator());
-    LOG.info("NumOfKeyFields: " + this.getNumOfKeyFields());
-    LOG.info("Line: " + new String (line));
-    LOG.info("Pos: " + pos);
+    int pos = UTF8ByteArrayUtils.findNthByte(line, (byte)this.getFieldSeparator(), this.getNumOfKeyFields());
     try {
       if (pos == -1) {
         key.set(line);
@@ -508,15 +334,10 @@
         Text val = new Text();
         // 3/4 Tool to Hadoop
         while ((answer = UTF8ByteArrayUtils.readLine((InputStream) clientIn_)) != null) {
-          // 4/4 Hadoop out
-          if (optSideEffect_) {
-            sideEffectOut_.write(answer);
-            sideEffectOut_.write('\n');
-            sideEffectOut_.flush();
-          } else {
-            splitKeyVal(answer, key, val);
-            output.collect(key, val);
-          }
+          
+          splitKeyVal(answer, key, val);
+          output.collect(key, val);
+          
           numRecWritten_++;
           long now = System.currentTimeMillis();
           if (now-lastStdoutReport > reporterOutDelay_) {
@@ -584,18 +405,6 @@
       } catch (IOException io) {
       }
       waitOutputThreads();
-      try {
-        if (optSideEffect_) {
-          logprintln("closing " + finalOutputURI);
-          if (sideEffectOut_ != null) sideEffectOut_.close();
-          logprintln("closed  " + finalOutputURI);
-          if (!useSingleSideOutputURI_) {
-            ((PhasedFileSystem)sideFs_).commit(); 
-          }
-        }
-      } catch (IOException io) {
-        io.printStackTrace();
-      }
       if (sim != null) sim.destroy();
     } catch (RuntimeException e) {
       logStackTrace(e);
@@ -692,18 +501,11 @@
   
   long minRecWrittenToEnableSkip_ = Long.MAX_VALUE;
 
-  int keyCols_;
-  final static int ALL_COLS = Integer.MAX_VALUE;
-
   long reporterOutDelay_ = 10*1000L; 
   long reporterErrDelay_ = 10*1000L; 
   long joinDelay_;
   JobConf job_;
   FileSystem fs_;
-  FileSystem sideFs_;
-
-  // generic MapRed parameters passed on by hadoopStreaming
-  int reportPortPlusOne_;
 
   boolean doPipe_;
   boolean debug_;
@@ -723,17 +525,6 @@
   String mapredKey_;
   int numExceptions_;
   StreamUtil.TaskId taskId_;
-
-  boolean optUseKey_ = true;
-
-  private boolean optSideEffect_;
-  private URI finalOutputURI;
-  private Path sideEffectPathFinal_;
-
-  private boolean useSingleSideOutputURI_;
-  private String sideOutputURI_;
-
-  private OutputStream sideEffectOut_;
 
   protected volatile Throwable outerrThreadsThrowable;
 

Modified: lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeMapper.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeMapper.java?view=diff&rev=534970&r1=534969&r2=534970
==============================================================================
--- lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeMapper.java
(original)
+++ lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeMapper.java
Thu May  3 12:33:27 2007
@@ -25,8 +25,10 @@
 import org.apache.hadoop.mapred.Mapper;
 import org.apache.hadoop.mapred.Reporter;
 import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.TextInputFormat;
 import org.apache.hadoop.util.StringUtils;
 
+import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.WritableComparable;
 import org.apache.hadoop.io.Writable;
 
@@ -36,6 +38,8 @@
  */
 public class PipeMapper extends PipeMapRed implements Mapper {
 
+  private boolean ignoreKey = false;
+  
   String getPipeCommand(JobConf job) {
     String str = job.get("stream.map.streamprocessor");
     if (str == null) {
@@ -50,17 +54,15 @@
     }
   }
 
-  String getKeyColPropName() {
-    return "mapKeyCols";
-  }
-
-  boolean getUseSideEffect() {
-    return StreamUtil.getUseMapSideEffect(job_);
-  }
-
   boolean getDoPipe() {
     return true;
   }
+  
+  public void configure(JobConf job) {
+    super.configure(job);
+    String inputFormatClassName = job.getClass("mapred.input.format.class", TextInputFormat.class).getCanonicalName();
+    this.ignoreKey = inputFormatClassName.equals(TextInputFormat.class.getCanonicalName());
+  }
 
   // Do NOT declare default constructor
   // (MapRed creates it reflectively)
@@ -86,7 +88,7 @@
 
       // 2/4 Hadoop to Tool
       if (numExceptions_ == 0) {
-        if (optUseKey_) {
+        if (!this.ignoreKey) {
           write(key);
           clientOut_.write('\t');
         }

Modified: lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeReducer.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeReducer.java?view=diff&rev=534970&r1=534969&r2=534970
==============================================================================
--- lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeReducer.java
(original)
+++ lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeReducer.java
Thu May  3 12:33:27 2007
@@ -57,10 +57,6 @@
     return (argv != null) && !StreamJob.REDUCE_NONE.equals(argv);
   }
 
-  String getKeyColPropName() {
-    return "reduceKeyCols";
-  }
-
   public void reduce(WritableComparable key, Iterator values, OutputCollector output,
                      Reporter reporter) throws IOException {
 

Modified: lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/StreamJob.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/StreamJob.java?view=diff&rev=534970&r1=534969&r2=534970
==============================================================================
--- lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/StreamJob.java
(original)
+++ lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/StreamJob.java
Thu May  3 12:33:27 2007
@@ -22,7 +22,6 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
-import java.net.URISyntaxException;
 import java.net.URLEncoder;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -41,7 +40,6 @@
 import org.apache.commons.cli2.option.PropertyOption;
 import org.apache.commons.cli2.resource.ResourceConstants;
 import org.apache.commons.cli2.util.HelpFormatter;
-import org.apache.commons.cli2.validation.FileValidator;
 import org.apache.commons.cli2.validation.InvalidArgumentException;
 import org.apache.commons.cli2.validation.Validator;
 import org.apache.commons.logging.*;
@@ -61,6 +59,7 @@
 import org.apache.hadoop.mapred.RunningJob;
 import org.apache.hadoop.mapred.SequenceFileInputFormat;
 import org.apache.hadoop.mapred.SequenceFileAsTextInputFormat;
+import org.apache.hadoop.mapred.TextInputFormat;
 import org.apache.hadoop.mapred.TextOutputFormat;
 import org.apache.hadoop.filecache.*;
 import org.apache.hadoop.util.*;
@@ -73,8 +72,7 @@
 
   protected static final Log LOG = LogFactory.getLog(StreamJob.class.getName());
   final static String REDUCE_NONE = "NONE";
-  private boolean reducerNone_;
-
+    
   /** -----------Streaming CLI Implementation  **/
   private DefaultOptionBuilder builder = 
     new DefaultOptionBuilder("-","-", false);
@@ -214,7 +212,6 @@
       
       inputSpecs_.addAll(cmdLine.getValues("-input"));
       output_ = (String) cmdLine.getValue("-output"); 
-      mapsideoutURI_ = (String) cmdLine.getValue("-mapsideoutput");
       
       mapCmd_ = (String)cmdLine.getValue("-mapper"); 
       comCmd_ = (String)cmdLine.getValue("-combiner"); 
@@ -450,20 +447,17 @@
       System.out.println("Options:");
       System.out.println("  -input    <path>     DFS input file(s) for the Map step");
       System.out.println("  -output   <path>     DFS output directory for the Reduce
step");
-      System.out.println("  -mapper   <cmd>      The streaming command to run");
-      System.out.println("  -combiner <cmd>      The streaming command to run");
-      System.out.println("  -reducer  <cmd>      The streaming command to run");
+      System.out.println("  -mapper   <cmd|JavaClassName>      The streaming command
to run");
+      System.out.println("  -combiner <JavaClassName> Combiner has to be a Java class");
+      System.out.println("  -reducer  <cmd|JavaClassName>      The streaming command
to run");
       System.out.println("  -file     <file>     File/dir to be shipped in the Job
jar file");
-      //Only advertise the standard way: [--config dir] in our launcher 
-      //System.out.println("  -cluster  <name>     Default uses hadoop-default.xml
and hadoop-site.xml");
-      //System.out.println("  -config   <file>     Optional. One or more paths to xml
config files");
       System.out.println("  -dfs    <h:p>|local  Optional. Override DFS configuration");
       System.out.println("  -jt     <h:p>|local  Optional. Override JobTracker configuration");
       System.out.println("  -additionalconfspec specfile  Optional.");
-      System.out.println("  -inputformat KeyValueTextInputFormat(default)|SequenceFileInputFormat|XmlTextInputFormat
 Optional.");
-      System.out.println("  -outputformat specfile  Optional.");
-      System.out.println("  -partitioner specfile  Optional.");
-      System.out.println("  -numReduceTasks specfile  Optional.");
+      System.out.println("  -inputformat TextInputFormat(default)|SequenceFileAsTextInputFormat|JavaClassName
Optional.");
+      System.out.println("  -outputformat TextOutputFormat(default)|JavaClassName  Optional.");
+      System.out.println("  -partitioner JavaClassName  Optional.");
+      System.out.println("  -numReduceTasks <num>  Optional.");
       System.out.println("  -inputreader <spec>  Optional.");
       System.out.println("  -jobconf  <n>=<v>    Optional. Add or override a
JobConf property");
       System.out.println("  -cmdenv   <n>=<v>    Optional. Pass env.var to streaming
commands");
@@ -478,10 +472,7 @@
     System.out.println("In -input: globbing on <path> is supported and can have multiple
-input");
     System.out.println("Default Map input format: a line is a record in UTF-8");
     System.out.println("  the key part ends at first TAB, the rest of the line is the value");
-    System.out.println("Custom Map input format: -inputreader package.MyRecordReader,n=v,n=v
");
-    System.out
-      .println("  comma-separated name-values can be specified to configure the InputFormat");
-    System.out.println("  Ex: -inputreader 'StreamXmlRecordReader,begin=<doc>,end=</doc>'");
+    System.out.println("Custom input format: -inputformat package.MyInputFormat ");
     System.out.println("Map output format, reduce input/output format:");
     System.out.println("  Format defined by what the mapper command outputs. Line-oriented");
     System.out.println();
@@ -489,34 +480,21 @@
     System.out.println("  working directory when the mapper and reducer are run.");
     System.out.println("  The location of this working directory is unspecified.");
     System.out.println();
-    //System.out.println("Use -cluster <name> to switch between \"local\" Hadoop and
one or more remote ");
-    //System.out.println("  Hadoop clusters. ");
-    //System.out.println("  The default is to use the normal hadoop-default.xml and hadoop-site.xml");
-    //System.out.println("  Else configuration will use $HADOOP_HOME/conf/hadoop-<name>.xml");
-    //System.out.println();
+    System.out.println("To set the number of reduce tasks (num. of output files):");
+    System.out.println("  -jobconf mapred.reduce.tasks=10");
     System.out.println("To skip the sort/combine/shuffle/sort/reduce step:");
-    System.out.println("  Use -reducer " + REDUCE_NONE);
+    System.out.println("  Use -numReduceTasks 0");
     System.out
       .println("  A Task's Map output then becomes a 'side-effect output' rather than a reduce
input");
     System.out
       .println("  This speeds up processing, This also feels more like \"in-place\" processing");
     System.out.println("  because the input filename and the map input order are preserved");
-    System.out.println("To specify a single side-effect output file");
-    System.out.println("    -mapsideoutput [file:/C:/win|file:/unix/|socket://host:port]");//-output
for side-effects will be soon deprecated
-    System.out.println("  If the jobtracker is local this is a local file");
-    System.out.println("  This currently requires -reducer NONE");
+    System.out.println("  This equivalent -reducer NONE");
     System.out.println();
-    System.out.println("To set the number of reduce tasks (num. of output files):");
-    System.out.println("  -jobconf mapred.reduce.tasks=10");
     System.out.println("To speed up the last reduces:");
     System.out.println("  -jobconf mapred.speculative.execution=true");
-    System.out.println("  Do not use this along -reducer " + REDUCE_NONE);
     System.out.println("To name the job (appears in the JobTracker Web UI):");
     System.out.println("  -jobconf mapred.job.name='My Job' ");
-    System.out.println("To specify that line-oriented input is in gzip format:");
-    System.out
-      .println("(at this time ALL input files must be gzipped and this is not recognized
based on file extension)");
-    System.out.println("   -jobconf stream.recordreader.compression=gzip ");
     System.out.println("To change the local temp directory:");
     System.out.println("  -jobconf dfs.data.dir=/tmp/dfs");
     System.out.println("  -jobconf stream.tmpdir=/tmp/streaming");
@@ -681,8 +659,6 @@
       config_.addFinalResource(new Path(pathName));
     }
 
-    testMerge_ = (-1 != userJobConfProps_.toString().indexOf("stream.testmerge"));
-
     // general MapRed job properties
     jobConf_ = new JobConf(config_);
     
@@ -695,25 +671,32 @@
     // (to resolve local vs. dfs drive letter differences) 
     // (mapred.working.dir will be lazily initialized ONCE and depends on FS)
     for (int i = 0; i < inputSpecs_.size(); i++) {
-      addInputSpec((String) inputSpecs_.get(i), i);
+      jobConf_.addInputPath(new Path(((String) inputSpecs_.get(i))));
     }
-    jobConf_.setBoolean("stream.inputtagged", inputTagged_);
     jobConf_.set("stream.numinputspecs", "" + inputSpecs_.size());
 
     String defaultPackage = this.getClass().getPackage().getName();
     Class c;
     Class fmt = null;
     if (inReaderSpec_ == null && inputFormatSpec_ == null) {
-      fmt = KeyValueTextInputFormat.class;
+      fmt = TextInputFormat.class;
     } else if (inputFormatSpec_ != null) {
-      if (inputFormatSpec_.equals(KeyValueTextInputFormat.class.getName())
-          || inputFormatSpec_.equals(KeyValueTextInputFormat.class.getCanonicalName())) {
-        fmt = KeyValueTextInputFormat.class;
-      } else if (inputFormatSpec_.equals(SequenceFileInputFormat.class.getName())
-          || inputFormatSpec_.equals(SequenceFileInputFormat.class.getCanonicalName())) {
-        fmt = SequenceFileInputFormat.class;
-      } else if (inputFormatSpec_.equals(SequenceFileAsTextInputFormat.class.getName())
-          || inputFormatSpec_.equals(SequenceFileAsTextInputFormat.class.getCanonicalName()))
{
+      if (inputFormatSpec_.equals(TextInputFormat.class.getName())
+          || inputFormatSpec_.equals(TextInputFormat.class.getCanonicalName())) {
+        fmt = TextInputFormat.class;
+      } else if (inputFormatSpec_.equals(KeyValueTextInputFormat.class
+          .getName())
+          || inputFormatSpec_.equals(KeyValueTextInputFormat.class
+              .getCanonicalName())) {
+      } else if (inputFormatSpec_.equals(SequenceFileInputFormat.class
+          .getName())
+          || inputFormatSpec_
+              .equals(org.apache.hadoop.mapred.SequenceFileInputFormat.class
+                  .getCanonicalName())) {
+      } else if (inputFormatSpec_.equals(SequenceFileAsTextInputFormat.class
+          .getName())
+          || inputFormatSpec_.equals(SequenceFileAsTextInputFormat.class
+              .getCanonicalName())) {
         fmt = SequenceFileAsTextInputFormat.class;
       } else {
         c = StreamUtil.goodClassOrNull(inputFormatSpec_, defaultPackage);
@@ -725,14 +708,7 @@
       }
     } 
     if (fmt == null) {
-      if (testMerge_ && false == hasSimpleInputSpecs_) {
-        // this ignores -inputreader
-        fmt = MergerInputFormat.class;
-      } else {
-        // need to keep this case to support custom -inputreader
-        // and their parameters ,n=v,n=v
-        fmt = StreamInputFormat.class;
-      }
+      fmt = StreamInputFormat.class;
     }
 
     jobConf_.setInputFormat(fmt);
@@ -757,13 +733,10 @@
       c = StreamUtil.goodClassOrNull(comCmd_, defaultPackage);
       if (c != null) {
         jobConf_.setCombinerClass(c);
-      } else {
-        jobConf_.setCombinerClass(PipeCombiner.class);
-        jobConf_.set("stream.combine.streamprocessor", URLEncoder.encode(comCmd_, "UTF-8"));
-      }
+      } 
     }
 
-    reducerNone_ = false;
+    boolean reducerNone_ = false;
     if (redCmd_ != null) {
       reducerNone_ = redCmd_.equals(REDUCE_NONE);
       if (redCmd_.compareToIgnoreCase("aggregate") == 0) {
@@ -801,9 +774,7 @@
     }
     
     setUserJobConfProps(false);
-    // output setup is done late so we can customize for reducerNone_
-    //jobConf_.setOutputDir(new File(output_));
-    setOutputSpec();
+    jobConf_.setOutputPath(new Path(output_));
     fmt = null;
     if (outputFormatSpec_!= null) {
       c = StreamUtil.goodClassOrNull(outputFormatSpec_, defaultPackage);
@@ -812,11 +783,7 @@
       } 
     }
     if (fmt == null) {
-      if (testMerge_) {
-        fmt = MuxOutputFormat.class;
-      } else {
-        fmt = TextOutputFormat.class;
-      }
+      fmt = TextOutputFormat.class;
     }
     jobConf_.setOutputFormat(fmt);
 
@@ -831,6 +798,9 @@
       int numReduceTasks = Integer.parseInt(numReduceTasksSpec_);
       jobConf_.setNumReduceTasks(numReduceTasks);
     }
+    if (reducerNone_) {
+      jobConf_.setNumReduceTasks(0);
+    }
     
     // last, allow user to override anything
     // (although typically used with properties we didn't touch)
@@ -880,78 +850,6 @@
     }
     msg("====");
   }
-  
-  /** InputSpec-s encode: a glob pattern x additional column files x additional joins */
-  protected void addInputSpec(String inSpec, int index) {
-    if (!testMerge_) {
-      jobConf_.addInputPath(new Path(inSpec));
-    } else {
-      CompoundDirSpec spec = new CompoundDirSpec(inSpec, true);
-      msg("Parsed -input:\n" + spec.toTableString());
-      if (index == 0) {
-        hasSimpleInputSpecs_ = (spec.paths_.length == 0);
-        msg("hasSimpleInputSpecs_=" + hasSimpleInputSpecs_);
-      }
-      String primary = spec.primarySpec();
-      if (!seenPrimary_.add(primary)) {
-        // this won't detect glob overlaps and noncanonical path variations
-        fail("Primary used in multiple -input spec: " + primary);
-      }
-      jobConf_.addInputPath(new Path(primary));
-      // during Job execution, will reparse into a CompoundDirSpec 
-      jobConf_.set("stream.inputspecs." + index, inSpec);
-    }
-  }
-
-  /** uses output_ and mapsideoutURI_ */
-  protected void setOutputSpec() throws IOException {
-    CompoundDirSpec spec = new CompoundDirSpec(output_, false);
-    msg("Parsed -output:\n" + spec.toTableString());
-    String primary = spec.primarySpec();
-    String channel0;
-    // TODO simplify cases, encapsulate in a StreamJobConf
-    if (!reducerNone_) {
-      channel0 = primary;
-    } else {
-      if (mapsideoutURI_ != null) {
-        // user can override in case this is in a difft filesystem..
-        try {
-          URI uri = new URI(mapsideoutURI_);
-          if (uri.getScheme() == null || uri.getScheme().equals("file")) { // || uri.getScheme().equals("hdfs")
-            if (!new Path(uri.getSchemeSpecificPart()).isAbsolute()) {
-              fail("Must be absolute: " + mapsideoutURI_);
-            }
-          } else if (uri.getScheme().equals("socket")) {
-            // ok
-          } else {
-            fail("Invalid scheme: " + uri.getScheme() + " for -mapsideoutput " + mapsideoutURI_);
-          }
-        } catch (URISyntaxException e) {
-          throw (IOException) new IOException().initCause(e);
-        }
-      }
-      // an empty reduce output named "part-00002" will go here and not collide.
-      channel0 = primary + ".NONE";
-      // the side-effect of the first split of an input named "part-00002" 
-      // will go in this directory
-      jobConf_.set("stream.sideoutput.dir", primary);
-      // oops if user overrides low-level this isn't set yet :-(
-      boolean localjt = StreamUtil.isLocalJobTracker(jobConf_);
-      // just a guess user may prefer remote..
-      jobConf_.setBoolean("stream.sideoutput.localfs", localjt);
-    }
-    // a path in fs.name.default filesystem
-    System.out.println(channel0);
-    System.out.println(new Path(channel0));
-    jobConf_.setOutputPath(new Path(channel0));
-    // will reparse remotely
-    jobConf_.set("stream.outputspec", output_);
-    if (null != mapsideoutURI_) {
-      // a path in "jobtracker's filesystem"
-      // overrides sideoutput.dir
-      jobConf_.set("stream.sideoutput.uri", mapsideoutURI_);
-    }
-  }
 
   protected String getJobTrackerHostPort() {
     return jobConf_.get("mapred.job.tracker");
@@ -1099,15 +997,12 @@
 
   // command-line arguments
   protected ArrayList inputSpecs_ = new ArrayList(); // <String>
-  protected boolean inputTagged_ = false;
   protected TreeSet seenPrimary_ = new TreeSet(); // <String>
   protected boolean hasSimpleInputSpecs_;
   protected ArrayList packageFiles_ = new ArrayList(); // <String>
   protected ArrayList shippedCanonFiles_ = new ArrayList(); // <String>
-  //protected ArrayList userJobConfProps_ = new ArrayList(); // <String> name=value
   protected TreeMap<String, String> userJobConfProps_ = new TreeMap<String, String>();

   protected String output_;
-  protected String mapsideoutURI_;
   protected String mapCmd_;
   protected String comCmd_;
   protected String redCmd_;
@@ -1124,8 +1019,6 @@
   protected String partitionerSpec_;
   protected String numReduceTasksSpec_;
   protected String additionalConfSpec_;
-
-  protected boolean testMerge_;
 
   // Use to communicate config to the external processes (ex env.var.HADOOP_USER)
   // encoding "a=b c=d"

Modified: lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/StreamUtil.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/StreamUtil.java?view=diff&rev=534970&r1=534969&r2=534970
==============================================================================
--- lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/StreamUtil.java
(original)
+++ lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/StreamUtil.java
Thu May  3 12:33:27 2007
@@ -453,20 +453,6 @@
     return res;
   }
 
-  static boolean getUseMapSideEffect(JobConf job) {
-    String reduce = job.get("stream.reduce.streamprocessor");
-    if (reduce == null) {
-      return false;
-    }
-    try {
-      reduce = URLDecoder.decode(reduce, "UTF-8");
-    } catch (UnsupportedEncodingException e) {
-      System.err.println("stream.reduce.streamprocessor in jobconf not found");
-      return false;
-    }
-    return StreamJob.REDUCE_NONE.equals(reduce);
-  }
-
   public static void touch(File file) throws IOException {
     file = file.getAbsoluteFile();
     FileOutputStream out = new FileOutputStream(file);

Modified: lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/UTF8ByteArrayUtils.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/UTF8ByteArrayUtils.java?view=diff&rev=534970&r1=534969&r2=534970
==============================================================================
--- lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/UTF8ByteArrayUtils.java
(original)
+++ lucene/hadoop/trunk/src/contrib/streaming/src/java/org/apache/hadoop/streaming/UTF8ByteArrayUtils.java
Thu May  3 12:33:27 2007
@@ -19,10 +19,8 @@
 package org.apache.hadoop.streaming;
 
 import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.io.PushbackInputStream;
 
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapred.LineRecordReader;

Modified: lucene/hadoop/trunk/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreaming.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreaming.java?view=diff&rev=534970&r1=534969&r2=534970
==============================================================================
--- lucene/hadoop/trunk/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreaming.java
(original)
+++ lucene/hadoop/trunk/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreaming.java
Thu May  3 12:33:27 2007
@@ -38,10 +38,11 @@
   protected String input = "roses.are.red\nviolets.are.blue\nbunnies.are.pink\n";
   // map behaves like "/usr/bin/tr . \\n"; (split words into lines)
   protected String map = StreamUtil.makeJavaCommand(TrApp.class, new String[]{".", "\\n"});
-  // combine, reduce behave like /usr/bin/uniq. But also prepend lines with C, R.
+  // reduce behave like /usr/bin/uniq. But also prepend lines with R.
+  // command-line combiner does not have any effect any more.
   protected String combine  = StreamUtil.makeJavaCommand(UniqApp.class, new String[]{"C"});
   protected String reduce = StreamUtil.makeJavaCommand(UniqApp.class, new String[]{"R"});
-  protected String outputExpect = "RCare\t\nRCblue\t\nRCbunnies\t\nRCpink\t\nRCred\t\nRCroses\t\nRCviolets\t\n";
+  protected String outputExpect = "Rare\t\nRblue\t\nRbunnies\t\nRpink\t\nRred\t\nRroses\t\nRviolets\t\n";
 
   private StreamJob job;
 

Modified: lucene/hadoop/trunk/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TrApp.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TrApp.java?view=diff&rev=534970&r1=534969&r2=534970
==============================================================================
--- lucene/hadoop/trunk/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TrApp.java
(original)
+++ lucene/hadoop/trunk/src/contrib/streaming/src/test/org/apache/hadoop/streaming/TrApp.java
Thu May  3 12:33:27 2007
@@ -41,7 +41,7 @@
     // test that some JobConf properties are exposed as expected     
     // Note the dots translated to underscore: 
     // property names have been escaped in PipeMapRed.safeEnvVarName()
-    expect("mapred_input_format_class", "org.apache.hadoop.mapred.KeyValueTextInputFormat");
+    expect("mapred_input_format_class", "org.apache.hadoop.mapred.TextInputFormat");
     expect("mapred_job_tracker", "local");
     //expect("mapred_local_dir", "build/test/mapred/local");
     expectDefined("mapred_local_dir");



Mime
View raw message