HADOOP-779. Fix contrib/streaming to work correctly with gzipped input. Contributed by Hairong.

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

CHANGES.txt

@@ -5,6 +5,9 @@ Trunk (unreleased changes)
  1. HADOOP-780. Use ReflectionUtils to instantiate key and value
     objects. (ab)
 
+ 2. HADOOP-779. Fix contrib/streaming to work correctly with gzipped
+    input files.  (Hairong Kuang via cutting)
+
 
 Release 0.9.0 - 2006-12-01
 

src/contrib/build-contrib.xml

@@ -103,7 +103,7 @@
      srcdir="${src.test}"
      includes="**/*.java"
      destdir="${build.test}"
-     debug="${debug}">
+     debug="${javac.debug}">
       <classpath refid="test.classpath"/>
     </javac>
   </target>

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

@@ -45,7 +45,7 @@ public class StreamLineRecordReader extends StreamBaseRecordReader {
     super(in, split, reporter, job, fs);
     gzipped_ = StreamInputFormat.isGzippedInput(job);
     if (gzipped_) {
-      din_ = new DataInputStream(new GZIPInputStream(in_));
+      din_ = new BufferedInputStream( (new GZIPInputStream(in_) ) );
     } else {
       din_ = in_;
     }
@@ -88,40 +88,24 @@ public class StreamLineRecordReader extends StreamBaseRecordReader {
     Text tValue = (Text) value;
     byte[] line;
 
-    while (true) {
-      if (gzipped_) {
-        // figure EOS from readLine
-      } else {
-        long pos = in_.getPos();
-        if (pos >= end_) return false;
-      }
-
-      line = UTF8ByteArrayUtils.readLine((InputStream) in_);
-      if (line == null) return false;
-      try {
-        Text.validateUTF8(line);
-      } catch (MalformedInputException m) {
-        System.err.println("line=" + line + "|" + new Text(line));
-        System.out.flush();
-      }
-      try {
-        int tab = UTF8ByteArrayUtils.findTab(line);
-        if (tab == -1) {
-          tKey.set(line);
-          tValue.set("");
-        } else {
-          UTF8ByteArrayUtils.splitKeyVal(line, tKey, tValue, tab);
-        }
-        break;
-      } catch (MalformedInputException e) {
-        LOG.warn(StringUtils.stringifyException(e));
-      }
+    if ( !gzipped_  ) {
+      long pos = in_.getPos();
+      if (pos >= end_) return false;
+    }
+    
+    line = UTF8ByteArrayUtils.readLine((InputStream) din_);
+    if (line == null) return false;
+    int tab = UTF8ByteArrayUtils.findTab(line);
+    if (tab == -1) {
+      tKey.set(line);
+      tValue.set("");
+    } else {
+      UTF8ByteArrayUtils.splitKeyVal(line, tKey, tValue, tab);
     }
     numRecStats(line, 0, line.length);
     return true;
   }
 
   boolean gzipped_;
-  GZIPInputStream zin_;
-  DataInputStream din_; // GZIP or plain  
+  InputStream din_; // GZIP or plain  
 }

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

@@ -0,0 +1,59 @@
+/**
+ * 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.FileOutputStream;
+import java.io.IOException;
+import java.util.zip.GZIPOutputStream;
+
+/**
+ * This class tests gzip input streaming in MapReduce local mode.
+ */
+public class TestGzipInput extends TestStreaming
+{
+
+  public TestGzipInput() throws IOException {
+  }
+  
+  protected void createInput() throws IOException
+  {
+    GZIPOutputStream out = new GZIPOutputStream(
+        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,
+        "-combiner", combine,
+        "-reducer", reduce,
+        "-jobconf", "stream.recordreader.compression=gzip"
+    };
+    
+  }
+
+  public static void main(String[]args) throws Exception
+  {
+    new TestGzipInput().testCommandLine();
+  }
+
+}

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

@@ -33,17 +33,17 @@ public class TestStreaming extends TestCase
 
   // "map" command: grep -E (red|green|blue)
   // reduce command: uniq
-  String INPUT_FILE = "input.txt";
-  String OUTPUT_DIR = "out";
-  String input = "roses.are.red\nviolets.are.blue\nbunnies.are.pink\n";
+  protected File INPUT_FILE = new File("input.txt");
+  protected File OUTPUT_DIR = new File("out");
+  protected String input = "roses.are.red\nviolets.are.blue\nbunnies.are.pink\n";
   // map behaves like "/usr/bin/tr . \\n"; (split words into lines)
-  String map = StreamUtil.makeJavaCommand(TrApp.class, new String[]{".", "\\n"});
+  protected String map = StreamUtil.makeJavaCommand(TrApp.class, new String[]{".", "\\n"});
   // combine, reduce behave like /usr/bin/uniq. But also prepend lines with C, R.
-  String combine  = StreamUtil.makeJavaCommand(UniqApp.class, new String[]{"C"});
-  String reduce = StreamUtil.makeJavaCommand(UniqApp.class, new String[]{"R"});
-  String outputExpect = "RCare\t\nRCblue\t\nRCbunnies\t\nRCpink\t\nRCred\t\nRCroses\t\nRCviolets\t\n";
+  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";
 
-  StreamJob job;
+  private StreamJob job;
 
   public TestStreaming() throws IOException
   {
@@ -52,14 +52,27 @@ public class TestStreaming extends TestCase
     utilTest.redirectIfAntJunit();
   }
 
-  void createInput() throws IOException
+  protected void createInput() throws IOException
   {
-    String path = new File(".", INPUT_FILE).getAbsolutePath();// needed from junit forked vm
-    DataOutputStream out = new DataOutputStream(new FileOutputStream(path));
+    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,
+        "-combiner", combine,
+        "-reducer", reduce,
+        //"-verbose",
+        //"-jobconf", "stream.debug=set"
+        "-jobconf", "keep.failed.task.files=true"
+        };
+  }
+  
   public void testCommandLine()
   {
     try {
@@ -68,30 +81,23 @@ public class TestStreaming extends TestCase
 
       // During tests, the default Configuration will use a local mapred
       // So don't specify -config or -cluster
-      String argv[] = new String[] {
-          "-input", INPUT_FILE,
-          "-output", OUTPUT_DIR,
-          "-mapper", map,
-          "-combiner", combine,
-          "-reducer", reduce,
-          //"-verbose",
-          //"-jobconf", "stream.debug=set"
-          "-jobconf", "keep.failed.task.files=true",
-      };
-      job = new StreamJob(argv, mayExit);      
+      job = new StreamJob(genArgs(), mayExit);      
       job.go();
-      File outFile = new File(".", OUTPUT_DIR + "/part-00000").getAbsoluteFile();
+      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 {
+      INPUT_FILE.delete();
+      OUTPUT_DIR.delete();
     }
   }
 
-  void failTrace(Exception e)
+  private void failTrace(Exception e)
   {
     StringWriter sw = new StringWriter();
     e.printStackTrace(new PrintWriter(sw));