HDOOP-11456: Change Dr. Elephant's heuristic from MapperInputSize to …

…MapperTime

app-conf/HeuristicConf.xml

@@ -16,9 +16,9 @@
 
   <heuristic>
     <applicationtype>mapreduce</applicationtype>
-    <heuristicname>Mapper Input Size</heuristicname>
-    <classname>com.linkedin.drelephant.mapreduce.heuristics.MapperInputSizeHeuristic</classname>
-    <viewname>views.html.helpMapperInputSize</viewname>
+    <heuristicname>Mapper Time</heuristicname>
+    <classname>com.linkedin.drelephant.mapreduce.heuristics.MapperTimeHeuristic</classname>
+    <viewname>views.html.helpMapperTime</viewname>
     <hadoopversions>
       <version>1</version>
       <version>2</version>

app/com/linkedin/drelephant/ElephantRunner.java

@@ -5,7 +5,6 @@
 import com.linkedin.drelephant.analysis.AnalyticJobGeneratorHadoop1;
 import com.linkedin.drelephant.analysis.HadoopSystemContext;
 import com.linkedin.drelephant.analysis.AnalyticJobGeneratorHadoop2;
-import com.linkedin.drelephant.notifications.EmailThread;
 import java.io.IOException;
 import java.security.PrivilegedAction;
 import java.util.List;
