hadoop-mapreduce-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cdoug...@apache.org
Subject svn commit: r811582 - in /hadoop/mapreduce/trunk: ./ src/java/org/apache/hadoop/mapred/lib/ src/java/org/apache/hadoop/mapreduce/lib/output/ src/test/mapred/org/apache/hadoop/mapreduce/lib/output/
Date Sat, 05 Sep 2009 01:06:30 GMT
Author: cdouglas
Date: Sat Sep  5 01:06:30 2009
New Revision: 811582

URL: http://svn.apache.org/viewvc?rev=811582&view=rev
Log:
MAPREDUCE-370. Update MultipleOutputs to use the API, merge funcitonality
of MultipleOutputFormat. Contributed by Amareshwari Sriramadasu

Added:
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.java
    hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/lib/output/TestMRMultipleOutputs.java
      - copied, changed from r811565, hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/lib/TestMultipleOutputs.java
Modified:
    hadoop/mapreduce/trunk/CHANGES.txt
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleOutputFormat.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleOutputs.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleSequenceFileOutputFormat.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleTextOutputFormat.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.java

Modified: hadoop/mapreduce/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/CHANGES.txt?rev=811582&r1=811581&r2=811582&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/CHANGES.txt (original)
+++ hadoop/mapreduce/trunk/CHANGES.txt Sat Sep  5 01:06:30 2009
@@ -286,6 +286,9 @@
     MAPREDUCE-936. Allow a load difference for fairshare scheduler.
     (Zheng Shao via dhruba)
 
