HADOOP-1247. Add support to contrib/streaming for aggregate package. Contributed by Runping.

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

CHANGES.txt

@@ -297,6 +297,9 @@ Trunk (unreleased changes)
 88. HADOOP-1272.  Extract inner classes from FSNamesystem into separate 
     classes.  (Dhruba Borthakur via tomwhite)
 
+89. HADOOP-1247.  Add support to contrib/streaming for aggregate
+    package, formerly called Abacus.  (Runping Qi via cutting)
+
 
 Release 0.12.3 - 2007-04-06
 

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

@@ -26,13 +26,10 @@ import java.net.URISyntaxException;
 import java.net.URLEncoder;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.ListIterator;
 import java.util.Map;
-import java.util.Set;
 import java.util.TreeMap;
 import java.util.TreeSet;
 
@@ -49,10 +46,11 @@ import org.apache.commons.cli2.validation.InvalidArgumentException;
 import org.apache.commons.cli2.validation.Validator;
 import org.apache.commons.logging.*;
 
+import org.apache.hadoop.mapred.lib.aggregate.ValueAggregatorCombiner;
+import org.apache.hadoop.mapred.lib.aggregate.ValueAggregatorReducer;
 import org.apache.hadoop.conf.Configuration;
 
 import org.apache.hadoop.io.Text;
-import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 
 import org.apache.hadoop.mapred.FileAlreadyExistsException;
@@ -66,7 +64,7 @@ import org.apache.hadoop.mapred.SequenceFileAsTextInputFormat;
 import org.apache.hadoop.mapred.TextOutputFormat;
 import org.apache.hadoop.filecache.*;
 import org.apache.hadoop.util.*;
-import org.apache.log4j.helpers.OptionConverter;
+
 /** All the client-side work happens here.
  * (Jar packaging, MapRed job submission and monitoring)
  * @author Michel Tourn
@@ -213,7 +211,6 @@ public class StreamJob {
       verbose_ =  cmdLine.hasOption("-verbose");
       detailedUsage_ = cmdLine.hasOption("-info");
       debug_ = cmdLine.hasOption("-debug")? debug_ + 1 : debug_;
-      inputTagged_ = cmdLine.hasOption("-inputtagged"); 
       
       inputSpecs_.addAll(cmdLine.getValues("-input"));
       output_ = (String) cmdLine.getValue("-output"); 
@@ -709,19 +706,14 @@ public class StreamJob {
     if (inReaderSpec_ == null && inputFormatSpec_ == null) {
       fmt = KeyValueTextInputFormat.class;
     } else if (inputFormatSpec_ != null) {
-      if ((inputFormatSpec_.compareToIgnoreCase("KeyValueTextInputFormat") == 0)
-          || (inputFormatSpec_
-              .compareToIgnoreCase("org.apache.hadoop.mapred.KeyValueTextInputFormat") == 0)) {
+      if (inputFormatSpec_.equals(KeyValueTextInputFormat.class.getName())
+          || inputFormatSpec_.equals(KeyValueTextInputFormat.class.getCanonicalName())) {
         fmt = KeyValueTextInputFormat.class;
-      } else if ((inputFormatSpec_
-                  .compareToIgnoreCase("SequenceFileInputFormat") == 0)
-                 || (inputFormatSpec_
-                     .compareToIgnoreCase("org.apache.hadoop.mapred.SequenceFileInputFormat") == 0)) {
+      } else if (inputFormatSpec_.equals(SequenceFileInputFormat.class.getName())
+          || inputFormatSpec_.equals(SequenceFileInputFormat.class.getCanonicalName())) {
         fmt = SequenceFileInputFormat.class;
-      } else if ((inputFormatSpec_
-                  .compareToIgnoreCase("SequenceFileToLineInputFormat") == 0)
-                 || (inputFormatSpec_
-                     .compareToIgnoreCase("org.apache.hadoop.mapred.SequenceFileToLineInputFormat") == 0)) {
+      } else if (inputFormatSpec_.equals(SequenceFileAsTextInputFormat.class.getName())
+          || inputFormatSpec_.equals(SequenceFileAsTextInputFormat.class.getCanonicalName())) {
         fmt = SequenceFileAsTextInputFormat.class;
       } else {
         c = StreamUtil.goodClassOrNull(inputFormatSpec_, defaultPackage);
@@ -774,12 +766,19 @@ public class StreamJob {
     reducerNone_ = false;
     if (redCmd_ != null) {
       reducerNone_ = redCmd_.equals(REDUCE_NONE);
-      c = StreamUtil.goodClassOrNull(redCmd_, defaultPackage);
-      if (c != null) {
-        jobConf_.setReducerClass(c);
+      if (redCmd_.compareToIgnoreCase("aggregate") == 0) {
+        jobConf_.setReducerClass(ValueAggregatorReducer.class);
+        jobConf_.setCombinerClass(ValueAggregatorCombiner.class);
       } else {
-        jobConf_.setReducerClass(PipeReducer.class);
-        jobConf_.set("stream.reduce.streamprocessor", URLEncoder.encode(redCmd_, "UTF-8"));
+
+        c = StreamUtil.goodClassOrNull(redCmd_, defaultPackage);
+        if (c != null) {
+          jobConf_.setReducerClass(c);
+        } else {
+          jobConf_.setReducerClass(PipeReducer.class);
+          jobConf_.set("stream.reduce.streamprocessor", URLEncoder.encode(
+              redCmd_, "UTF-8"));
+        }
       }
     }
 

src/contrib/streaming/src/test/org/apache/hadoop/streaming/StreamAggregate.java

@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.streaming;
+
+import java.io.*;
+
+import org.apache.hadoop.streaming.Environment;
+
+/** 
+    Used to test the usage of external applications without adding
+    platform-specific dependencies.
+ */
+public class StreamAggregate extends TrApp
+{
+
+  public StreamAggregate()
+  {
+    super('.', ' ');
+  }
+
+  public void go() throws IOException
+  {
+    testParentJobConfToEnvVars();
+    BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
+    String line;
+
+    while ((line = in.readLine()) != null) {
+      String [] words = line.split(" ");
+      for (int i = 0; i< words.length; i++) {
+        String out = "LongValueSum:" + words[i].trim() + "\t" + "1";
+        System.out.println(out);
+      }
+    }
+  }
+
+  public static void main(String[] args) throws IOException
+  {
+    TrApp app = new StreamAggregate();
+    app.go();
+  }
+}

