HADOOP-3961. Fix task disk space requirement estimates for virtual

input jobs. Delays limiting task placement until after 10% of the maps
have finished. (Ari Rabkin via omalley)


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

CHANGES.txt

@@ -388,6 +388,10 @@ Trunk (unreleased changes)
     HADOOP-4030. Remove lzop from the default list of codecs. (Arun Murthy via
     cdouglas)
 
+    HADOOP-3961. Fix task disk space requirement estimates for virtual
+    input jobs. Delays limiting task placement until after 10% of the maps
+    have finished. (Ari Rabkin via omalley)
+
 Release 0.18.1 - Unreleased
 
   BUG FIXES

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

@@ -37,23 +37,23 @@ public class ResourceEstimator {
 
 
   /**
-   * Estimated ratio of output to input size for map tasks. 
+   * Estimated ratio of output to (input size+1) for map tasks. 
    */
   private double mapBlowupRatio;
+  
+  /**
+   * How much relative weight to put on the current estimate.
+   * Each completed map has unit weight.
+   */
   private double estimateWeight;
-  private JobInProgress job;
-
-  //guess a factor of two blowup due to temp space for merge
-  public static final double INITIAL_BLOWUP_GUESS = 1; 
-
-  //initial estimate is weighted as much as this fraction of the real datapoints
-  static final double INITIAL_EST_WEIGHT_PERCENT = 0.05; 
-
+  final private JobInProgress job;
+  final private int threshholdToUse;
 
   public ResourceEstimator(JobInProgress job) {
-    mapBlowupRatio = INITIAL_BLOWUP_GUESS;
     this.job = job;
-    estimateWeight = INITIAL_EST_WEIGHT_PERCENT * job.desiredMaps();
+    threshholdToUse = job.desiredMaps()/ 10;
+    mapBlowupRatio = 0;
+    estimateWeight = 1;
   }
 
 
@@ -69,45 +69,71 @@ public class ResourceEstimator {
     mapBlowupRatio = b;
   }
 
-
-
-  public void updateWithCompletedTask(TaskStatus ts, TaskInProgress tip) {
-
+  void updateWithCompletedTask(TaskStatus ts, TaskInProgress tip) {
+    
     //-1 indicates error, which we don't average in.
     if(tip.isMapTask() &&  ts.getOutputSize() != -1)  {
       double blowupOnThisTask = ts.getOutputSize() / 
-        (double) tip.getMapInputSize();
+        ((double) tip.getMapInputSize() + 1);
       
       LOG.info("measured blowup on " + tip.getTIPId() + " was " +
-          ts.getOutputSize() + "/" +tip.getMapInputSize() + " = " 
+          ts.getOutputSize() + "/" +(tip.getMapInputSize()+1) + " = " 
           + blowupOnThisTask);
       
-      double newEstimate = blowupOnThisTask / estimateWeight + 
-          ((estimateWeight - 1) / estimateWeight) * getBlowupRatio();
-      estimateWeight++; 
+      double newEstimate;
+      synchronized(this) {
+        newEstimate = blowupOnThisTask / estimateWeight + 
+            ((estimateWeight - 1) / estimateWeight) * getBlowupRatio();
+        estimateWeight++; 
+      }
       setBlowupRatio(newEstimate);
+      
+      LOG.info("new estimate is blowup = " + newEstimate);
     }
   }
 
   /**
-   * 
+   * @return estimated length of this job's total map output
+   */
+  protected long getEstimatedTotalMapOutputSize()  {
+    double estWeight;
+    synchronized(this) {
+      estWeight = this.estimateWeight;
+    }
+    
+    if(estWeight < threshholdToUse) {
+      return 0;
+    } else {
+      double blowup =getBlowupRatio();
+      long inputSize = job.getInputLength() + job.desiredMaps(); 
+      //add desiredMaps() so that randomwriter case doesn't blow up
+      long estimate = Math.round(inputSize * blowup * 2.0);
+  
+      LOG.debug("estimate total map output will be " + estimate +
+          " bytes. (blowup = 2*" + blowup + ")");
+      return estimate;
+    }
+  }
+  
+  /**
    * @return estimated length of this job's average map output
-   * @throws IOException if the split's getLength() does.
    */
-  public long getEstimatedMapOutputSize()  {
-    double blowup =getBlowupRatio();
-    long estimate =  
-      (long) (job.getInputLength() * blowup / job.desiredMaps() * 2.0);
-    LOG.info("estimate map will take " + estimate +
-        " bytes. (blowup = 2*" + blowup + ")");
+  long getEstimatedMapOutputSize() {
+    long estimate = getEstimatedTotalMapOutputSize()  / job.desiredMaps();
     return estimate;
   }
 
-
-  //estimate that each reduce gets an equal share of total map output
-  public long getEstimatedReduceInputSize() {
-    return 
-       getEstimatedMapOutputSize() * job.desiredMaps() / job.desiredReduces();
+  /**
+   * 
+   * @return estimated length of this job's average reduce input
+   */
+  long getEstimatedReduceInputSize() {
+    if(job.desiredReduces() == 0) {//no reduce output, so no size
+      return 0;
+    } else {
+      return getEstimatedTotalMapOutputSize() / job.desiredReduces();
+      //estimate that each reduce gets an equal share of total map output
+    }
   }
   
 

src/test/org/apache/hadoop/mapred/TestResourceEstimation.java

@@ -0,0 +1,58 @@
+/*
+ * 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.mapred;
+
+import junit.framework.TestCase;
+import org.apache.hadoop.mapred.JobClient.RawSplit;
+
+public class TestResourceEstimation extends TestCase {
+  
+
+  public void testResourceEstimator() throws Exception {
+    final int maps = 100;
+    final int reduces = 2;
+    final int singleMapOutputSize = 1000;
+    JobConf jc = new JobConf();
+    JobID jid = new JobID("testJT", 0);
+    jc.setNumMapTasks(maps);
+    jc.setNumReduceTasks(reduces);
+    
+    JobInProgress jip = new JobInProgress(jid, jc);
+    //unfortunately, we can't set job input size from here.
+    ResourceEstimator re = new ResourceEstimator(jip);
+    
+    for(int i = 0; i < maps / 10 -1; ++i) {
+
+      long estOutSize = re.getEstimatedMapOutputSize();
+      System.out.println(estOutSize);
+      assertEquals(0, estOutSize);
+      
+      TaskStatus ts = new MapTaskStatus();
+      ts.setOutputSize(singleMapOutputSize);
+      RawSplit split = new RawSplit();
+      split.setDataLength(0);
+      TaskInProgress tip = new TaskInProgress(jid, "", split, null, jc, jip, 0);
+      re.updateWithCompletedTask(ts, tip);
+    }
+    
+    assertEquals(2* singleMapOutputSize, re.getEstimatedMapOutputSize());
+    assertEquals(2* singleMapOutputSize * maps / reduces, re.getEstimatedReduceInputSize());
+    
+  }
+
+}