HADOOP-1274. Support different number of mappers and reducers per TaskTracker to allow administrators to better configure and utilize heterogenous clusters.

Configuration changes to hadoop-default.xml:
  add mapred.tasktracker.map.tasks.maximum (default value of 2)
  add mapred.tasktracker.reduce.tasks.maximum (default value of 2)
  remove mapred.tasktracker.tasks.maximum (deprecated for 0.16.0)


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

CHANGES.txt

@@ -51,6 +51,14 @@ Trunk (unreleased changes)
     HADOOP-1989. Support simulated DataNodes. This helps creating large virtual
     clusters for testing purposes.  (Sanjay Radia via dhruba)
     
+    HADOOP-1274. Support different number of mappers and reducers per
+    TaskTracker to  allow administrators to better configure and utilize
+    heterogenous clusters. 
+    Configuration changes to hadoop-default.xml:
+      add mapred.tasktracker.map.tasks.maximum (default value of 2)
+      add mapred.tasktracker.reduce.tasks.maximum (default value of 2)
+      remove mapred.tasktracker.tasks.maximum (deprecated for 0.16.0)
+
   OPTIMIZATIONS
 
     HADOOP-1898.  Release the lock protecting the last time of the last stack

conf/hadoop-default.xml

@@ -603,9 +603,17 @@ creations/deletions), or "all".</description>
 </property>
 
 <property>
-  <name>mapred.tasktracker.tasks.maximum</name>
+  <name>mapred.tasktracker.map.tasks.maximum</name>
   <value>2</value>
-  <description>The maximum number of tasks that will be run
+  <description>The maximum number of map tasks that will be run
+  simultaneously by a task tracker.
+  </description>
+</property>
+
+<property>
+  <name>mapred.tasktracker.reduce.tasks.maximum</name>
+  <value>2</value>
+  <description>The maximum number of reduce tasks that will be run
   simultaneously by a task tracker.
   </description>
 </property>

src/examples/org/apache/hadoop/examples/Sort.java

