hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From d...@apache.org
Subject svn commit: r781248 - in /hadoop/core/branches/branch-0.20: ./ src/mapred/org/apache/hadoop/mapred/ src/test/org/apache/hadoop/mapred/
Date Wed, 03 Jun 2009 03:32:00 GMT
Author: ddas
Date: Wed Jun  3 03:31:58 2009
New Revision: 781248

URL: http://svn.apache.org/viewvc?rev=781248&view=rev
Log:
HADOOP-5908. Fixes a problem to do with ArithmeticException in the JobTracker when there are
jobs with 0 maps. Contributed by Amar Kamat.

Modified:
    hadoop/core/branches/branch-0.20/CHANGES.txt
    hadoop/core/branches/branch-0.20/src/mapred/org/apache/hadoop/mapred/JobInProgress.java
    hadoop/core/branches/branch-0.20/src/mapred/org/apache/hadoop/mapred/ResourceEstimator.java
    hadoop/core/branches/branch-0.20/src/test/org/apache/hadoop/mapred/TestEmptyJob.java
    hadoop/core/branches/branch-0.20/src/test/org/apache/hadoop/mapred/UtilsForTests.java

Modified: hadoop/core/branches/branch-0.20/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/core/branches/branch-0.20/CHANGES.txt?rev=781248&r1=781247&r2=781248&view=diff
==============================================================================
--- hadoop/core/branches/branch-0.20/CHANGES.txt (original)
+++ hadoop/core/branches/branch-0.20/CHANGES.txt Wed Jun  3 03:31:58 2009
@@ -107,6 +107,9 @@
     HADOOP-5648. Fixes a build issue in not being able to generate gridmix.jar
     in hadoop binary tarball. (Giridharan Kesavan via gkesavan)
 
+    HADOOP-5908. Fixes a problem to do with ArithmeticException in the 
+    JobTracker when there are jobs with 0 maps. (Amar Kamat via ddas)
+
 Release 0.20.0 - 2009-04-15
 
   INCOMPATIBLE CHANGES

Modified: hadoop/core/branches/branch-0.20/src/mapred/org/apache/hadoop/mapred/JobInProgress.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/branch-0.20/src/mapred/org/apache/hadoop/mapred/JobInProgress.java?rev=781248&r1=781247&r2=781248&view=diff
==============================================================================
--- hadoop/core/branches/branch-0.20/src/mapred/org/apache/hadoop/mapred/JobInProgress.java
(original)
+++ hadoop/core/branches/branch-0.20/src/mapred/org/apache/hadoop/mapred/JobInProgress.java
Wed Jun  3 03:31:58 2009
@@ -583,6 +583,14 @@
     return inputLength;
   }
  
+  boolean isCleanupLaunched() {
+    return launchedCleanup;
+  }
+
+  boolean isSetupLaunched() {
+    return launchedSetup;
+  }
+
   /**
    * Get the list of map tasks
    * @return the raw array of maps for this job
@@ -1666,6 +1674,11 @@
                                           final int numUniqueHosts,
                                           final int maxCacheLevel,
                                           final double avgProgress) {
+    if (numMapTasks == 0) {
+      LOG.info("No maps to schedule for " + profile.getJobID());
+      return -1;
+    }
+
     String taskTracker = tts.getTrackerName();
     TaskInProgress tip = null;
     
@@ -1871,6 +1884,11 @@
                                              int clusterSize,
                                              int numUniqueHosts,
                                              double avgProgress) {
+    if (numReduceTasks == 0) {
+      LOG.info("No reduces to schedule for " + profile.getJobID());
+      return -1;
+    }
+
     String taskTracker = tts.getTrackerName();
     TaskInProgress tip = null;
     

Modified: hadoop/core/branches/branch-0.20/src/mapred/org/apache/hadoop/mapred/ResourceEstimator.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/branch-0.20/src/mapred/org/apache/hadoop/mapred/ResourceEstimator.java?rev=781248&r1=781247&r2=781248&view=diff
==============================================================================
--- hadoop/core/branches/branch-0.20/src/mapred/org/apache/hadoop/mapred/ResourceEstimator.java
(original)
+++ hadoop/core/branches/branch-0.20/src/mapred/org/apache/hadoop/mapred/ResourceEstimator.java
Wed Jun  3 03:31:58 2009
@@ -82,7 +82,10 @@
    * @return estimated length of this job's average map output
    */
   long getEstimatedMapOutputSize() {
-    long estimate = getEstimatedTotalMapOutputSize()  / job.desiredMaps();
+    long estimate = 0L;
+    if (job.desiredMaps() > 0) {
+      estimate = getEstimatedTotalMapOutputSize()  / job.desiredMaps();
+    }
     return estimate;
   }
 