src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreamAggregate.java

@@ -0,0 +1,116 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.streaming;
+
+import junit.framework.TestCase;
+import java.io.*;
+import java.util.*;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+/**
+ * This class tests hadoopStreaming in MapReduce local mode.
+ * It uses Hadoop Aggregate to count the numbers of word occurrences 
+ * in the input.
+ */
+public class TestStreamAggregate extends TestCase
+{
+  protected File INPUT_FILE = new File("stream_aggregate_input.txt");
+  protected File OUTPUT_DIR = new File("stream_aggregate_out");
+  protected String input = "roses are red\nviolets are blue\nbunnies are pink\n";
+  // map parses input lines and generates count entries for each word.
+  protected String map = StreamUtil.makeJavaCommand(StreamAggregate.class, new String[]{".", "\\n"});
+  // Use the aggregate combine, reducei to aggregate the counts
+  protected String outputExpect = "are\t3\nblue\t1\nbunnies\t1\npink\t1\nred\t1\nroses\t1\nviolets\t1\n";
+
+  private StreamJob job;
+
+  public TestStreamAggregate() throws IOException
+  {
+    UtilTest utilTest = new UtilTest(getClass().getName());
+    utilTest.checkUserDir();
+    utilTest.redirectIfAntJunit();
+  }
+
+  protected void createInput() throws IOException
+  {
+    DataOutputStream out = new DataOutputStream(
+                                                new FileOutputStream(INPUT_FILE.getAbsoluteFile()));
+    out.write(input.getBytes("UTF-8"));
+    out.close();
+  }
+
+  protected String[] genArgs() {
+    return new String[] {
+      "-input", INPUT_FILE.getAbsolutePath(),
+      "-output", OUTPUT_DIR.getAbsolutePath(),
+      "-mapper", map,
+      "-reducer", "aggregate",
+      //"-verbose",
+      //"-jobconf", "stream.debug=set"
+      "-jobconf", "keep.failed.task.files=true",
+      "-jobconf", "stream.tmpdir="+System.getProperty("test.build.data","/tmp")
+    };
+  }
+  
+  public void testCommandLine()
+  {
+    try {
+      try {
+        OUTPUT_DIR.getAbsoluteFile().delete();
+      } catch (Exception e) {
+      }
+
+      createInput();
+      boolean mayExit = false;
+
+      // During tests, the default Configuration will use a local mapred
+      // So don't specify -config or -cluster
+      job = new StreamJob(genArgs(), mayExit);      
+      job.go();
+      File outFile = new File(OUTPUT_DIR, "part-00000").getAbsoluteFile();
+      String output = StreamUtil.slurp(outFile);
+      outFile.delete();
+      System.err.println("outEx1=" + outputExpect);
+      System.err.println("  out1=" + output);
+      assertEquals(outputExpect, output);
+    } catch(Exception e) {
+      failTrace(e);
+    } finally {
+      File outFileCRC = new File(OUTPUT_DIR, ".part-00000.crc").getAbsoluteFile();
+      INPUT_FILE.delete();
+      outFileCRC.delete();
+      OUTPUT_DIR.getAbsoluteFile().delete();
+    }
+  }
+
+  private void failTrace(Exception e)
+  {
+    StringWriter sw = new StringWriter();
+    e.printStackTrace(new PrintWriter(sw));
+    fail(sw.toString());
+  }
+
+  public static void main(String[]args) throws Exception
+  {
+    new TestStreaming().testCommandLine();
+  }
+
+}