@@ -76,8 +76,12 @@ public class Sort extends Configured implements Tool {
     ClusterStatus cluster = client.getClusterStatus();
     int num_maps = cluster.getTaskTrackers() * 
     jobConf.getInt("test.sort.maps_per_host", 10);
-    int num_reduces = cluster.getTaskTrackers() * 
-    jobConf.getInt("test.sort.reduces_per_host", cluster.getMaxTasks());
+    int num_reduces = (int) (cluster.getMaxReduceTasks() * 0.9);
+    String sort_reduces = jobConf.get("test.sort.reduces_per_host");
+    if (sort_reduces != null) {
+       num_reduces = cluster.getTaskTrackers() * 
+                       Integer.parseInt(sort_reduces);
+    }
     Class<? extends InputFormat> inputFormatClass = 
       SequenceFileInputFormat.class;
     Class<? extends OutputFormat> outputFormatClass = 

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

@@ -54,7 +54,8 @@ public class ClusterStatus implements Writable {
   private int task_trackers;
   private int map_tasks;
   private int reduce_tasks;
-  private int max_tasks;
+  private int max_map_tasks;
+  private int max_reduce_tasks;
   private JobTracker.State state;
 
   ClusterStatus() {}
@@ -68,12 +69,13 @@ public class ClusterStatus implements Writable {
    * @param max the maximum no. of tasks in the cluster
    * @param state the {@link JobTracker.State} of the <code>JobTracker</code>
    */
-  ClusterStatus(int trackers, int maps, int reduces, int max,
-                JobTracker.State state) {
+  ClusterStatus(int trackers, int maps, int reduces, int maxMaps,
+                int maxReduces, JobTracker.State state) {
     task_trackers = trackers;
     map_tasks = maps;
     reduce_tasks = reduces;
-    max_tasks = max;
+    max_map_tasks = maxMaps;
+    max_reduce_tasks = maxReduces;
     this.state = state;
   }
   
@@ -109,11 +111,31 @@ public class ClusterStatus implements Writable {
    * Get the maximum capacity for running tasks in the cluster.
    * 
    * @return the maximum capacity for running tasks in the cluster.
+   * @deprecated Use {@link #getMaxMapTasks()} and/or
+   *  {@link #getMaxReduceTasks()}
    */
   public int getMaxTasks() {
-    return max_tasks;
+    return (max_map_tasks + max_reduce_tasks);
+  }
+  
+  /**
+   * Get the maximum capacity for running map tasks in the cluster.
+   * 
+   * @return the maximum capacity for running map tasks in the cluster.
+   */
+  public int getMaxMapTasks() {
+    return max_map_tasks;
   }
 
+  /**
+   * Get the maximum capacity for running reduce tasks in the cluster.
+   * 
+   * @return the maximum capacity for running reduce tasks in the cluster.
+   */
+  public int getMaxReduceTasks() {
+    return max_reduce_tasks;
+  }
+  
   /**
    * Get the current state of the <code>JobTracker</code>, 
    * as {@link JobTracker.State}
@@ -128,7 +150,8 @@ public class ClusterStatus implements Writable {
     out.writeInt(task_trackers);
     out.writeInt(map_tasks);
     out.writeInt(reduce_tasks);
-    out.writeInt(max_tasks);
+    out.writeInt(max_map_tasks);
+    out.writeInt(max_reduce_tasks);
     WritableUtils.writeEnum(out, state);
   }
 
@@ -136,7 +159,8 @@ public class ClusterStatus implements Writable {
     task_trackers = in.readInt();
     map_tasks = in.readInt();
     reduce_tasks = in.readInt();
-    max_tasks = in.readInt();
+    max_map_tasks = in.readInt();
+    max_reduce_tasks = in.readInt();
     state = WritableUtils.readEnum(in, JobTracker.State.class);
   }
 

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

@@ -35,8 +35,10 @@ interface InterTrackerProtocol extends VersionedProtocol {
    * version 5 introduced that removes locateMapOutputs and instead uses
    * getTaskCompletionEvents to figure finished maps and fetch the outputs
    * version 6 adds maxTasks to TaskTrackerStatus for HADOOP-1245
+   * version 7 replaces maxTasks by maxMapTasks and maxReduceTasks in 
+   * TaskTrackerStatus for HADOOP-1274
    */
-  public static final long versionID = 6L;
+  public static final long versionID = 7L;
   
   public final static int TRACKERS_OK = 0;
   public final static int UNKNOWN_TASKTRACKER = 1;

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

@@ -931,8 +931,8 @@ public class JobConf extends Configuration {
    * 
    * <p>The right number of reduces seems to be <code>0.95</code> or 
    * <code>1.75</code> multiplied by (&lt;<i>no. of nodes</i>&gt; * 
-   * <a href="{@docRoot}/../hadoop-default.html#mapred.tasktracker.tasks.maximum">
-   * mapred.tasktracker.tasks.maximum</a>).
+   * <a href="{@docRoot}/../hadoop-default.html#mapred.tasktracker.reduce.tasks.maximum">
+   * mapred.tasktracker.reduce.tasks.maximum</a>).
    * </p>
    * 
    * <p>With <code>0.95</code> all of the reduces can launch immediately and 

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

@@ -33,8 +33,10 @@ public interface JobSubmissionProtocol extends VersionedProtocol {
    *changed.
    *Changed to 4 since killTask(String,boolean) is added
    *Version 4: added jobtracker state to ClusterStatus
+   *Version 5: max_tasks in ClusterStatus is replaced by
+   * max_map_tasks and max_reduce_tasks for HADOOP-1274
    */
-  public static final long versionID = 4L;
+  public static final long versionID = 5L;
 
   /**
    * Allocate a name for the job.

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

@@ -506,8 +506,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
   private String trackerIdentifier;
   long startTime;
   int totalSubmissions = 0;
-
-  private int totalTaskCapacity;
+  private int totalMapTaskCapacity;
+  private int totalReduceTaskCapacity;
   private HostsFileReader hostsReader;
 
   //
@@ -1230,7 +1230,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
     if (oldStatus != null) {
       totalMaps -= oldStatus.countMapTasks();
       totalReduces -= oldStatus.countReduceTasks();
-      totalTaskCapacity -= oldStatus.getMaxTasks();
+      totalMapTaskCapacity -= oldStatus.getMaxMapTasks();
+      totalReduceTaskCapacity -= oldStatus.getMaxReduceTasks();
       if (status == null) {
         taskTrackers.remove(trackerName);
       }
@@ -1238,7 +1239,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
     if (status != null) {
       totalMaps += status.countMapTasks();
       totalReduces += status.countReduceTasks();
-      totalTaskCapacity += status.getMaxTasks();
+      totalMapTaskCapacity += status.getMaxMapTasks();
+      totalReduceTaskCapacity += status.getMaxReduceTasks();
       taskTrackers.put(trackerName, status);
     }
     return oldStatus != null;
@@ -1320,17 +1322,17 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
       }   
     }
 
-    int maxCurrentTasks = tts.getMaxTasks();
-    
+    int maxCurrentMapTasks = tts.getMaxMapTasks();
+    int maxCurrentReduceTasks = tts.getMaxReduceTasks();
     // find out the maximum number of maps or reduces that we are willing
     // to run on any node.
     int maxMapLoad = 0;
     int maxReduceLoad = 0;
     if (numTaskTrackers > 0) {
-      maxMapLoad = Math.min(maxCurrentTasks,
+      maxMapLoad = Math.min(maxCurrentMapTasks,
                             (int) Math.ceil((double) remainingMapLoad / 
                                             numTaskTrackers));
-      maxReduceLoad = Math.min(maxCurrentTasks,
+      maxReduceLoad = Math.min(maxCurrentReduceTasks,
                                (int) Math.ceil((double) remainingReduceLoad
                                                / numTaskTrackers));
     }
@@ -1380,10 +1382,10 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
           totalNeededMaps += job.desiredMaps();
           int padding = 0;
           if (numTaskTrackers > MIN_CLUSTER_SIZE_FOR_PADDING) {
-            padding = Math.min(maxCurrentTasks,
+            padding = Math.min(maxCurrentMapTasks,
                                (int)(totalNeededMaps * PAD_FRACTION));
           }
-          if (totalMaps + padding >= totalTaskCapacity) {
+          if (totalMaps + padding >= totalMapTaskCapacity) {
             break;
           }
         }
@@ -1418,10 +1420,10 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
           int padding = 0;
           if (numTaskTrackers > MIN_CLUSTER_SIZE_FOR_PADDING) {
             padding = 
-              Math.min(maxCurrentTasks,
+              Math.min(maxCurrentReduceTasks,
                        (int) (totalNeededReduces * PAD_FRACTION));
           }
-          if (totalReduces + padding >= totalTaskCapacity) {
+          if (totalReduces + padding >= totalReduceTaskCapacity) {
             break;
           }
         }
@@ -1577,7 +1579,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
       return new ClusterStatus(taskTrackers.size(),
                                totalMaps,
                                totalReduces,
-                               totalTaskCapacity,
+                               totalMapTaskCapacity,
+                               totalReduceTaskCapacity, 
                                state);          
     }
   }

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

@@ -324,7 +324,7 @@ class LocalJobRunner implements JobSubmissionProtocol {
   }
   
   public ClusterStatus getClusterStatus() {
-    return new ClusterStatus(1, map_tasks, reduce_tasks, 1, 
+    return new ClusterStatus(1, map_tasks, reduce_tasks, 1, 1, 
                              JobTracker.State.RUNNING);
   }
 

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

@@ -152,7 +152,8 @@ public class TaskTracker
   private static final String JOBCACHE = "jobcache";
   private JobConf fConf;
   private MapOutputFile mapOutputFile;
-  private int maxCurrentTasks;
+  private int maxCurrentMapTasks;
+  private int maxCurrentReduceTasks;
   private int failures;
   private int finishedCount[] = new int[1];
   private MapEventsFetcherThread mapEventsFetcher;
@@ -259,7 +260,9 @@ public class TaskTracker
         if (metricsRecord != null) {
           metricsRecord.setMetric("maps_running", mapTotal);
           metricsRecord.setMetric("reduces_running", reduceTotal);
-          metricsRecord.setMetric("taskSlots", (short)maxCurrentTasks);
+          metricsRecord.setMetric("mapTaskSlots", (short)maxCurrentMapTasks);
+          metricsRecord.setMetric("reduceTaskSlots", 
+                                       (short)maxCurrentReduceTasks);
           metricsRecord.incrMetric("tasks_completed", numCompletedTasks);
           metricsRecord.incrMetric("tasks_failed_timeout", timedoutTasks);
           metricsRecord.incrMetric("tasks_failed_ping", tasksFailedPing);
@@ -414,8 +417,10 @@ public class TaskTracker
       this.fConf.get("mapred.task.tracker.report.bindAddress", "127.0.0.1");
 
     // RPC initialization
+    int max = maxCurrentMapTasks > maxCurrentReduceTasks ? 
+                       maxCurrentMapTasks : maxCurrentReduceTasks;
     this.taskReportServer =
-      RPC.getServer(this, bindAddress, 0, maxCurrentTasks, false, this.fConf);
+      RPC.getServer(this, bindAddress, 0, max, false, this.fConf);
     this.taskReportServer.start();
 
     // get the assigned address
@@ -693,11 +698,46 @@ public class TaskTracker
     this.mapEventsFetcher.interrupt();
   }
 
+  /**
+   * Handles deprecated "mapred.tasktracker.tasks.maximum" 
+   * @param newMax new max values specified through 
+   * mapred.tasktracker.map.tasks.maximum or 
+   * mapred.tasktracker.reduce.tasks.maximum
+   * @param oldMax old max value specified through 
+   * mapred.tasktracker.tasks.maximum
+   * @param def default value if max tasks not specified at all.
+   * @return new value supercedes old value. If both new and old values 
+   * are not set, default value is returned.
+   */
+  private int handleDeprecatedMaxTasks(String newMax, 
+                                       String oldMax,
+                                       int def) {
+    try {
+      if (newMax != null) {
+        return Integer.parseInt(newMax);
+      }
+      if (oldMax != null ) {
+        LOG.warn("mapred.tasktracker.tasks.maximum is deprecated. Use " +
+                 "mapred.tasktracker.map.tasks.maximum and " +
+                 "mapred.tasktracker.reduce.tasks.maximum instead.");
+        return Integer.parseInt(oldMax);
+      }
+    } catch (NumberFormatException ne) {
+      return def;
+    }
+    return def;  
+  }
+  
   /**
    * Start with the local machine name, and the default JobTracker
    */
   public TaskTracker(JobConf conf) throws IOException {
-    maxCurrentTasks = conf.getInt("mapred.tasktracker.tasks.maximum", 2);
+    maxCurrentMapTasks = handleDeprecatedMaxTasks(
+                           conf.get("mapred.tasktracker.map.tasks.maximum"),
+                           conf.get("mapred.tasktracker.tasks.maximum"), 2);
+    maxCurrentReduceTasks = handleDeprecatedMaxTasks(
+                         conf.get("mapred.tasktracker.reduce.tasks.maximum"),
+                         conf.get("mapred.tasktracker.tasks.maximum"), 2);
     this.fConf = conf;
     this.jobTrackAddr = JobTracker.getAddress(conf);
     this.mapOutputFile = new MapOutputFile();
@@ -868,7 +908,9 @@ public class TaskTracker
       synchronized (this) {
         status = new TaskTrackerStatus(taskTrackerName, localHostname, 
                                        httpPort, cloneAndResetRunningTaskStatuses(), 
-                                       failures, maxCurrentTasks); 
+                                       failures, 
+                                       maxCurrentMapTasks,
+                                       maxCurrentReduceTasks); 
       }
     } else {
       LOG.info("Resending 'status' to '" + jobTrackAddr.getHostName() +
@@ -881,8 +923,8 @@ public class TaskTracker
     boolean askForNewTask;
     long localMinSpaceStart;
     synchronized (this) {
-      askForNewTask = (mapTotal < maxCurrentTasks || 
-                       reduceTotal < maxCurrentTasks) &&
+      askForNewTask = (mapTotal < maxCurrentMapTasks || 
+                       reduceTotal < maxCurrentReduceTasks) &&
                       acceptNewTasks; 
       localMinSpaceStart = minSpaceStart;
     }

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

@@ -46,7 +46,8 @@ class TaskTrackerStatus implements Writable {
   List<TaskStatus> taskReports;
     
   volatile long lastSeen;
-  int maxTasks;
+  private int maxMapTasks;
+  private int maxReduceTasks;
     
   /**
    */
@@ -58,15 +59,16 @@ class TaskTrackerStatus implements Writable {
    */
   public TaskTrackerStatus(String trackerName, String host, 
                            int httpPort, List<TaskStatus> taskReports, 
-                           int failures, int maxTasks) {
+                           int failures, int maxMapTasks,
+                           int maxReduceTasks) {
     this.trackerName = trackerName;
     this.host = host;
     this.httpPort = httpPort;
 
     this.taskReports = new ArrayList<TaskStatus>(taskReports);
     this.failures = failures;
-
-    this.maxTasks = maxTasks;
+    this.maxMapTasks = maxMapTasks;
+    this.maxReduceTasks = maxReduceTasks;
   }
 
   /**
@@ -158,10 +160,12 @@ class TaskTrackerStatus implements Writable {
    * and 1 reduce concurrently).
    * @return maximum tasks this node supports
    */
-  public int getMaxTasks() {
-    return maxTasks;
+  public int getMaxMapTasks() {
+    return maxMapTasks;
   }
-  
+  public int getMaxReduceTasks() {
+    return maxReduceTasks;
+  }  
   ///////////////////////////////////////////
   // Writable
   ///////////////////////////////////////////
@@ -170,8 +174,8 @@ class TaskTrackerStatus implements Writable {
     UTF8.writeString(out, host);
     out.writeInt(httpPort);
     out.writeInt(failures);
-    out.writeInt(maxTasks);
-
+    out.writeInt(maxMapTasks);
+    out.writeInt(maxReduceTasks);
     out.writeInt(taskReports.size());
     for (TaskStatus taskStatus : taskReports) {
       TaskStatus.writeTaskStatus(out, taskStatus);
@@ -183,8 +187,8 @@ class TaskTrackerStatus implements Writable {
     this.host = UTF8.readString(in);
     this.httpPort = in.readInt();
     this.failures = in.readInt();
-    this.maxTasks = in.readInt();
-
+    this.maxMapTasks = in.readInt();
+    this.maxReduceTasks = in.readInt();
     taskReports.clear();
     int numTasks = in.readInt();
     for (int i = 0; i < numTasks; i++) {

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

@@ -472,9 +472,12 @@ public class SortValidator {
           jobConf.getInt("test.sortvalidate.maps_per_host", 10);
       }
       if (noReduces == -1) {
-        noReduces = cluster.getTaskTrackers() * 
-          jobConf.getInt("test.sortvalidate.reduces_per_host", 
-                         cluster.getMaxTasks());
+        noReduces = (int) (cluster.getMaxReduceTasks() * 0.9);
+        String sortReduces = jobConf.get("test.sortvalidate.reduces_per_host");
+        if (sortReduces != null) {
+           noReduces = cluster.getTaskTrackers() * 
+                           Integer.parseInt(sortReduces);
+        }
       }
       jobConf.setNumMapTasks(noMaps);
       jobConf.setNumReduceTasks(noReduces);

src/webapps/job/jobtracker.jsp

@@ -68,19 +68,22 @@
                                    JobTracker tracker) throws IOException {
     ClusterStatus status = tracker.getClusterStatus();
     String tasksPerNode = status.getTaskTrackers() > 0 ?
-      percentFormat.format(((double)status.getMaxTasks()) / status.getTaskTrackers()) :
+      percentFormat.format(((double)(status.getMaxMapTasks() +
+                      status.getMaxReduceTasks())) / status.getTaskTrackers()):
       "-";
     out.print("<table border=\"2\" cellpadding=\"5\" cellspacing=\"2\">\n"+
               "<tr><th>Maps</th><th>Reduces</th>" + 
               "<th>Total Submissions</th>" +
-              "<th>Nodes</th><th>Task Capacity</th><th>Avg. Tasks/Node</th></tr>\n");
+              "<th>Nodes</th><th>Map Task Capacity</th>" +
+              "<th>Reduce Task Capacity</th><th>Avg. Tasks/Node</th></tr>\n");
     out.print("<tr><td>" + status.getMapTasks() + "</td><td>" +
               status.getReduceTasks() + "</td><td>" + 
               tracker.getTotalSubmissions() +
               "</td><td><a href=\"machines.jsp\">" +
               status.getTaskTrackers() +
-              "</a></td><td>" + status.getMaxTasks() +
-	      "</td><td>" + tasksPerNode +
+              "</a></td><td>" + status.getMaxMapTasks() +
+              "</td><td>" + status.getMaxReduceTasks() +
+              "</td><td>" + tasksPerNode +
               "</td></tr></table>\n");
   }%>
 

src/webapps/job/machines.jsp

@@ -25,7 +25,9 @@
       out.print("<table border=\"2\" cellpadding=\"5\" cellspacing=\"2\">\n");
       out.print("<tr><td align=\"center\" colspan=\"6\"><b>Task Trackers</b></td></tr>\n");
       out.print("<tr><td><b>Name</b></td><td><b>Host</b></td>" +
-                "<td><b># running tasks</b></td><td><b>Max Tasks</b></td>" +
+                "<td><b># running tasks</b></td>" +
+                "<td><b>Max Map Tasks</b></td>" +
+                "<td><b>Max Reduce Tasks</b></td>" +
                 "<td><b>Failures</b></td>" +
                 "<td><b>Seconds since heartbeat</b></td></tr>\n");
       int maxFailures = 0;
@@ -50,7 +52,8 @@
         out.print(tt.getHost() + ":" + tt.getHttpPort() + "/\">");
         out.print(tt.getTrackerName() + "</a></td><td>");
         out.print(tt.getHost() + "</td><td>" + numCurTasks +
-                  "</td><td>" + tt.getMaxTasks() + 
+                  "</td><td>" + tt.getMaxMapTasks() +
+                  "</td><td>" + tt.getMaxReduceTasks() + 
                   "</td><td>" + numFailures + 
                   "</td><td>" + sinceHeartbeat + "</td></tr>\n");
       }