+    MAPREDUCE-370. Update MultipleOutputs to use the API, merge funcitonality
+    of MultipleOutputFormat. (Amareshwari Sriramadasu via cdouglas)
+
   BUG FIXES
 
     MAPREDUCE-878. Rename fair scheduler design doc to 

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleOutputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleOutputFormat.java?rev=811582&r1=811581&r2=811582&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleOutputFormat.java
(original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleOutputFormat.java
Sat Sep  5 01:06:30 2009
@@ -47,7 +47,10 @@
  * Case three: This class is used for a map only job. The job wants to use an
  * output file name that depends on both the keys and the input file name,
  * 
+ * @deprecated Use 
+ * {@link org.apache.hadoop.mapreduce.lib.output.MultipleOutputs} instead
  */
+@Deprecated
 public abstract class MultipleOutputFormat<K, V>
 extends FileOutputFormat<K, V> {
 

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleOutputs.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleOutputs.java?rev=811582&r1=811581&r2=811582&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleOutputs.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleOutputs.java Sat
Sep  5 01:06:30 2009
@@ -112,7 +112,10 @@
  *
  * }
  * </pre>
+ * @deprecated Use 
+ * {@link org.apache.hadoop.mapreduce.lib.output.MultipleOutputs} instead
  */
+@Deprecated
 public class MultipleOutputs {
 
   private static final String NAMED_OUTPUTS = "mo.namedOutputs";

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleSequenceFileOutputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleSequenceFileOutputFormat.java?rev=811582&r1=811581&r2=811582&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleSequenceFileOutputFormat.java
(original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleSequenceFileOutputFormat.java
Sat Sep  5 01:06:30 2009
@@ -29,7 +29,10 @@
 /**
  * This class extends the MultipleOutputFormat, allowing to write the output data 
  * to different output files in sequence file output format. 
+ * @deprecated Use 
+ * {@link org.apache.hadoop.mapreduce.lib.output.MultipleOutputs} instead
  */
+@Deprecated
 public class MultipleSequenceFileOutputFormat <K,V>
 extends MultipleOutputFormat<K, V> {
 

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleTextOutputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleTextOutputFormat.java?rev=811582&r1=811581&r2=811582&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleTextOutputFormat.java
(original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/lib/MultipleTextOutputFormat.java
Sat Sep  5 01:06:30 2009
@@ -29,7 +29,10 @@
 /**
  * This class extends the MultipleOutputFormat, allowing to write the output
  * data to different output files in Text output format.
+ * @deprecated Use 
+ * {@link org.apache.hadoop.mapreduce.lib.output.MultipleOutputs} instead
  */
+@Deprecated
 public class MultipleTextOutputFormat<K, V>
     extends MultipleOutputFormat<K, V> {
 

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.java?rev=811582&r1=811581&r2=811582&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.java
(original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.java
Sat Sep  5 01:06:30 2009
@@ -42,6 +42,8 @@
   /** Construct output file names so that, when an output directory listing is
    * sorted lexicographically, positions correspond to output partitions.*/
   private static final NumberFormat NUMBER_FORMAT = NumberFormat.getInstance();
+  protected static final String BASE_OUTPUT_NAME = "mapreduce.output.basename";
+  protected static final String PART = "part";
   static {
     NUMBER_FORMAT.setMinimumIntegerDigits(5);
     NUMBER_FORMAT.setGroupingUsed(false);
@@ -253,8 +255,22 @@
                                  String extension) throws IOException{
     FileOutputCommitter committer = 
       (FileOutputCommitter) getOutputCommitter(context);
-    return new Path(committer.getWorkPath(), getUniqueFile(context, "part", 
-                                                           extension));
+    return new Path(committer.getWorkPath(), getUniqueFile(context, 
+      getOutputName(context), extension));
+  }
+
+  /**
+   * Get the base output name for the output file.
+   */
+  protected static String getOutputName(JobContext job) {
+    return job.getConfiguration().get(BASE_OUTPUT_NAME, PART);
+  }
+
+  /**
+   * Set the base output name for output file to be created.
+   */
+  protected static void setOutputName(JobContext job, String name) {
+    job.getConfiguration().set(BASE_OUTPUT_NAME, name);
   }
 
   public synchronized 

Added: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.java?rev=811582&view=auto
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.java
(added)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.java
Sat Sep  5 01:06:30 2009
@@ -0,0 +1,444 @@
+/**
+ * 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.mapreduce.lib.output;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.*;
+import org.apache.hadoop.util.ReflectionUtils;
+
+import java.io.IOException;
+import java.util.*;
+
+/**
+ * The MultipleOutputs class simplifies writing output data 
+ * to multiple outputs
+ * 
+ * <p> 
+ * Case one: writing to additional outputs other than the job default output.
+ *
+ * Each additional output, or named output, may be configured with its own
+ * <code>OutputFormat</code>, with its own key class and with its own value
+ * class.
+ * 
+ * <p>
+ * Case two: to write data to different files provided by user
+ * </p>
+ * 
+ * <p>
+ * MultipleOutputs supports counters, by default they are disabled. The 
+ * counters group is the {@link MultipleOutputs} class name. The names of the 
+ * counters are the same as the output name. These count the number records 
+ * written to each output name.
+ * </p>
+ * 
+ * Usage pattern for job submission:
+ * <pre>
+ *
+ * Job job = new Job();
+ *
+ * FileInputFormat.setInputPath(job, inDir);
+ * FileOutputFormat.setOutputPath(job, outDir);
+ *
+ * job.setMapperClass(MOMap.class);
+ * job.setReducerClass(MOReduce.class);
+ * ...
+ *
+ * // Defines additional single text based output 'text' for the job
+ * MultipleOutputs.addNamedOutput(job, "text", TextOutputFormat.class,
+ * LongWritable.class, Text.class);
+ *
+ * // Defines additional sequence-file based output 'sequence' for the job
+ * MultipleOutputs.addNamedOutput(job, "seq",
+ *   SequenceFileOutputFormat.class,
+ *   LongWritable.class, Text.class);
+ * ...
+ *
+ * job.waitForCompletion(true);
+ * ...
+ * </pre>
+ * <p>
+ * Usage in Reducer:
+ * <pre>
+ * <K, V> String generateFileName(K k, V v) {
+ *   return k.toString() + "_" + v.toString();
+ * }
+ * 
+ * public class MOReduce extends
+ *   Reducer&lt;WritableComparable, Writable,WritableComparable, Writable&gt; {
+ * private MultipleOutputs mos;
+ * public void setup(Context context) {
+ * ...
+ * mos = new MultipleOutputs(context);
+ * }
+ *
+ * public void reduce(WritableComparable key, Iterator&lt;Writable&gt; values,
+ * Context context)
+ * throws IOException {
+ * ...
+ * mos.write("text", , key, new Text("Hello"));
+ * mos.write("seq", LongWritable(1), new Text("Bye"), "seq_a");
+ * mos.write("seq", LongWritable(2), key, new Text("Chau"), "seq_b");
+ * mos.write(key, new Text("value"), generateFileName(key, new Text("value")));
+ * ...
+ * }
+ *
+ * public void cleanup(Context) throws IOException {
+ * mos.close();
+ * ...
+ * }
+ *
+ * }
+ * </pre>
+ */
+public class MultipleOutputs<KEYOUT, VALUEOUT> {
+
+  private static final String MULTIPLE_OUTPUTS = "mapreduce.multipleoutputs";
+
+  private static final String MO_PREFIX = 
+    "mapreduce.multipleoutputs.namedOutput.";
+
+  private static final String FORMAT = ".format";
+  private static final String KEY = ".key";
+  private static final String VALUE = ".value";
+  private static final String COUNTERS_ENABLED = 
+    "mapreduce.multipleoutputs.counters";
+
+  /**
+   * Counters group used by the counters of MultipleOutputs.
+   */
+  private static final String COUNTERS_GROUP = MultipleOutputs.class.getName();
+
+  /**
+   * Checks if a named output name is valid token.
+   *
+   * @param namedOutput named output Name
+   * @throws IllegalArgumentException if the output name is not valid.
+   */
+  private static void checkTokenName(String namedOutput) {
+    if (namedOutput == null || namedOutput.length() == 0) {
+      throw new IllegalArgumentException(
+        "Name cannot be NULL or emtpy");
+    }
+    for (char ch : namedOutput.toCharArray()) {
+      if ((ch >= 'A') && (ch <= 'Z')) {
+        continue;
+      }
+      if ((ch >= 'a') && (ch <= 'z')) {
+        continue;
+      }
+      if ((ch >= '0') && (ch <= '9')) {
+        continue;
+      }
+      throw new IllegalArgumentException(
+        "Name cannot be have a '" + ch + "' char");
+    }
+  }
+
+  /**
+   * Checks if output name is valid.
+   *
+   * name cannot be the name used for the default output
+   * @param outputPath base output Name
+   * @throws IllegalArgumentException if the output name is not valid.
+   */
+  private static void checkBaseOutputPath(String outputPath) {
+    if (outputPath.equals(FileOutputFormat.PART)) {
+      throw new IllegalArgumentException("output name cannot be 'part'");
+    }
+  }
+  
+  /**
+   * Checks if a named output name is valid.
+   *
+   * @param namedOutput named output Name
+   * @throws IllegalArgumentException if the output name is not valid.
+   */
+  private static void checkNamedOutputName(JobContext job,
+      String namedOutput, boolean alreadyDefined) {
+    checkTokenName(namedOutput);
+    checkBaseOutputPath(namedOutput);
+    List<String> definedChannels = getNamedOutputsList(job);
+    if (alreadyDefined && definedChannels.contains(namedOutput)) {
+      throw new IllegalArgumentException("Named output '" + namedOutput +
+        "' already alreadyDefined");
+    } else if (!alreadyDefined && !definedChannels.contains(namedOutput)) {
+      throw new IllegalArgumentException("Named output '" + namedOutput +
+        "' not defined");
+    }
+  }
+
+  // Returns list of channel names.
+  private static List<String> getNamedOutputsList(JobContext job) {
+    List<String> names = new ArrayList<String>();
+    StringTokenizer st = new StringTokenizer(
+      job.getConfiguration().get(MULTIPLE_OUTPUTS, ""), " ");
+    while (st.hasMoreTokens()) {
+      names.add(st.nextToken());
+    }
+    return names;
+  }
+
+  // Returns the named output OutputFormat.
+  @SuppressWarnings("unchecked")
+  private static Class<? extends OutputFormat<?, ?>> getNamedOutputFormatClass(
+    JobContext job, String namedOutput) {
+    return (Class<? extends OutputFormat<?, ?>>)
+      job.getConfiguration().getClass(MO_PREFIX + namedOutput + FORMAT, null,
+      OutputFormat.class);
+  }
+
+  // Returns the key class for a named output.
+  private static Class<?> getNamedOutputKeyClass(JobContext job,
+                                                String namedOutput) {
+    return job.getConfiguration().getClass(MO_PREFIX + namedOutput + KEY, null,
+      WritableComparable.class);
+  }
+
+  // Returns the value class for a named output.
+  private static Class<? extends Writable> getNamedOutputValueClass(
+      JobContext job, String namedOutput) {
+    return job.getConfiguration().getClass(MO_PREFIX + namedOutput + VALUE,
+      null, Writable.class);
+  }
+
+  /**
+   * Adds a named output for the job.
+   * <p/>
+   *
+   * @param job               job to add the named output
+   * @param namedOutput       named output name, it has to be a word, letters
+   *                          and numbers only, cannot be the word 'part' as
+   *                          that is reserved for the default output.
+   * @param outputFormatClass OutputFormat class.
+   * @param keyClass          key class
+   * @param valueClass        value class
+   */
+  @SuppressWarnings("unchecked")
+  public static void addNamedOutput(Job job, String namedOutput,
+      Class<? extends OutputFormat> outputFormatClass,
+      Class<?> keyClass, Class<?> valueClass) {
+    checkNamedOutputName(job, namedOutput, true);
+    Configuration conf = job.getConfiguration();
+    conf.set(MULTIPLE_OUTPUTS,
+      conf.get(MULTIPLE_OUTPUTS, "") + " " + namedOutput);
+    conf.setClass(MO_PREFIX + namedOutput + FORMAT, outputFormatClass,
+      OutputFormat.class);
+    conf.setClass(MO_PREFIX + namedOutput + KEY, keyClass, Object.class);
+    conf.setClass(MO_PREFIX + namedOutput + VALUE, valueClass, Object.class);
+  }
+
+  /**
+   * Enables or disables counters for the named outputs.
+   * 
+   * The counters group is the {@link MultipleOutputs} class name.
+   * The names of the counters are the same as the named outputs. These
+   * counters count the number records written to each output name.
+   * By default these counters are disabled.
+   *
+   * @param job    job  to enable counters
+   * @param enabled indicates if the counters will be enabled or not.
+   */
+  public static void setCountersEnabled(Job job, boolean enabled) {
+    job.getConfiguration().setBoolean(COUNTERS_ENABLED, enabled);
+  }
+
+  /**
+   * Returns if the counters for the named outputs are enabled or not.
+   * By default these counters are disabled.
+   *
+   * @param job    the job 
+   * @return TRUE if the counters are enabled, FALSE if they are disabled.
+   */
+  public static boolean getCountersEnabled(JobContext job) {
+    return job.getConfiguration().getBoolean(COUNTERS_ENABLED, false);
+  }
+
+  /**
+   * Wraps RecordWriter to increment counters. 
+   */
+  @SuppressWarnings("unchecked")
+  private static class RecordWriterWithCounter extends RecordWriter {
+    private RecordWriter writer;
+    private String counterName;
+    private TaskInputOutputContext context;
+
+    public RecordWriterWithCounter(RecordWriter writer, String counterName,
+                                   TaskInputOutputContext context) {
+      this.writer = writer;
+      this.counterName = counterName;
+      this.context = context;
+    }
+
+    @SuppressWarnings({"unchecked"})
+    public void write(Object key, Object value) 
+        throws IOException, InterruptedException {
+      context.getCounter(COUNTERS_GROUP, counterName).increment(1);
+      writer.write(key, value);
+    }
+
+    public void close(TaskAttemptContext context) 
+        throws IOException, InterruptedException {
+      writer.close(context);
+    }
+  }
+
+  // instance code, to be used from Mapper/Reducer code
+
+  private TaskInputOutputContext<?, ?, KEYOUT, VALUEOUT> context;
+  private Set<String> namedOutputs;
+  private Map<String, RecordWriter<?, ?>> recordWriters;
+  private boolean countersEnabled;
+  
+  /**
+   * Creates and initializes multiple outputs support,
+   * it should be instantiated in the Mapper/Reducer setup method.
+   *
+   * @param context the TaskInputOutputContext object
+   */
+  public MultipleOutputs(
+      TaskInputOutputContext<?, ?, KEYOUT, VALUEOUT> context) {
+    this.context = context;
+    namedOutputs = Collections.unmodifiableSet(
+      new HashSet<String>(MultipleOutputs.getNamedOutputsList(context)));
+    recordWriters = new HashMap<String, RecordWriter<?, ?>>();
+    countersEnabled = getCountersEnabled(context);
+  }
+
+  /**
+   * Write key and value to the namedOutput.
+   *
+   * Output path is a unique file generated for the namedOutput.
+   * For example, {namedOutput}-(m|r)-{part-number}
+   * 
+   * @param namedOutput the named output name
+   * @param key         the key
+   * @param value       the value
+   */
+  @SuppressWarnings("unchecked")
+  public <K, V> void write(String namedOutput, K key, V value)
+      throws IOException, InterruptedException {
+    write(namedOutput, key, value, namedOutput);
+  }
+
+  /**
+   * Write key and value to baseOutputPath using the namedOutput.
+   * 
+   * @param namedOutput    the named output name
+   * @param key            the key
+   * @param value          the value
+   * @param baseOutputPath base-output path to write the record to.
+   * Note: Framework will generate unique filename for the baseOutputPath
+   */
+  @SuppressWarnings("unchecked")
+  public <K, V> void write(String namedOutput, K key, V value,
+      String baseOutputPath) throws IOException, InterruptedException {
+    checkNamedOutputName(context, namedOutput, false);
+    checkBaseOutputPath(baseOutputPath);
+    if (!namedOutputs.contains(namedOutput)) {
+      throw new IllegalArgumentException("Undefined named output '" +
+        namedOutput + "'");
+    }
+    TaskAttemptContext taskContext = getContext(namedOutput);
+    getRecordWriter(taskContext, baseOutputPath).write(key, value);
+  }
+
+  /**
+   * Write key value to an output file name.
+   * 
+   * Gets the record writer from job's output format.  
+   * Job's output format should be a FileOutputFormat.
+   * 
+   * @param key       the key
+   * @param value     the value
+   * @param baseOutputPath base-output path to write the record to.
+   * Note: Framework will generate unique filename for the baseOutputPath
+   */
+  @SuppressWarnings("unchecked")
+  public void write(KEYOUT key, VALUEOUT value, String baseOutputPath) 
+      throws IOException, InterruptedException {
+    checkBaseOutputPath(baseOutputPath);
+    TaskAttemptContext taskContext = new TaskAttemptContext(
+      context.getConfiguration(), context.getTaskAttemptID());
+    getRecordWriter(taskContext, baseOutputPath).write(key, value);
+  }
+
+  // by being synchronized MultipleOutputTask can be use with a
+  // MultithreadedMapper.
+  @SuppressWarnings("unchecked")
+  private synchronized RecordWriter getRecordWriter(
+      TaskAttemptContext taskContext, String baseFileName) 
+      throws IOException, InterruptedException {
+    
+    // look for record-writer in the cache
+    RecordWriter writer = recordWriters.get(baseFileName);
+    
+    // If not in cache, create a new one
+    if (writer == null) {
+      // get the record writer from context output format
+      FileOutputFormat.setOutputName(taskContext, baseFileName);
+      try {
+        writer = ((OutputFormat) ReflectionUtils.newInstance(
+          taskContext.getOutputFormatClass(), taskContext.getConfiguration()))
+          .getRecordWriter(taskContext);
+      } catch (ClassNotFoundException e) {
+        throw new IOException(e);
+      }
+ 
+      // if counters are enabled, wrap the writer with context 
+      // to increment counters 
+      if (countersEnabled) {
+        writer = new RecordWriterWithCounter(writer, baseFileName, context);
+      }
+      
+      // add the record-writer to the cache
+      recordWriters.put(baseFileName, writer);
+    }
+    return writer;
+  }
+
+   // Create a taskAttemptContext for the named output with 
+   // output format and output key/value types put in the context
+  private TaskAttemptContext getContext(String nameOutput) throws IOException {
+    // The following trick leverages the instantiation of a record writer via
+    // the job thus supporting arbitrary output formats.
+    Job job = new Job(context.getConfiguration());
+    job.setOutputFormatClass(getNamedOutputFormatClass(context, nameOutput));
+    job.setOutputKeyClass(getNamedOutputKeyClass(context, nameOutput));
+    job.setOutputValueClass(getNamedOutputValueClass(context, nameOutput));
+    TaskAttemptContext taskContext = new TaskAttemptContext(
+      job.getConfiguration(), context.getTaskAttemptID());
+    return taskContext;
+  }
+  
+  /**
+   * Closes all the opened outputs.
+   * 
+   * This should be called from cleanup method of map/reduce task.
+   * If overridden subclasses must invoke <code>super.close()</code> at the
+   * end of their <code>close()</code>
+   * 
+   */
+  @SuppressWarnings("unchecked")
+  public void close() throws IOException, InterruptedException {
+    for (RecordWriter writer : recordWriters.values()) {
+      writer.close(context);
+    }
+  }
+}

Copied: hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/lib/output/TestMRMultipleOutputs.java
(from r811565, hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/lib/TestMultipleOutputs.java)
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/lib/output/TestMRMultipleOutputs.java?p2=hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/lib/output/TestMRMultipleOutputs.java&p1=hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/lib/TestMultipleOutputs.java&r1=811565&r2=811582&rev=811582&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/lib/TestMultipleOutputs.java
(original)
+++ hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapreduce/lib/output/TestMRMultipleOutputs.java
Sat Sep  5 01:06:30 2009
@@ -15,25 +15,26 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.hadoop.mapred.lib;
+package org.apache.hadoop.mapreduce.lib.output;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.SequenceFile;
 import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapred.*;
+import org.apache.hadoop.mapred.HadoopTestCase;
+import org.apache.hadoop.mapreduce.*;
 
 import java.io.BufferedReader;
-import java.io.DataOutputStream;
 import java.io.IOException;
 import java.io.InputStreamReader;
-import java.util.Iterator;
 
-public class TestMultipleOutputs extends HadoopTestCase {
+public class TestMRMultipleOutputs extends HadoopTestCase {
 
-  public TestMultipleOutputs() throws IOException {
+  public TestMRMultipleOutputs() throws IOException {
     super(HadoopTestCase.LOCAL_MR, HadoopTestCase.LOCAL_FS, 1, 1);
   }
 
@@ -45,216 +46,180 @@
     _testMultipleOutputs(true);
   }
 
-  private static final Path ROOT_DIR = new Path("testing/mo");
+  private static String localPathRoot = 
+    System.getProperty("test.build.data", "/tmp");
+  private static final Path ROOT_DIR = new Path(localPathRoot, "testing/mo");
   private static final Path IN_DIR = new Path(ROOT_DIR, "input");
   private static final Path OUT_DIR = new Path(ROOT_DIR, "output");
-
-  private Path getDir(Path dir) {
-    // Hack for local FS that does not have the concept of a 'mounting point'
-    if (isLocalFS()) {
-      String localPathRoot = System.getProperty("test.build.data", "/tmp")
-        .replace(' ', '+');
-      dir = new Path(localPathRoot, dir);
-    }
-    return dir;
-  }
+  private static String TEXT = "text";
+  private static String SEQUENCE = "sequence";
 
   public void setUp() throws Exception {
     super.setUp();
-    Path rootDir = getDir(ROOT_DIR);
-    Path inDir = getDir(IN_DIR);
-
-    JobConf conf = createJobConf();
+    Configuration conf = createJobConf();
     FileSystem fs = FileSystem.get(conf);
-    fs.delete(rootDir, true);
-    if (!fs.mkdirs(inDir)) {
-      throw new IOException("Mkdirs failed to create " + inDir.toString());
-    }
+    fs.delete(ROOT_DIR, true);
   }
 
   public void tearDown() throws Exception {
-    Path rootDir = getDir(ROOT_DIR);
-
-    JobConf conf = createJobConf();
+    Configuration conf = createJobConf();
     FileSystem fs = FileSystem.get(conf);
-    fs.delete(rootDir, true);
+    fs.delete(ROOT_DIR, true);
     super.tearDown();
   }
 
   protected void _testMultipleOutputs(boolean withCounters) throws Exception {
-    Path inDir = getDir(IN_DIR);
-    Path outDir = getDir(OUT_DIR);
-
-    JobConf conf = createJobConf();
-    FileSystem fs = FileSystem.get(conf);
-
-    DataOutputStream file = fs.create(new Path(inDir, "part-0"));
-    file.writeBytes("a\nb\n\nc\nd\ne");
-    file.close();
-
-    file = fs.create(new Path(inDir, "part-1"));
-    file.writeBytes("a\nb\n\nc\nd\ne");
-    file.close();
-
-    conf.setJobName("mo");
-    conf.setInputFormat(TextInputFormat.class);
+    String input = "a\nb\nc\nd\ne\nc\nd\ne";
 
-    conf.setOutputKeyClass(LongWritable.class);
-    conf.setOutputValueClass(Text.class);
+    Configuration conf = createJobConf();
+    Job job = MapReduceTestUtil.createJob(conf, IN_DIR, OUT_DIR, 2, 1, input);
 
-    conf.setMapOutputKeyClass(LongWritable.class);
-    conf.setMapOutputValueClass(Text.class);
-
-    conf.setOutputFormat(TextOutputFormat.class);
-    conf.setOutputKeyClass(LongWritable.class);
-    conf.setOutputValueClass(Text.class);
-
-    MultipleOutputs.addNamedOutput(conf, "text", TextOutputFormat.class,
+    job.setJobName("mo");
+    MultipleOutputs.addNamedOutput(job, TEXT, TextOutputFormat.class,
       LongWritable.class, Text.class);
-    MultipleOutputs.addMultiNamedOutput(conf, "sequence",
-      SequenceFileOutputFormat.class, LongWritable.class, Text.class);
-
-    MultipleOutputs.setCountersEnabled(conf, withCounters);
+    MultipleOutputs.addNamedOutput(job, SEQUENCE,
+      SequenceFileOutputFormat.class, IntWritable.class, Text.class);
 
-    conf.setMapperClass(MOMap.class);
-    conf.setReducerClass(MOReduce.class);
+    MultipleOutputs.setCountersEnabled(job, withCounters);
 
-    FileInputFormat.setInputPaths(conf, inDir);
-    FileOutputFormat.setOutputPath(conf, outDir);
+    job.setMapperClass(MOMap.class);
+    job.setReducerClass(MOReduce.class);
 
-    JobClient jc = new JobClient(conf);
-    RunningJob job = jc.submitJob(conf);
-    while (!job.isComplete()) {
-      Thread.sleep(100);
-    }
+    job.waitForCompletion(true);
 
     // assert number of named output part files
     int namedOutputCount = 0;
-    FileStatus[] statuses = fs.listStatus(outDir);
+    int valueBasedOutputCount = 0;
+    FileSystem fs = OUT_DIR.getFileSystem(conf);
+    FileStatus[] statuses = fs.listStatus(OUT_DIR);
     for (FileStatus status : statuses) {
-      if (status.getPath().getName().equals("text-m-00000") ||
-        status.getPath().getName().equals("text-m-00001") ||
-        status.getPath().getName().equals("text-r-00000") ||
-        status.getPath().getName().equals("sequence_A-m-00000") ||
-        status.getPath().getName().equals("sequence_A-m-00001") ||
-        status.getPath().getName().equals("sequence_B-m-00000") ||
-        status.getPath().getName().equals("sequence_B-m-00001") ||
-        status.getPath().getName().equals("sequence_B-r-00000") ||
-        status.getPath().getName().equals("sequence_C-r-00000")) {
+      String fileName = status.getPath().getName();
+      if (fileName.equals("text-m-00000") ||
+          fileName.equals("text-m-00001") ||
+          fileName.equals("text-r-00000") ||
+          fileName.equals("sequence_A-m-00000") ||
+          fileName.equals("sequence_A-m-00001") ||
+          fileName.equals("sequence_B-m-00000") ||
+          fileName.equals("sequence_B-m-00001") ||
+          fileName.equals("sequence_B-r-00000") ||
+          fileName.equals("sequence_C-r-00000")) {
         namedOutputCount++;
+      } else if (fileName.equals("a-r-00000") ||
+          fileName.equals("b-r-00000") ||
+          fileName.equals("c-r-00000") ||
+          fileName.equals("d-r-00000") ||
+          fileName.equals("e-r-00000")) {
+        valueBasedOutputCount++;
       }
     }
     assertEquals(9, namedOutputCount);
+    assertEquals(5, valueBasedOutputCount);
 
     // assert TextOutputFormat files correctness
     BufferedReader reader = new BufferedReader(
       new InputStreamReader(fs.open(
-        new Path(FileOutputFormat.getOutputPath(conf), "text-r-00000"))));
+        new Path(FileOutputFormat.getOutputPath(job), "text-r-00000"))));
     int count = 0;
     String line = reader.readLine();
     while (line != null) {
-      assertTrue(line.endsWith("text"));
+      assertTrue(line.endsWith(TEXT));
       line = reader.readLine();
       count++;
     }
     reader.close();
     assertFalse(count == 0);
-
+    
     // assert SequenceOutputFormat files correctness
     SequenceFile.Reader seqReader =
-      new SequenceFile.Reader(fs, new Path(FileOutputFormat.getOutputPath(conf),
+      new SequenceFile.Reader(fs, new Path(FileOutputFormat.getOutputPath(job),
         "sequence_B-r-00000"), conf);
 
-    assertEquals(LongWritable.class, seqReader.getKeyClass());
+    assertEquals(IntWritable.class, seqReader.getKeyClass());
     assertEquals(Text.class, seqReader.getValueClass());
 
     count = 0;
-    LongWritable key = new LongWritable();
+    IntWritable key = new IntWritable();
     Text value = new Text();
     while (seqReader.next(key, value)) {
-      assertEquals("sequence", value.toString());
+      assertEquals(SEQUENCE, value.toString());
       count++;
     }
     seqReader.close();
     assertFalse(count == 0);
 
-    Counters.Group counters =
-      job.getCounters().getGroup(MultipleOutputs.class.getName());
-    if (!withCounters) {
-      assertEquals(0, counters.size());
-    }
-    else {
-      assertEquals(4, counters.size());
-      assertEquals(4, counters.getCounter("text"));
-      assertEquals(2, counters.getCounter("sequence_A"));
-      assertEquals(4, counters.getCounter("sequence_B"));
-      assertEquals(2, counters.getCounter("sequence_C"));
-
+    if (withCounters) {
+      CounterGroup counters =
+        job.getCounters().getGroup(MultipleOutputs.class.getName());
+      assertEquals(9, counters.size());
+      assertEquals(4, counters.findCounter(TEXT).getValue());
+      assertEquals(2, counters.findCounter(SEQUENCE + "_A").getValue());
+      assertEquals(4, counters.findCounter(SEQUENCE + "_B").getValue());
+      assertEquals(2, counters.findCounter(SEQUENCE + "_C").getValue());
+      assertEquals(2, counters.findCounter("a").getValue());
+      assertEquals(2, counters.findCounter("b").getValue());
+      assertEquals(4, counters.findCounter("c").getValue());
+      assertEquals(4, counters.findCounter("d").getValue());
+      assertEquals(4, counters.findCounter("e").getValue());
     }
-
   }
 
   @SuppressWarnings({"unchecked"})
-  public static class MOMap implements Mapper<LongWritable, Text, LongWritable,
+  public static class MOMap extends Mapper<LongWritable, Text, LongWritable,
     Text> {
 
     private MultipleOutputs mos;
 
-    public void configure(JobConf conf) {
-      mos = new MultipleOutputs(conf);
+    public void setup(Context context) {
+      mos = new MultipleOutputs(context);
     }
 
-    public void map(LongWritable key, Text value,
-                    OutputCollector<LongWritable, Text> output,
-                    Reporter reporter)
-      throws IOException {
-      if (!value.toString().equals("a")) {
-        output.collect(key, value);
-      } else {
-        mos.getCollector("text", reporter).collect(key, new Text("text"));
-        mos.getCollector("sequence", "A", reporter).collect(key,
-          new Text("sequence"));
-        mos.getCollector("sequence", "B", reporter).collect(key,
-          new Text("sequence"));
+    public void map(LongWritable key, Text value, Context context)
+        throws IOException, InterruptedException {
+      context.write(key, value);
+      if (value.toString().equals("a")) {
+        mos.write(TEXT, key, new Text(TEXT));
+        mos.write(SEQUENCE, new IntWritable(1), new Text(SEQUENCE),
+          (SEQUENCE + "_A"));
+        mos.write(SEQUENCE, new IntWritable(2), new Text(SEQUENCE),
+          (SEQUENCE + "_B"));
       }
     }
 
-    public void close() throws IOException {
+    public void cleanup(Context context) 
+        throws IOException, InterruptedException {
       mos.close();
     }
   }
 
   @SuppressWarnings({"unchecked"})
-  public static class MOReduce implements Reducer<LongWritable, Text,
+  public static class MOReduce extends Reducer<LongWritable, Text,
     LongWritable, Text> {
 
     private MultipleOutputs mos;
-
-    public void configure(JobConf conf) {
-      mos = new MultipleOutputs(conf);
-    }
-
-    public void reduce(LongWritable key, Iterator<Text> values,
-                       OutputCollector<LongWritable, Text> output,
-                       Reporter reporter)
-      throws IOException {
-      while (values.hasNext()) {
-        Text value = values.next();
+    
+    public void setup(Context context) {
+      mos = new MultipleOutputs(context);
+   }
+
+    public void reduce(LongWritable key, Iterable<Text> values, 
+        Context context) throws IOException, InterruptedException {
+      for (Text value : values) {
+        mos.write(key, value, value.toString());
         if (!value.toString().equals("b")) {
-          output.collect(key, value);
+          context.write(key, value);
         } else {
-          mos.getCollector("text", reporter).collect(key, new Text("text"));
-          mos.getCollector("sequence", "B", reporter).collect(key,
-            new Text("sequence"));
-          mos.getCollector("sequence", "C", reporter).collect(key,
-            new Text("sequence"));
+          mos.write(TEXT, key, new Text(TEXT));
+          mos.write(SEQUENCE, new IntWritable(2), new Text(SEQUENCE),
+            (SEQUENCE + "_B"));
+          mos.write(SEQUENCE, new IntWritable(3), new Text(SEQUENCE),
+            (SEQUENCE + "_C"));
         }
       }
     }
 
-    public void close() throws IOException {
+    public void cleanup(Context context) 
+        throws IOException, InterruptedException {
       mos.close();
     }
   }
-
 }



Mime
View raw message