HADOOP-657. Free disk space should be modelled and used by the scheduler

to make scheduling decisions. (Ari Rabkin via omalley)


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

CHANGES.txt

@@ -72,6 +72,9 @@ Trunk (unreleased changes)
 
     HADOOP-153. Provides a way to skip bad records. (Sharad Agarwal via ddas)
 
+    HADOOP-657. Free disk space should be modelled and used by the scheduler
+    to make scheduling decisions. (Ari Rabkin via omalley)
+
   IMPROVEMENTS
 
     HADOOP-3732. Delay intialization of datanode block verification till

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

@@ -800,7 +800,8 @@ public class JobClient extends Configured implements MRConstants, Tool  {
     private String splitClass;
     private BytesWritable bytes = new BytesWritable();
     private String[] locations;
-      
+    long dataLength;
+
     public void setBytes(byte[] data, int offset, int length) {
       bytes.set(data, offset, length);
     }
@@ -831,6 +832,7 @@ public class JobClient extends Configured implements MRConstants, Tool  {
       
     public void readFields(DataInput in) throws IOException {
       splitClass = Text.readString(in);
+      dataLength = in.readLong();
       bytes.readFields(in);
       int len = WritableUtils.readVInt(in);
       locations = new String[len];
@@ -841,12 +843,21 @@ public class JobClient extends Configured implements MRConstants, Tool  {
       
     public void write(DataOutput out) throws IOException {
       Text.writeString(out, splitClass);
+      out.writeLong(dataLength);
       bytes.write(out);
       WritableUtils.writeVInt(out, locations.length);
       for(int i = 0; i < locations.length; i++) {
         Text.writeString(out, locations[i]);
       }        
     }
+
+    public long getDataLength() {
+      return dataLength;
+    }
+    public void setDataLength(long l) {
+      dataLength = l;
+    }
+    
   }
     
   private static final int CURRENT_SPLIT_FILE_VERSION = 0;
@@ -871,6 +882,7 @@ public class JobClient extends Configured implements MRConstants, Tool  {
       rawSplit.setClassName(split.getClass().getName());
       buffer.reset();
       split.write(buffer);
+      rawSplit.setDataLength(split.getLength());
       rawSplit.setBytes(buffer.getData(), 0, buffer.getLength());
       rawSplit.setLocations(split.getLocations());
       rawSplit.write(out);

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

@@ -119,6 +119,9 @@ class JobInProgress {
   private Map<String, Integer> trackerToFailuresMap = 
     new TreeMap<String, Integer>();
     
+  //Confine estimation algorithms to an "oracle" class that JIP queries.
+  private ResourceEstimator resourceEstimator; 
+  
   long startTime;
   long finishTime;
 
@@ -129,6 +132,7 @@ class JobInProgress {
   private JobID jobId;
   private boolean hasSpeculativeMaps;
   private boolean hasSpeculativeReduces;
+  private long inputLength = 0;
 
   // Per-job counters
   public static enum Counter { 
@@ -220,6 +224,7 @@ class JobInProgress {
     this.runningMapCache = new IdentityHashMap<Node, Set<TaskInProgress>>();
     this.nonRunningReduces = new LinkedList<TaskInProgress>();    
     this.runningReduces = new LinkedHashSet<TaskInProgress>();
+    this.resourceEstimator = new ResourceEstimator(this);
   }
 
   /**
@@ -335,10 +340,12 @@ class JobInProgress {
     numMapTasks = splits.length;
     maps = new TaskInProgress[numMapTasks];
     for(int i=0; i < numMapTasks; ++i) {
+      inputLength += splits[i].getDataLength();
       maps[i] = new TaskInProgress(jobId, jobFile, 
                                    splits[i], 
                                    jobtracker, conf, this, i);
     }
+    LOG.info("Input size for job "+ jobId + " = " + inputLength);
     if (numMapTasks > 0) { 
       LOG.info("Split info for job:" + jobId);
       nonRunningMapCache = createCache(splits, maxLevel);
@@ -434,6 +441,10 @@ class JobInProgress {
       this.priority = priority;
     }
   }
+  
+  long getInputLength() {
+    return inputLength;
+  }
  
   /**
    * Get the list of map tasks
@@ -1076,6 +1087,17 @@ class JobInProgress {
     Node node = jobtracker.getNode(tts.getHost());
     Node nodeParentAtMaxLevel = null;
     
+
+    long outSize = resourceEstimator.getEstimatedMapOutputSize();
+    if(tts.getAvailableSpace() < outSize) {
+      LOG.warn("No room for map task. Node " + node + 
+               " has " + tts.getAvailableSpace() + 
+               " bytes free; but we expect map to take " + outSize);
+
+      return -1; //see if a different TIP might work better. 
+    }
+    
+    
     // For scheduling a map task, we have two caches and a list (optional)
     //  I)   one for non-running task
     //  II)  one for running task (this is for handling speculation)
@@ -1272,6 +1294,15 @@ class JobInProgress {
       return -1;
     }
 
+    long outSize = resourceEstimator.getEstimatedReduceInputSize();
+    if(tts.getAvailableSpace() < outSize) {
+      LOG.warn("No room for reduce task. Node " + taskTracker + " has " +
+               tts.getAvailableSpace() + 
+               " bytes free; but we expect reduce input to take " + outSize);
+
+      return -1; //see if a different TIP might work better. 
+    }
+    
     // 1. check for a never-executed reduce tip
     // reducers don't have a cache and so pass -1 to explicitly call that out
     tip = findTaskFromList(nonRunningReduces, tts, numUniqueHosts, false);
@@ -1342,6 +1373,7 @@ class JobInProgress {
 
     // Mark the TIP as complete
     tip.completed(taskid);
+    resourceEstimator.updateWithCompletedTask(status, tip);
 
     // Update jobhistory 
     String taskTrackerName = jobtracker.getNode(jobtracker.getTaskTracker(

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

@@ -0,0 +1,114 @@
+/**
+ * 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 java.util.concurrent.atomic.*;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * Class responsible for modeling the resource consumption of running tasks.
+ * 
+ * For now, we just do temp space for maps
+ * 
+ * There is one ResourceEstimator per JobInProgress
+ *
+ */
+public class ResourceEstimator {
+
+  //Log with JobInProgress
+  private static final Log LOG = LogFactory.getLog(
+      "org.apache.hadoop.mapred.ResourceEstimator");
+
+
+  /**
+   * Estimated ratio of output to input size for map tasks. 
+   */
+  private double mapBlowupRatio;
+  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; 
+
+
+  public ResourceEstimator(JobInProgress job) {
+    mapBlowupRatio = INITIAL_BLOWUP_GUESS;
+    this.job = job;
+    estimateWeight = INITIAL_EST_WEIGHT_PERCENT * job.desiredMaps();
+  }
+
+
+  /**
+   * Have private access methods to abstract away synchro.
+   * @return
+   */
+  private synchronized double getBlowupRatio() {
+    return mapBlowupRatio;
+  }
+
+  private synchronized void setBlowupRatio(double b)  {
+    mapBlowupRatio = b;
+  }
+
+
+
+  public 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();
+      
+      LOG.info("measured blowup on " + tip.getTIPId() + " was " +
+          ts.getOutputSize() + "/" +tip.getMapInputSize() + " = " 
+          + blowupOnThisTask);
+      
+      double newEstimate = blowupOnThisTask / estimateWeight + 
+          ((estimateWeight - 1) / estimateWeight) * getBlowupRatio();
+      estimateWeight++; 
+      setBlowupRatio(newEstimate);
+    }
+  }
+
+  /**
+   * 
+   * @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 + ")");
+    return estimate;
+  }
+
+
+  //estimate that each reduce gets an equal share of total map output
+  public long getEstimatedReduceInputSize() {
+    return 
+       getEstimatedMapOutputSize() * job.desiredMaps() / job.desiredReduces();
+  }
+  
+
+}

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

@@ -820,6 +820,14 @@ class TaskInProgress {
     return ret.toString();
   }
 
+  public long getMapInputSize() {
+    if(isMapTask()) {
+      return rawSplit.getDataLength();
+    } else {
+      return 0;
+    }
+  }
+  
   public void clearSplit() {
     rawSplit.clearBytes();
   }

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

@@ -52,6 +52,7 @@ abstract class TaskStatus implements Writable, Cloneable {
     
   private long startTime; 
   private long finishTime; 
+  private long outputSize;
     
   private Phase phase = Phase.STARTING; 
   private Counters counters;
@@ -224,6 +225,21 @@ abstract class TaskStatus implements Writable, Cloneable {
     this.counters = counters;
   }
   
+  /**
+   * Returns the number of bytes of output from this map.
+   */
+  public long getOutputSize() {
+    return outputSize;
+  }
+  
+  /**
+   * Set the size on disk of this task's output.
+   * @param l the number of map output bytes
+   */
+  void setOutputSize(long l)  {
+    outputSize = l;
+  }
+  
   /**
    * Get the list of maps from which output-fetches failed.
    * 
@@ -278,6 +294,7 @@ abstract class TaskStatus implements Writable, Cloneable {
     
     this.phase = status.getPhase();
     this.counters = status.getCounters();
+    this.outputSize = status.outputSize;
   }
   
   /**
@@ -313,6 +330,7 @@ abstract class TaskStatus implements Writable, Cloneable {
     out.writeLong(startTime);
     out.writeLong(finishTime);
     out.writeBoolean(includeCounters);
+    out.writeLong(outputSize);
     if (includeCounters) {
       counters.write(out);
     }
@@ -330,6 +348,7 @@ abstract class TaskStatus implements Writable, Cloneable {
     this.finishTime = in.readLong(); 
     counters = new Counters();
     this.includeCounters = in.readBoolean();
+    this.outputSize = in.readLong();
     if (includeCounters) {
       counters.readFields(in);
     }

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

@@ -1025,6 +1025,7 @@ public class TaskTracker
     if (askForNewTask) {
       checkLocalDirs(fConf.getLocalDirs());
       askForNewTask = enoughFreeSpace(localMinSpaceStart);
+      status.setAvailableSpace( getFreeSpace() );
     }
       
     //
@@ -1243,8 +1244,8 @@ public class TaskTracker
   }
     
   /**
-   * Check if all of the local directories have enough
-   * free space
+   * Check if any of the local directories has enough
+   * free space  (more than minSpace)
    * 
    * If not, do not try to get a new task assigned 
    * @return
@@ -1254,6 +1255,11 @@ public class TaskTracker
     if (minSpace == 0) {
       return true;
     }
+    return minSpace < getFreeSpace();
+  }
+  
+  private long getFreeSpace() throws IOException {
+    long biggestSeenSoFar = 0;
     String[] localDirs = fConf.getLocalDirs();
     for (int i = 0; i < localDirs.length; i++) {
       DF df = null;
@@ -1264,13 +1270,51 @@ public class TaskTracker
         localDirsDf.put(localDirs[i], df);
       }
 
-      if (df.getAvailable() > minSpace)
-        return true;
+      long availOnThisVol = df.getAvailable();
+      if (availOnThisVol > biggestSeenSoFar) {
+        biggestSeenSoFar = availOnThisVol;
+      }
     }
-
-    return false;
+    
+    //Should ultimately hold back the space we expect running tasks to use but 
+    //that estimate isn't currently being passed down to the TaskTrackers    
+    return biggestSeenSoFar;
   }
     
+  /**
+   * Try to get the size of output for this task.
+   * Returns -1 if it can't be found.
+   * @return
+   */
+  long tryToGetOutputSize(TaskAttemptID taskId, JobConf conf) {
+    
+    try{
+      TaskInProgress tip;
+      synchronized(this) {
+        tip = tasks.get(taskId);
+      }
+      if(tip == null)
+         return -1;
+      
+      MapOutputFile mapOutputFile = new MapOutputFile();
+      mapOutputFile.setJobId(taskId.getJobID());
+      mapOutputFile.setConf(conf);
+      
+      Path tmp_output =  mapOutputFile.getOutputFile(taskId);
+      if(tmp_output == null)
+        return 0;
+      FileSystem localFS = FileSystem.getLocal(conf);
+      FileStatus stat = localFS.getFileStatus(tmp_output);
+      if(stat == null)
+        return 0;
+      else
+        return stat.getLen();
+    } catch(IOException e) {
+      LOG.info(e);
+      return -1;
+    }
+  }
+  
   /**
    * Start a new task.
    * All exceptions are handled locally, so that we don't mess up the
@@ -1596,6 +1640,8 @@ public class TaskTracker
       this.done = true;
       
       LOG.info("Task " + task.getTaskID() + " is done.");
+      LOG.info("reported output size for " + task.getTaskID() +  "  was " + taskStatus.getOutputSize());
+
     }
 
     /**
@@ -1660,7 +1706,7 @@ public class TaskTracker
                                      localJobConf). toString());
               } catch (IOException e) {
                 LOG.warn("Working Directory of the task " + task.getTaskID() +
-                		 "doesnt exist. Throws expetion " +
+                		 "doesnt exist. Caught exception " +
                           StringUtils.stringifyException(e));
               }
               // Build the command  
@@ -2195,6 +2241,7 @@ public class TaskTracker
     for(TaskInProgress tip: runningTasks.values()) {
       TaskStatus status = tip.getStatus();
       status.setIncludeCounters(sendCounters);
+      status.setOutputSize(tryToGetOutputSize(status.getTaskID(), fConf));
       // send counters for finished or failed tasks.
       if (status.getRunState() != TaskStatus.State.RUNNING) {
         status.setIncludeCounters(true);

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

@@ -48,11 +48,13 @@ class TaskTrackerStatus implements Writable {
   volatile long lastSeen;
   private int maxMapTasks;
   private int maxReduceTasks;
+  long availableSpace; //space available on this node
     
   /**
    */
   public TaskTrackerStatus() {
     taskReports = new ArrayList<TaskStatus>();
+    this.availableSpace = Long.MAX_VALUE; //not measured by default.
   }
 
   /**
@@ -69,6 +71,7 @@ class TaskTrackerStatus implements Writable {
     this.failures = failures;
     this.maxMapTasks = maxMapTasks;
     this.maxReduceTasks = maxReduceTasks;
+    this.availableSpace = Long.MAX_VALUE; //not measured by default.
   }
 
   /**
@@ -166,6 +169,20 @@ class TaskTrackerStatus implements Writable {
   public int getMaxReduceTasks() {
     return maxReduceTasks;
   }  
+  
+  /**
+   * Will return LONG_MAX if space hasn't been measured yet.
+   * @return bytes of available local disk space on this tasktracker.
+   */
+  public long getAvailableSpace() {
+    return availableSpace;
+  }
+  
+  public void setAvailableSpace(long a) {
+    availableSpace = a;
+  }
+  
+  
   ///////////////////////////////////////////
   // Writable
   ///////////////////////////////////////////
@@ -177,6 +194,7 @@ class TaskTrackerStatus implements Writable {
     out.writeInt(maxMapTasks);
     out.writeInt(maxReduceTasks);
     out.writeInt(taskReports.size());
+    out.writeLong(availableSpace);
     for (TaskStatus taskStatus : taskReports) {
       TaskStatus.writeTaskStatus(out, taskStatus);
     }
@@ -191,6 +209,7 @@ class TaskTrackerStatus implements Writable {
     this.maxReduceTasks = in.readInt();
     taskReports.clear();
     int numTasks = in.readInt();
+    this.availableSpace = in.readLong();
     for (int i = 0; i < numTasks; i++) {
       taskReports.add(TaskStatus.readTaskStatus(in));
     }