Modified: hadoop/core/branches/branch-0.20/src/test/org/apache/hadoop/mapred/TestEmptyJob.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/branch-0.20/src/test/org/apache/hadoop/mapred/TestEmptyJob.java?rev=781248&r1=781247&r2=781248&view=diff
==============================================================================
--- hadoop/core/branches/branch-0.20/src/test/org/apache/hadoop/mapred/TestEmptyJob.java (original)
+++ hadoop/core/branches/branch-0.20/src/test/org/apache/hadoop/mapred/TestEmptyJob.java Wed
Jun  3 03:31:58 2009
@@ -26,6 +26,7 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -47,6 +48,26 @@
 
   MiniMRCluster mr = null;
 
+  /** Committer with cleanup waiting on a signal
+   */
+  static class CommitterWithDelayCleanup extends FileOutputCommitter {
+    @Override
+    public void cleanupJob(JobContext context) throws IOException {
+      Configuration conf = context.getConfiguration();
+      Path share = new Path(conf.get("share"));
+      FileSystem fs = FileSystem.get(conf);
+
+      
+      while (true) {
+        if (fs.exists(share)) {
+          break;
+        }
+        UtilsForTests.waitFor(100);
+      }
+      super.cleanupJob(context);
+    }
+  }
+
   /**
    * Simple method running a MapReduce job with no input data. Used to test that
    * such a job is successful.
@@ -62,8 +83,13 @@
     // create an empty input dir
     final Path inDir = new Path(TEST_ROOT_DIR, "testing/empty/input");
     final Path outDir = new Path(TEST_ROOT_DIR, "testing/empty/output");
+    final Path inDir2 = new Path(TEST_ROOT_DIR, "testing/dummy/input");
+    final Path outDir2 = new Path(TEST_ROOT_DIR, "testing/dummy/output");
+    final Path share = new Path(TEST_ROOT_DIR, "share");
+
     JobConf conf = mr.createJobConf();
     FileSystem fs = FileSystem.get(fileSys, conf);
+    fs.delete(new Path(TEST_ROOT_DIR), true);
     fs.delete(outDir, true);
     if (!fs.mkdirs(inDir)) {
       LOG.warn("Can't create " + inDir);
@@ -75,6 +101,7 @@
     conf.setJobName("empty");
     // use an InputFormat which returns no split
     conf.setInputFormat(EmptyInputFormat.class);
+    conf.setOutputCommitter(CommitterWithDelayCleanup.class);
     conf.setOutputKeyClass(Text.class);
     conf.setOutputValueClass(IntWritable.class);
     conf.setMapperClass(IdentityMapper.class);
@@ -83,11 +110,53 @@
     FileOutputFormat.setOutputPath(conf, outDir);
     conf.setNumMapTasks(numMaps);
     conf.setNumReduceTasks(numReduces);
+    conf.set("share", share.toString());
 
     // run job and wait for completion
     JobClient jc = new JobClient(conf);
     RunningJob runningJob = jc.submitJob(conf);
+    JobInProgress job = mr.getJobTrackerRunner().getJobTracker().getJob(runningJob.getID());
+    
+    while (true) {
+      if (job.isCleanupLaunched()) {
+        LOG.info("Waiting for cleanup to be launched for job " 
+                 + runningJob.getID());
+        break;
+      }
+      UtilsForTests.waitFor(100);
+    }
+    
+    // submit another job so that the map load increases and scheduling happens
+    LOG.info("Launching dummy job ");
+    RunningJob dJob = null;
+    try {
+      JobConf dConf = new JobConf(conf);
+      dConf.setOutputCommitter(FileOutputCommitter.class);
+      dJob = UtilsForTests.runJob(dConf, inDir2, outDir2, 2, 0);
+    } catch (Exception e) {
+      LOG.info("Exception ", e);
+      throw new IOException(e);
+    }
+    
+    while (true) {
+      LOG.info("Waiting for job " + dJob.getID() + " to complete");
+      try {
+        Thread.sleep(100);
+      } catch (InterruptedException e) {
+      }
+      if (dJob.isComplete()) {
+        break;
+      }
+    }
+    
+    // check if the second job is successful
+    assertTrue(dJob.isSuccessful());
+
+    // signal the cleanup
+    fs.create(share).close();
+    
     while (true) {
+      LOG.info("Waiting for job " + runningJob.getID() + " to complete");
       try {
         Thread.sleep(100);
       } catch (InterruptedException e) {
@@ -148,7 +217,7 @@
       throws IOException {
     FileSystem fileSys = null;
     try {
-      final int taskTrackers = 1;
+      final int taskTrackers = 2;
       JobConf conf = new JobConf();
       fileSys = FileSystem.get(conf);
 

Modified: hadoop/core/branches/branch-0.20/src/test/org/apache/hadoop/mapred/UtilsForTests.java
URL: http://svn.apache.org/viewvc/hadoop/core/branches/branch-0.20/src/test/org/apache/hadoop/mapred/UtilsForTests.java?rev=781248&r1=781247&r2=781248&view=diff
==============================================================================
--- hadoop/core/branches/branch-0.20/src/test/org/apache/hadoop/mapred/UtilsForTests.java
(original)
+++ hadoop/core/branches/branch-0.20/src/test/org/apache/hadoop/mapred/UtilsForTests.java
Wed Jun  3 03:31:58 2009
@@ -535,6 +535,12 @@
   // Start a job and return its RunningJob object
   static RunningJob runJob(JobConf conf, Path inDir, Path outDir)
                     throws IOException {
+    return runJob(conf, inDir, outDir, conf.getNumMapTasks(), conf.getNumReduceTasks());
+  }
+
+  // Start a job and return its RunningJob object
+  static RunningJob runJob(JobConf conf, Path inDir, Path outDir, int numMaps, 
+                           int numReds) throws IOException {
 
     FileSystem fs = FileSystem.get(conf);
     fs.delete(outDir, true);
@@ -543,9 +549,11 @@
     }
     String input = "The quick brown fox\n" + "has many silly\n"
         + "red fox sox\n";
-    DataOutputStream file = fs.create(new Path(inDir, "part-0"));
-    file.writeBytes(input);
-    file.close();
+    for (int i = 0; i < numMaps; ++i) {
+      DataOutputStream file = fs.create(new Path(inDir, "part-" + i));
+      file.writeBytes(input);
+      file.close();
+    }    
 
     conf.setInputFormat(TextInputFormat.class);
     conf.setOutputKeyClass(LongWritable.class);
@@ -553,8 +561,8 @@
 
     FileInputFormat.setInputPaths(conf, inDir);
     FileOutputFormat.setOutputPath(conf, outDir);
-    conf.setNumMapTasks(1);
-    conf.setNumReduceTasks(1);
+    conf.setNumMapTasks(numMaps);
+    conf.setNumReduceTasks(numReds);
 
     JobClient jobClient = new JobClient(conf);
     RunningJob job = jobClient.submitJob(conf);



Mime
View raw message