From bd443420777e87f109f0ded7a9e07af50ab5ebd9 Mon Sep 17 00:00:00 2001
From: Akshay Rai
Date: Wed, 3 Sep 2014 03:42:59 -0700
Subject: [PATCH 001/272] HADOOP-6636 : Dr. Elephant heuristic to detect tasks
nearing their queue limit. Currently disabled. It will be enabled through
pluggable heuristics later.
---
.../drelephant/analysis/HeuristicResult.java | 4 +
.../heuristics/JobQueueLimitHeuristic.java | 96 +++++++++++++++++++
app/controllers/Application.java | 5 +-
app/views/helpJobQueueLimit.scala.html | 55 +++++++++++
.../JobQueueLimitHeuristicTest.java | 68 +++++++++++++
5 files changed, 226 insertions(+), 2 deletions(-)
create mode 100644 app/com/linkedin/drelephant/analysis/heuristics/JobQueueLimitHeuristic.java
create mode 100644 app/views/helpJobQueueLimit.scala.html
create mode 100644 test/com/linkedin/drelephant/analysis/heuristics/JobQueueLimitHeuristicTest.java
diff --git a/app/com/linkedin/drelephant/analysis/HeuristicResult.java b/app/com/linkedin/drelephant/analysis/HeuristicResult.java
index c716f665d..3a18ade3e 100644
--- a/app/com/linkedin/drelephant/analysis/HeuristicResult.java
+++ b/app/com/linkedin/drelephant/analysis/HeuristicResult.java
@@ -63,4 +63,8 @@ public void addDetail(String... parts) {
detailsColumns = parts.length;
}
}
+
+ public void setSeverity(Severity severity){
+ this.severity = severity;
+ }
}
diff --git a/app/com/linkedin/drelephant/analysis/heuristics/JobQueueLimitHeuristic.java b/app/com/linkedin/drelephant/analysis/heuristics/JobQueueLimitHeuristic.java
new file mode 100644
index 000000000..533ea22ba
--- /dev/null
+++ b/app/com/linkedin/drelephant/analysis/heuristics/JobQueueLimitHeuristic.java
@@ -0,0 +1,96 @@
+package com.linkedin.drelephant.analysis.heuristics;
+
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import com.linkedin.drelephant.analysis.Heuristic;
+import com.linkedin.drelephant.analysis.HeuristicResult;
+import com.linkedin.drelephant.analysis.Severity;
+import com.linkedin.drelephant.hadoop.HadoopJobData;
+import com.linkedin.drelephant.hadoop.HadoopTaskData;
+
+
+public class JobQueueLimitHeuristic implements Heuristic {
+ public static final String heuristicName = "Job Queue Timeout Limit";
+
+ @Override
+ public String getHeuristicName() {
+ return heuristicName;
+ }
+
+ @Override
+ public HeuristicResult apply(HadoopJobData data) {
+ HeuristicResult result = new HeuristicResult(heuristicName, Severity.NONE);
+ Properties jobConf = data.getJobConf();
+ long queueTimeoutLimit = TimeUnit.MINUTES.toMillis(15);
+
+ // Fetch the Queue to which the job is submitted.
+ String queueName = jobConf.getProperty("mapred.job.queue.name");
+ if (queueName == null) {
+ throw new IllegalStateException("Queue Name not found.");
+ }
+
+ // Compute severity if job is submitted to default queue else set severity to NONE.
+ HadoopTaskData[] mapTasks = data.getMapperData();
+ HadoopTaskData[] redTasks = data.getReducerData();
+ Severity[] mapTasksSeverity = new Severity[mapTasks.length];
+ Severity[] redTasksSeverity = new Severity[redTasks.length];
+ if (queueName.equals("default")) {
+ result.addDetail("Queue: ", queueName);
+ result.addDetail("Number of Map tasks", Integer.toString(mapTasks.length));
+ result.addDetail("Number of Reduce tasks", Integer.toString(redTasks.length));
+
+ // Calculate Severity of Mappers
+ mapTasksSeverity = getTasksSeverity(mapTasks, queueTimeoutLimit);
+ result.addDetail("Number of Map tasks that are in severe state (14 to 14.5 min)",
+ Long.toString(getSeverityFrequency(Severity.SEVERE, mapTasksSeverity)));
+ result.addDetail("Number of Map tasks that are in critical state (over 14.5 min)",
+ Long.toString(getSeverityFrequency(Severity.CRITICAL, mapTasksSeverity)));
+
+ // Calculate Severity of Reducers
+ redTasksSeverity = getTasksSeverity(redTasks, queueTimeoutLimit);
+ result.addDetail("Number of Reduce tasks that are in severe state (14 to 14.5 min)",
+ Long.toString(getSeverityFrequency(Severity.SEVERE, redTasksSeverity)));
+ result.addDetail("Number of Reduce tasks that are in critical state (over 14.5 min)",
+ Long.toString(getSeverityFrequency(Severity.CRITICAL, redTasksSeverity)));
+
+ // Calculate Job severity
+ result.setSeverity(Severity.max(Severity.max(mapTasksSeverity), Severity.max(redTasksSeverity)));
+
+ } else {
+ result.addDetail("This Heuristic is not applicable to " + queueName + " queue");
+ result.setSeverity(Severity.NONE);
+ }
+ return result;
+ }
+
+ private Severity[] getTasksSeverity(HadoopTaskData[] tasks, long queueTimeout) {
+ Severity[] tasksSeverity = new Severity[tasks.length];
+ int i = 0;
+ for (HadoopTaskData task : tasks) {
+ tasksSeverity[i] = getQueueLimitSeverity(task.getRunTime(), queueTimeout);
+ i++;
+ }
+ return tasksSeverity;
+ }
+
+ private long getSeverityFrequency(Severity severity, Severity[] tasksSeverity) {
+ long count = 0;
+ for (Severity taskSeverity : tasksSeverity) {
+ if (taskSeverity.equals(severity)) {
+ count++;
+ }
+ }
+ return count;
+ }
+
+ private Severity getQueueLimitSeverity(long taskTime, long queueTimeout) {
+ long timeUnit = TimeUnit.SECONDS.toMillis(30); // 30s
+ if (queueTimeout == 0) {
+ return Severity.NONE;
+ }
+ return Severity.getSeverityAscending(taskTime, queueTimeout - 4 * timeUnit, queueTimeout - 3 * timeUnit,
+ queueTimeout - 2 * timeUnit, queueTimeout - timeUnit);
+ }
+
+}
diff --git a/app/controllers/Application.java b/app/controllers/Application.java
index 86be3d406..8e7e03fec 100644
--- a/app/controllers/Application.java
+++ b/app/controllers/Application.java
@@ -9,10 +9,8 @@
import java.util.List;
import java.util.Map;
-
import model.JobResult;
import views.html.*;
-
import play.api.templates.Html;
import play.data.DynamicForm;
import play.data.Form;
@@ -23,6 +21,7 @@
import com.avaje.ebean.ExpressionList;
import com.linkedin.drelephant.ElephantAnalyser;
import com.linkedin.drelephant.analysis.Severity;
+import com.linkedin.drelephant.analysis.heuristics.JobQueueLimitHeuristic;
import com.linkedin.drelephant.analysis.heuristics.MapperDataSkewHeuristic;
import com.linkedin.drelephant.analysis.heuristics.MapperInputSizeHeuristic;
import com.linkedin.drelephant.analysis.heuristics.MapperSpeedHeuristic;
@@ -138,6 +137,8 @@ public static Result help() {
page = helpReducerTime.render();
} else if (topic.equals(ShuffleSortHeuristic.heuristicName)) {
page = helpShuffleSort.render();
+ } else if (topic.equals(JobQueueLimitHeuristic.heuristicName)) {
+ page = helpJobQueueLimit.render();
} else if (topic.equals(ElephantAnalyser.NO_DATA)) {
page = helpNoData.render();
}
diff --git a/app/views/helpJobQueueLimit.scala.html b/app/views/helpJobQueueLimit.scala.html
new file mode 100644
index 000000000..ad426b461
--- /dev/null
+++ b/app/views/helpJobQueueLimit.scala.html
@@ -0,0 +1,55 @@
+
The task execution limit on the Default queue is 15 minutes (900 seconds). You really, really want to be clear of
+that limit or your job may fail as data grows or if you get assigned some slow nodes. So while running a job on the
+default queue make sure each task runs between 5 to 15 minutes.
+
This analysis shows an indication of how long your tasks have taken to run on the Default Queue.
Number of Map tasks that are in severe state (14 to 14.5 min)
+
3
+
+
+
Number of Map tasks that are in critical state (over 14.5 min)
+
0
+
+
+
Number of Reduce tasks that are in severe state (14 to 14.5 min)
+
5
+
+
+
Number of Reduce tasks that are in critical state (over 14.5 min)
+
1
+
+
+
+
+
+
+
Suggestions
+
+ If a Job has task times approaching 15 minutes then you must tune down your task runtimes(usually by increasing the
+ number of tasks) or by moving it to the marathon queue. The marathon queue is built for long running jobs so that
+ faster jobs can finish in a reasonable time without having to worry about resources being available.
+
\ No newline at end of file
diff --git a/test/com/linkedin/drelephant/analysis/heuristics/JobQueueLimitHeuristicTest.java b/test/com/linkedin/drelephant/analysis/heuristics/JobQueueLimitHeuristicTest.java
new file mode 100644
index 000000000..d65dc052a
--- /dev/null
+++ b/test/com/linkedin/drelephant/analysis/heuristics/JobQueueLimitHeuristicTest.java
@@ -0,0 +1,68 @@
+package com.linkedin.drelephant.analysis.heuristics;
+
+import java.io.IOException;
+import java.util.Properties;
+
+import org.junit.Test;
+
+import junit.framework.TestCase;
+
+import com.linkedin.drelephant.analysis.Constants;
+import com.linkedin.drelephant.analysis.Heuristic;
+import com.linkedin.drelephant.analysis.HeuristicResult;
+import com.linkedin.drelephant.analysis.Severity;
+import com.linkedin.drelephant.hadoop.HadoopCounterHolder;
+import com.linkedin.drelephant.hadoop.HadoopJobData;
+import com.linkedin.drelephant.hadoop.HadoopTaskData;
+
+
+public class JobQueueLimitHeuristicTest extends TestCase {
+
+ Heuristic heuristic = new JobQueueLimitHeuristic();
+ private static final int numTasks = Constants.SHUFFLE_SORT_MAX_SAMPLE_SIZE;
+
+ @Test
+ public void testRuntimeCritical() throws IOException {
+ assertEquals(Severity.CRITICAL, analyzeJob((long) (14.5 * 60 * 1000), "default"));
+ }
+
+ public void testRuntimeSevere() throws IOException {
+ assertEquals(Severity.SEVERE, analyzeJob(14 * 60 * 1000, "default"));
+ }
+
+ public void testRuntimeModerate() throws IOException {
+ assertEquals(Severity.MODERATE, analyzeJob((long) (13.5 * 60 * 1000), "default"));
+ }
+
+ public void testRuntimeLow() throws IOException {
+ assertEquals(Severity.LOW, analyzeJob(13 * 60 * 1000, "default"));
+ }
+
+ public void testRuntimeNone() throws IOException {
+ assertEquals(Severity.NONE, analyzeJob(12 * 60 * 1000, "default"));
+ }
+
+ public void testNonDefaultRuntimeNone() throws IOException {
+ assertEquals(Severity.NONE, analyzeJob(15 * 60 * 1000, "non-default"));
+ }
+
+ private Severity analyzeJob(long runtime, String queueName) throws IOException {
+ HadoopCounterHolder dummyCounter = new HadoopCounterHolder(null);
+ HadoopTaskData[] mappers = new HadoopTaskData[2 * numTasks / 3];
+ HadoopTaskData[] reducers = new HadoopTaskData[numTasks / 3];
+ Properties jobConf = new Properties();
+ jobConf.put("mapred.job.queue.name", queueName);
+ int i = 0;
+ for (; i < 2 * numTasks / 3; i++) {
+ mappers[i] = new HadoopTaskData(dummyCounter, new long[] { 0, runtime, 0, 0 });
+ }
+ for (i = 0; i < numTasks / 3; i++) {
+ reducers[i] = new HadoopTaskData(dummyCounter, new long[] { 0, runtime, 0, 0 });
+ }
+ HadoopJobData data =
+ new HadoopJobData().setCounters(dummyCounter).setReducerData(reducers).setMapperData(mappers)
+ .setJobConf(jobConf);
+ HeuristicResult result = heuristic.apply(data);
+ return result.getSeverity();
+ }
+}
From 3d9223cee9f389252f4d9732ac439ad3cd95a39d Mon Sep 17 00:00:00 2001
From: Mark Wagner
Date: Tue, 9 Sep 2014 14:10:37 -0700
Subject: [PATCH 002/272] HADOOP-7028: Update version to 0.5-SNAPSHOT
---
build.sbt | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/build.sbt b/build.sbt
index 1bf6c1ebe..59b0d0167 100644
--- a/build.sbt
+++ b/build.sbt
@@ -2,7 +2,7 @@ import play.Project._
name := "dr-elephant"
-version := "0.4"
+version := "0.5-SNAPSHOT"
javacOptions in Compile ++= Seq("-source", "1.6", "-target", "1.6")
From 87defee68fb56c43c87734eb047a2d6a8dc15bd1 Mon Sep 17 00:00:00 2001
From: fli
Date: Fri, 12 Sep 2014 14:50:59 -0700
Subject: [PATCH 003/272] HADOOP-6642: Enforce code style conventions for Dr.
Elephant
---
app/Global.java | 10 +-
app/com/linkedin/drelephant/DrElephant.java | 10 +-
.../linkedin/drelephant/ElephantAnalyser.java | 114 +++----
.../linkedin/drelephant/ElephantFetcher.java | 6 +-
.../drelephant/ElephantFetcherClassic.java | 62 ++--
.../drelephant/ElephantFetcherYarn.java | 128 ++++----
.../linkedin/drelephant/ElephantRunner.java | 33 +-
.../linkedin/drelephant/InfoExtractor.java | 6 +-
.../drelephant/analysis/Constants.java | 27 +-
.../drelephant/analysis/Heuristic.java | 5 +-
.../drelephant/analysis/HeuristicResult.java | 105 +++----
.../drelephant/analysis/Severity.java | 157 +++++-----
.../heuristics/GenericDataSkewHeuristic.java | 99 +++---
.../heuristics/JobQueueLimitHeuristic.java | 6 +-
.../heuristics/MapperDataSkewHeuristic.java | 19 +-
.../heuristics/MapperInputSizeHeuristic.java | 103 +++----
.../heuristics/MapperSpeedHeuristic.java | 111 ++++---
.../heuristics/ReducerDataSkewHeuristic.java | 17 +-
.../heuristics/ReducerTimeHeuristic.java | 105 +++----
.../heuristics/ShuffleSortHeuristic.java | 98 +++---
.../hadoop/HadoopCounterHolder.java | 140 +++++----
.../drelephant/hadoop/HadoopJobData.java | 54 ++--
.../drelephant/hadoop/HadoopSecurity.java | 30 +-
.../drelephant/hadoop/HadoopTaskData.java | 104 +++----
.../linkedin/drelephant/math/Statistics.java | 289 +++++++++---------
.../drelephant/notifications/EmailThread.java | 151 ++++-----
app/com/linkedin/drelephant/util/Utils.java | 118 +++----
app/controllers/Application.java | 53 ++--
app/model/JobHeuristicResult.java | 48 +--
app/model/JobResult.java | 11 +-
app/model/JobType.java | 27 +-
app/model/StringResult.java | 3 +-
app/views/help.scala.html | 2 +-
app/views/search.scala.html | 2 +-
.../JobQueueLimitHeuristicTest.java | 14 +-
.../MapperDataSkewHeuristicTest.java | 24 +-
.../MapperInputSizeHeuristicTest.java | 28 +-
.../heuristics/MapperSpeedHeuristicTest.java | 30 +-
.../ReducerDataSkewHeuristicTest.java | 26 +-
.../heuristics/ReducerTimeHeuristicTest.java | 28 +-
.../heuristics/ShuffleSortHeuristicTest.java | 39 +--
41 files changed, 1225 insertions(+), 1217 deletions(-)
diff --git a/app/Global.java b/app/Global.java
index 02b04175d..ed2ca80fd 100644
--- a/app/Global.java
+++ b/app/Global.java
@@ -13,7 +13,7 @@
public class Global extends GlobalSettings {
- DrElephant drElephant;
+ DrElephant _drElephant;
public void onStart(Application app) {
Logger.info("Application has started");
@@ -21,8 +21,8 @@ public void onStart(Application app) {
fixJavaKerberos();
try {
- drElephant = new DrElephant();
- drElephant.start();
+ _drElephant = new DrElephant();
+ _drElephant.start();
} catch (IOException e) {
Logger.error("Application start failed...", e);
}
@@ -30,8 +30,8 @@ public void onStart(Application app) {
public void onStop(Application app) {
Logger.info("Application shutdown...");
- if (drElephant != null) {
- drElephant.kill();
+ if (_drElephant != null) {
+ _drElephant.kill();
}
}
diff --git a/app/com/linkedin/drelephant/DrElephant.java b/app/com/linkedin/drelephant/DrElephant.java
index c471029cf..df34b9f7b 100644
--- a/app/com/linkedin/drelephant/DrElephant.java
+++ b/app/com/linkedin/drelephant/DrElephant.java
@@ -4,20 +4,20 @@
public class DrElephant extends Thread {
- private ElephantRunner elephant;
+ private ElephantRunner _elephant;
public DrElephant() throws IOException {
- elephant = new ElephantRunner();
+ _elephant = new ElephantRunner();
}
@Override
public void run() {
- elephant.run();
+ _elephant.run();
}
public void kill() {
- if (elephant != null) {
- elephant.kill();
+ if (_elephant != null) {
+ _elephant.kill();
}
}
}
diff --git a/app/com/linkedin/drelephant/ElephantAnalyser.java b/app/com/linkedin/drelephant/ElephantAnalyser.java
index 6ad090dfa..28d2c6034 100644
--- a/app/com/linkedin/drelephant/ElephantAnalyser.java
+++ b/app/com/linkedin/drelephant/ElephantAnalyser.java
@@ -9,75 +9,79 @@
import com.linkedin.drelephant.analysis.Heuristic;
import com.linkedin.drelephant.analysis.HeuristicResult;
import com.linkedin.drelephant.analysis.Severity;
-import com.linkedin.drelephant.analysis.heuristics.*;
+import com.linkedin.drelephant.analysis.heuristics.MapperDataSkewHeuristic;
+import com.linkedin.drelephant.analysis.heuristics.MapperInputSizeHeuristic;
+import com.linkedin.drelephant.analysis.heuristics.MapperSpeedHeuristic;
+import com.linkedin.drelephant.analysis.heuristics.ReducerDataSkewHeuristic;
+import com.linkedin.drelephant.analysis.heuristics.ReducerTimeHeuristic;
+import com.linkedin.drelephant.analysis.heuristics.ShuffleSortHeuristic;
import com.linkedin.drelephant.hadoop.HadoopJobData;
import model.JobType;
-public class ElephantAnalyser {
- public static final String NO_DATA = "No Data Received";
- private static final ElephantAnalyser instance = new ElephantAnalyser();
- private HeuristicResult nodata;
- private List heuristics = new ArrayList();
- public List heuristicNames = new ArrayList();
+public class ElephantAnalyser {
+ public static final String NO_DATA = "No Data Received";
+ private static final ElephantAnalyser ANALYZER = new ElephantAnalyser();
- public ElephantAnalyser() {
- nodata = new HeuristicResult(NO_DATA, Severity.LOW);
- addHeuristic(new MapperDataSkewHeuristic());
- addHeuristic(new ReducerDataSkewHeuristic());
- addHeuristic(new MapperInputSizeHeuristic());
- addHeuristic(new MapperSpeedHeuristic());
- addHeuristic(new ReducerTimeHeuristic());
- addHeuristic(new ShuffleSortHeuristic());
- }
+ private HeuristicResult _nodata;
+ private List _heuristics = new ArrayList();
+ public List _heuristicNames = new ArrayList();
- public void addHeuristic(Heuristic heuristic) {
- heuristics.add(heuristic);
- heuristicNames.add(heuristic.getHeuristicName());
- }
+ public ElephantAnalyser() {
+ _nodata = new HeuristicResult(NO_DATA, Severity.LOW);
+ addHeuristic(new MapperDataSkewHeuristic());
+ addHeuristic(new ReducerDataSkewHeuristic());
+ addHeuristic(new MapperInputSizeHeuristic());
+ addHeuristic(new MapperSpeedHeuristic());
+ addHeuristic(new ReducerTimeHeuristic());
+ addHeuristic(new ShuffleSortHeuristic());
+ }
- public HeuristicResult[] analyse(HadoopJobData data) {
- if (data.getMapperData().length == 0 && data.getReducerData().length == 0) {
- return new HeuristicResult[]{nodata};
- }
+ public void addHeuristic(Heuristic heuristic) {
+ _heuristics.add(heuristic);
+ _heuristicNames.add(heuristic.getHeuristicName());
+ }
- List results = new ArrayList();
- for (Heuristic heuristic : heuristics) {
- results.add(heuristic.apply(data));
- }
- return results.toArray(new HeuristicResult[results.size()]);
+ public HeuristicResult[] analyse(HadoopJobData data) {
+ if (data.getMapperData().length == 0 && data.getReducerData().length == 0) {
+ return new HeuristicResult[] { _nodata };
}
- public JobType getJobType(HadoopJobData data) {
- String pigVersion = data.getJobConf().getProperty("pig.version");
- if (pigVersion != null && !pigVersion.isEmpty()) {
- return JobType.PIG;
- }
- String hiveMapredMode = data.getJobConf().getProperty("hive.mapred.mode");
- if (hiveMapredMode != null && !hiveMapredMode.isEmpty()) {
- return JobType.HIVE;
- }
-
- return JobType.HADOOPJAVA;
+ List results = new ArrayList();
+ for (Heuristic heuristic : _heuristics) {
+ results.add(heuristic.apply(data));
}
+ return results.toArray(new HeuristicResult[results.size()]);
+ }
-
- public Map getMetaUrls(HadoopJobData data) {
- Map result = new HashMap();
- final String prefix = "meta.url.";
- Properties jobConf = data.getJobConf();
- for (Map.Entry