@@ -26,7 +25,6 @@ public class ElephantRunner implements Runnable {
   private static final Logger logger = Logger.getLogger(ElephantRunner.class);
   private AtomicBoolean _running = new AtomicBoolean(true);
   private long lastRun;
-  private EmailThread _emailer = new EmailThread();
   private HadoopSecurity _hadoopSecurity;
   private ExecutorService _service;
   private BlockingQueue<AnalyticJob> _jobQueue;
@@ -61,7 +59,6 @@ public void run() {
         public Void run() {
           //TODO: We need to catch exception here
           HadoopSystemContext.load();
-          _emailer.start();
           loadAnalyticJobGenerator();
           ElephantContext.init();
 
@@ -171,7 +168,6 @@ private void waitInterval() {
 
   public void kill() {
     _running.set(false);
-    _emailer.kill();
     _service.shutdownNow();
   }
 }

app/com/linkedin/drelephant/mapreduce/heuristics/MapperTimeHeuristic.java

@@ -0,0 +1,89 @@
+package com.linkedin.drelephant.mapreduce.heuristics;
+
+import com.linkedin.drelephant.mapreduce.MapReduceApplicationData;
+import com.linkedin.drelephant.mapreduce.MapReduceCounterHolder;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.linkedin.drelephant.analysis.HadoopSystemContext;
+import com.linkedin.drelephant.analysis.Heuristic;
+import com.linkedin.drelephant.analysis.HeuristicResult;
+import com.linkedin.drelephant.analysis.Severity;
+import com.linkedin.drelephant.mapreduce.MapReduceTaskData;
+import com.linkedin.drelephant.math.Statistics;
+
+import org.apache.commons.io.FileUtils;
+
+
+public class MapperTimeHeuristic implements Heuristic<MapReduceApplicationData> {
+  public static final String HEURISTIC_NAME = "Mapper Time";
+
+  @Override
+  public String getHeuristicName() {
+    return HEURISTIC_NAME;
+  }
+
+  @Override
+  public HeuristicResult apply(MapReduceApplicationData data) {
+    MapReduceTaskData[] tasks = data.getMapperData();
+
+    List<Long> inputBytes = new ArrayList<Long>();
+    List<Long> runtimesMs = new ArrayList<Long>();
+
+    for (MapReduceTaskData task : tasks) {
+      inputBytes.add(task.getCounters().get(MapReduceCounterHolder.CounterName.HDFS_BYTES_READ));
+      if (task.timed()) {
+        runtimesMs.add(task.getTotalRunTimeMs());
+      }
+    }
+
+    long averageSize = Statistics.average(inputBytes);
+    long averageTimeMs = Statistics.average(runtimesMs);
+
+    Severity shortTaskSeverity = shortTaskSeverity(tasks.length, averageTimeMs);
+    Severity longTaskSeverity = longTaskSeverity(tasks.length, averageTimeMs);
+    Severity severity = Severity.max(shortTaskSeverity, longTaskSeverity);
+
+    HeuristicResult result = new HeuristicResult(HEURISTIC_NAME, severity);
+
+    result.addDetail("Number of tasks", Integer.toString(tasks.length));
+    result.addDetail("Average task input size", FileUtils.byteCountToDisplaySize(averageSize));
+    result.addDetail("Average task runtime", Statistics.readableTimespan(averageTimeMs));
+
+    return result;
+  }
+
+  private Severity shortTaskSeverity(long numTasks, long averageTimeMs) {
+    // We want to identify jobs with short task runtime
+    Severity severity = getShortRuntimeSeverity(averageTimeMs);
+    // Severity is reduced if number of tasks is small.
+    Severity numTaskSeverity = getNumTasksSeverity(numTasks);
+    return Severity.min(severity, numTaskSeverity);
+  }
+
+  private Severity longTaskSeverity(long numTasks, long averageTimeMs) {
+    // We want to identify jobs with long task runtime
+    Severity severity = getLongRuntimeSeverity(averageTimeMs);
+    // Severity is reduced if number of tasks is large
+    Severity numTaskSeverity = getNumTasksSeverityReverse(numTasks);
+    return Severity.min(severity, numTaskSeverity);
+  }
+
+  public static Severity getShortRuntimeSeverity(long runtimeMs) {
+    return Severity.getSeverityDescending(runtimeMs, 10 * Statistics.MINUTE_IN_MS, 4 * Statistics.MINUTE_IN_MS,
+        2 * Statistics.MINUTE_IN_MS, 1 * Statistics.MINUTE_IN_MS);
+  }
+
+  public static Severity getLongRuntimeSeverity(long runtimeMs) {
+    return Severity.getSeverityAscending(runtimeMs, 15 * Statistics.MINUTE_IN_MS, 30 * Statistics.MINUTE_IN_MS, 1 * Statistics.HOUR_IN_MS,
+        2 * Statistics.HOUR_IN_MS);
+  }
+
+  public static Severity getNumTasksSeverity(long numTasks) {
+    return Severity.getSeverityAscending(numTasks, 50, 101, 500, 1000);
+  }
+
+  public static Severity getNumTasksSeverityReverse(long numTasks) {
+    return Severity.getSeverityDescending(numTasks, 500, 100, 50, 10);
+  }
+}

app/com/linkedin/drelephant/mapreduce/heuristics/ReducerTimeHeuristic.java

@@ -48,18 +48,20 @@ public HeuristicResult apply(MapReduceApplicationData data) {
 
   private Severity shortTimeSeverity(long runtimeMs, long numTasks) {
     Severity timeSeverity = getShortRuntimeSeverity(runtimeMs);
+    // Severity is adjusted based on number of tasks
     Severity taskSeverity = getNumTasksSeverity(numTasks);
     return Severity.min(timeSeverity, taskSeverity);
   }
 
   private Severity longTimeSeverity(long runtimeMs, long numTasks) {
     Severity timeSeverity = getLongRuntimeSeverity(runtimeMs);
+    // Severity is adjusted based on number of tasks
     Severity taskSeverity = getNumTasksSeverityReverse(numTasks);
     return Severity.min(timeSeverity, taskSeverity);
   }
 
   public static Severity getShortRuntimeSeverity(long runtimeMs) {
-    return Severity.getSeverityDescending(runtimeMs, 10 * Statistics.MINUTE_IN_MS, 5 * Statistics.MINUTE_IN_MS,
+    return Severity.getSeverityDescending(runtimeMs, 10 * Statistics.MINUTE_IN_MS, 4 * Statistics.MINUTE_IN_MS,
         2 * Statistics.MINUTE_IN_MS, 1 * Statistics.MINUTE_IN_MS);
   }
 
@@ -69,10 +71,10 @@ public static Severity getLongRuntimeSeverity(long runtimeMs) {
   }
 
   public static Severity getNumTasksSeverity(long numTasks) {
-    return Severity.getSeverityAscending(numTasks, 10, 51, 200, 500);
+    return Severity.getSeverityAscending(numTasks, 50, 101, 500, 1000);
   }
 
   public static Severity getNumTasksSeverityReverse(long numTasks) {
-    return Severity.getSeverityDescending(numTasks, 100, 49, 20, 10);
+    return Severity.getSeverityDescending(numTasks, 500, 100, 50, 10);
   }
 }

app/views/helpMapperInputSize.scala.html → app/views/helpMapperTime.scala.html

@@ -3,13 +3,13 @@
     This should allow you to better tweak the number of mappers for your job.<br>
     There are two possible situations that needs some tuning.
 </p>
-<h3>Too many small files</h3>
+<h3>Mapper time too short</h3>
 <p>
     This usually happens when the Hadoop job has:
 <ul>
     <li>A <u>large</u> number of mappers</li>
+    <li><u>Short</u> mapper avg runtime</li>
     <li><u>Small</u> file size</li>
-    <li><u>Short</u> mapper runtime</li>
 </ul>
 </p>
 <h5>Example</h5>
@@ -22,7 +22,7 @@ <h4 class="list-group-item-heading">Mapper Input Size</h4>
             <tbody>
             <tr>
                 <td>Number of tasks</td>
-                <td>916</td>
+                <td>1516</td>
             </tr>
             <tr>
                 <td>Average task input size</td>
@@ -39,10 +39,9 @@ <h4 class="list-group-item-heading">Mapper Input Size</h4>
 </p>
 <h4>Suggestions</h4>
 <p>
-    Note: By default HadoopJava will not combine small files. if that’s the case, you should either use CombineFileInputFormat or use Pig/Hive.
-   <br>
+
    <br>
-    Otherwise, you should tune mapper split size to reduce number of mappers and let each mapper process larger data <br>
+    You should tune mapper split size to reduce number of mappers and let each mapper process larger data <br>
     The parameter for changing split size is: <br>
     <ul>
     <li>mapred.max/min.split.size (Hadoop 1)</li>
@@ -59,6 +58,8 @@ <h4>Suggestions</h4>
     The split size is controlled by formula <u>max(minSplitSize, min(maxSplitSize, blockSize))</u>. By default, blockSize=512MB and minSplit < blockSize < maxSplit. <br>
     You should always refer to this formula.<br>
     In the case above, try <u>increase min split size</u> and let each mapper process larger data.<br>
+    <strong>[Note]</strong> By default HadoopJava will not combine small files, so each mapper cannot process more than one file, and changing split size won't help. If that is your case, you should either try CombineFileInputFormat or use Pig/Hive.
+   <br>
     See <a href="http://go/hadooptuningtips">Hadoop Tuning Tips</a> for further information.<br>
 
 </p>
@@ -67,8 +68,8 @@ <h3>Large files/Unsplittable files</h3>
     This usually happens when the Hadoop job has:
 <ul>
     <li>A <u>small</u> number of mappers</li>
+    <li><u>Long</u> mapper avg runtime</li>
     <li><u>Large</u> file size (a few GB's)</li>
-    <li><u>Long</u> mapper runtime</li>
 </ul>
 </p>
 <h5>Example</h5>
@@ -81,15 +82,15 @@ <h4 class="list-group-item-heading">Mapper Input Size</h4>
             <tbody>
             <tr>
                 <td>Number of tasks</td>
-                <td>100</td>
+                <td>50</td>
             </tr>
             <tr>
                 <td>Average task input</td>
                 <td>3 GB</td>
             </tr>
             <tr>
                 <td>Average task runtime</td>
-                <td>32 min 30s</td>
+                <td>52 min 30s</td>
             </tr>
             </tbody>
         </table>