hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cdoug...@apache.org
Subject svn commit: r669343 - in /hadoop/core/trunk: CHANGES.txt src/mapred/org/apache/hadoop/mapred/JobConf.java src/mapred/org/apache/hadoop/mapred/MapTask.java src/mapred/org/apache/hadoop/mapred/ReduceTask.java
Date Thu, 19 Jun 2008 00:43:06 GMT
Author: cdouglas
Date: Wed Jun 18 17:43:05 2008
New Revision: 669343

URL: http://svn.apache.org/viewvc?rev=669343&view=rev
Log:
HADOOP-3586. Provide deprecated, backwards compatibile semantics for the
combiner to be run once and only once on each record.

Modified:
    hadoop/core/trunk/CHANGES.txt
    hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/JobConf.java
    hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/MapTask.java
    hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/ReduceTask.java

Modified: hadoop/core/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/CHANGES.txt?rev=669343&r1=669342&r2=669343&view=diff
==============================================================================
--- hadoop/core/trunk/CHANGES.txt (original)
+++ hadoop/core/trunk/CHANGES.txt Wed Jun 18 17:43:05 2008
@@ -606,6 +606,9 @@
     HADOOP-3520.  TestDFSUpgradeFromImage triggers a race condition in the
     Upgrade Manager. Fixed. (dhruba)
 
+    HADOOP-3586. Provide deprecated, backwards compatibile semantics for the
+    combiner to be run once and only once on each record. (cdouglas)
+
 Release 0.17.1 - Unreleased
 
   INCOMPATIBLE CHANGES

Modified: hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/JobConf.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/JobConf.java?rev=669343&r1=669342&r2=669343&view=diff
==============================================================================
--- hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/JobConf.java (original)
+++ hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/JobConf.java Wed Jun 18 17:43:05
2008
@@ -799,6 +799,20 @@
   }
   
   /**
+   * If true, ensures the combiner is run once and only once on output from
+   * the map. Otherwise, combiner may be run zero or more times.
+   */
+  @Deprecated
+  public void setCombineOnceOnly(JobConf conf, boolean value) {
+    conf.setBoolean("mapred.combine.once", value);
+  }
+
+  @Deprecated
+  public boolean getCombineOnceOnly() {
+    return getBoolean("mapred.combine.once", false);
+  }
+
+  /**
    * Should speculative execution be used for this job? 
    * Defaults to <code>true</code>.
    * 

Modified: hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/MapTask.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/MapTask.java?rev=669343&r1=669342&r2=669343&view=diff
==============================================================================
--- hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/MapTask.java (original)
+++ hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/MapTask.java Wed Jun 18 17:43:05
2008
@@ -31,7 +31,9 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
+import java.util.NoSuchElementException;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -826,11 +828,30 @@
             writer = new IFile.Writer(job, out, keyClass, valClass, codec);
 
             if (i == partition) {
-              final long recordStart = out.getPos();
-              writer.append(key, value);
-              // Note that our map byte count will not be accurate with
-              // compression
-              mapOutputByteCounter.increment(out.getPos() - recordStart);
+              if (job.getCombineOnceOnly()) {
+                Reducer combiner =
+                  (Reducer)ReflectionUtils.newInstance(combinerClass, job);
+                combineCollector.setWriter(writer);
+                combiner.reduce(key, new Iterator<V>() {
+                    private boolean done = false;
+                    public boolean hasNext() { return !done; }
+                    public V next() {
+                      if (done)
+                        throw new NoSuchElementException();
+                      done = true;
+                      return value;
+                    }
+                    public void remove() {
+                      throw new UnsupportedOperationException();
+                    }
+                  }, combineCollector, reporter);
+              } else {
+                final long recordStart = out.getPos();
+                writer.append(key, value);
+                // Note that our map byte count will not be accurate with
+                // compression
+                mapOutputByteCounter.increment(out.getPos() - recordStart);
+              }
             }
             writer.close();
 
@@ -1030,7 +1051,8 @@
           segmentStart = finalOut.getPos();
           Writer<K, V> writer = 
               new Writer<K, V>(job, finalOut, keyClass, valClass, codec);
-          if (null == combinerClass || numSpills < minSpillsForCombine) {
+          if (null == combinerClass || job.getCombineOnceOnly() ||
+              numSpills < minSpillsForCombine) {
             Merger.writeFile(kvIter, writer, reporter);
           } else {
             combineCollector.setWriter(writer);

Modified: hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/ReduceTask.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/ReduceTask.java?rev=669343&r1=669342&r2=669343&view=diff
==============================================================================
--- hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/ReduceTask.java (original)
+++ hadoop/core/trunk/src/mapred/org/apache/hadoop/mapred/ReduceTask.java Wed Jun 18 17:43:05
2008
@@ -1286,7 +1286,9 @@
       this.numCopiers = conf.getInt("mapred.reduce.parallel.copies", 5);
       this.maxInFlight = 4 * numCopiers;
       this.maxBackoff = conf.getInt("mapred.reduce.copy.backoff", 300);
-      this.combinerClass = conf.getCombinerClass();
+      this.combinerClass = conf.getCombineOnceOnly()
+        ? null
+        : conf.getCombinerClass();
       combineCollector = (null != combinerClass)
         ? new CombineOutputCollector(reduceCombineOutputCounter)
         : null;



Mime
View raw message