HADOOP-115. Permit reduce input and output types to differ. Contributed by Runping Qi.

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

CHANGES.txt

@@ -41,6 +41,10 @@ Trunk (unreleased)
 10. HADOOP-161. Add hashCode() method to DFS's Block.
     (Milind Bhandarkar via cutting)
 
+11. HADOOP-115. Map output types may now be specified.  These are also
+    used as reduce input types, thus permitting reduce input types to
+    differ from reduce output types.  (Runping Qi via cutting)
+
 
 Release 0.2.1 - 2006-05-12
 

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

@@ -269,14 +269,67 @@ public class JobConf extends Configuration {
   public Class getInputValueClass() {
     return getClass("mapred.input.value.class", UTF8.class, Writable.class);
   }
+
   public void setInputValueClass(Class theClass) {
     setClass("mapred.input.value.class", theClass, Writable.class);
   }
-
+  
+  /**
+   * Get the key class for the map output data. If it is not set, use the
+   * (final) output ket class This allows the map output key class to be
+   * different than the final output key class
+   * 
+   * @return map output key class
+   */
+  public Class getMapOutputKeyClass() {
+    Class retv = getClass("mapred.mapoutput.key.class", null,
+			  WritableComparable.class);
+    if (retv == null) {
+      retv = getOutputKeyClass();
+    }
+    return retv;
+  }
+  
+  /**
+   * Set the key class for the map output data. This allows the user to
+   * specify the map output key class to be different than the final output
+   * value class
+   */
+  public void setMapOutputKeyClass(Class theClass) {
+    setClass("mapred.mapoutput.key.class", theClass,
+             WritableComparable.class);
+  }
+  
+  /**
+   * Get the value class for the map output data. If it is not set, use the
+   * (final) output value class This allows the map output value class to be
+   * different than the final output value class
+   * 
+   * @return map output value class
+   */
+  public Class getMapOutputValueClass() {
+    Class retv = getClass("mapred.mapoutput.value.class", null,
+			  Writable.class);
+    if (retv == null) {
+      retv = getOutputValueClass();
+    }
+    return retv;
+  }
+  
+  /**
+   * Set the value class for the map output data. This allows the user to
+   * specify the map output value class to be different than the final output
+   * value class
+   */
+  public void setMapOutputValueClass(Class theClass) {
+    setClass("mapred.mapoutput.value.class", theClass, Writable.class);
+  }
+  
   public Class getOutputKeyClass() {
     return getClass("mapred.output.key.class",
                     LongWritable.class, WritableComparable.class);
   }
+  
   public void setOutputKeyClass(Class theClass) {
     setClass("mapred.output.key.class", theClass, WritableComparable.class);
   }
@@ -286,7 +339,7 @@ public class JobConf extends Configuration {
                               WritableComparator.class);
     if (theClass != null)
       return (WritableComparator)newInstance(theClass);
-    return WritableComparator.get(getOutputKeyClass());
+    return WritableComparator.get(getMapOutputKeyClass());
   }
 
   public void setOutputKeyComparatorClass(Class theClass) {
@@ -301,6 +354,7 @@ public class JobConf extends Configuration {
     setClass("mapred.output.value.class", theClass, Writable.class);
   }
 
+
   public Class getMapperClass() {
     return getClass("mapred.mapper.class", IdentityMapper.class, Mapper.class);
   }

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

@@ -38,8 +38,8 @@ public class MapFileOutputFormat extends OutputFormatBase {
 
     final MapFile.Writer out =
       new MapFile.Writer(fs, file.toString(),
-                         job.getOutputKeyClass(),
-                         job.getOutputValueClass(),
+                         job.getMapOutputKeyClass(),
+                         job.getMapOutputValueClass(),
                          job.getBoolean("mapred.output.compress", false));
 
     return new RecordWriter() {

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

@@ -77,8 +77,8 @@ class MapTask extends Task {
         outs[i] =
           new SequenceFile.Writer(FileSystem.getNamed("local", job),
                                   this.mapOutputFile.getOutputFile(getTaskId(), i),
-                                  job.getOutputKeyClass(),
-                                  job.getOutputValueClass());
+                                  job.getMapOutputKeyClass(),
+                                  job.getMapOutputValueClass());
       }
 
       final Partitioner partitioner =

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

@@ -173,8 +173,8 @@ class ReduceTask extends Task {
 
   public void run(JobConf job, final TaskUmbilicalProtocol umbilical)
     throws IOException {
-    Class keyClass = job.getOutputKeyClass();
-    Class valueClass = job.getOutputValueClass();
+    Class keyClass = job.getMapOutputKeyClass();
+    Class valueClass = job.getMapOutputValueClass();
     Reducer reducer = (Reducer)job.newInstance(job.getReducerClass());
     reducer.configure(job);
     FileSystem lfs = FileSystem.getNamed("local", job);