[SPARK-6418] Add simple per-stage visualization to the UI [WIP]

core/src/main/resources/org/apache/spark/ui/static/jobs-graph.js

@@ -0,0 +1,138 @@
+/* memoize the formatted graph data */
+var chart;
+var svg;
+var tableData;
+var divisorAndTitle;
+var numTasks;
+var divisor;
+var xTitle;
+
+function renderJobsGraphs(data) {
+  /* show visualization toggle */
+  $(".expand-visualization-arrow").toggleClass('arrow-closed');
+  $(".expand-visualization-arrow").toggleClass('arrow-open');
+  if ($(".expand-visualization-arrow").hasClass("arrow-closed")) {
+    $("#chartContainer").empty();
+    return;
+  }
+
+  /* no data to graph */
+  if (!Object.keys(data).length) {
+    return;
+  }
+
+  /* format data for dimple.js */
+  if (!tableData) {
+    tableData = [];
+    var startTime = getMin(data["Launch Time"]);
+    numTasks = Math.min(1000, data["Launch Time"].length);
+
+    /*data update */
+    data["Launch Time"] = data["Launch Time"].map(function(launchTime) {
+      return launchTime - startTime;
+    });
+    var maxTime = 0;
+    for (i = 0; i < numTasks; i++) {
+      var time = 0;
+      for (var key in data) {
+        time += data[key][i];
+      }
+      maxTime = Math.max(time, maxTime);
+    }
+    setDivisiorAndTitle(maxTime);
+    for (i = 0; i < numTasks; i++) {
+      for (var key in data) {
+        job = {};
+        job["Task #"] = i;
+        job["Task"] = key;
+        job["Time"] = data[key][i] / divisor;
+        tableData.push(job);
+      }
+    }
+  }
+
+  var height = Math.max(Math.min(numTasks * 50, 2000), 200);
+  svg = dimple.newSvg("#chartContainer", "100%", height);
+  chart = new dimple.chart(svg);
+  chart.setMargins(60, 80, 60, 60);
+
+  var x = chart.addMeasureAxis("x", "Time");
+  x.fontSize = "12px";
+  x.title = xTitle;
+
+  var y = chart.addCategoryAxis("y", "Task #");
+  y.fontSize = "12px";
+
+  var s = chart.addSeries("Task", dimple.plot.bar);
+  s.data = tableData;
+  s.addOrderRule(getOrderRule());
+
+  chart.addLegend(20, 10, "80%", 60, "left");
+  (chart.legends[0]).fontSize = "12px";
+
+  s.getTooltipText = function(dat) {
+    return ["Task #: " + dat["yField"][0],
+      "Phase: " + dat["aggField"][0],
+      "Time (ms): " + dat["xValue"] * divisor
+    ];
+  };
+
+  chart.draw(100);
+  svg.selectAll(".dimple-launch-time").remove();
+  numTicks(y, Math.floor(numTasks / 20));
+}
+
+function getMin(arr) {
+  return Math.min.apply(null, arr);
+}
+
+function getOrderRule() {
+  return ["Launch Time", "Scheduler Delay", "Task Deserialization Time",
+    "Duration", "Result Serialization Time", "Getting Result Time", "GC Time"
+  ];
+}
+
+function setDivisiorAndTitle(maxTime) {
+  var sec = 1000;
+  var min = sec * 60;
+  var hr = min * 60;
+  if (maxTime >= hr) {
+    divisor = hr;
+    xTitle = "Time (hr)";
+  } else if (maxTime >= min) {
+    divisor = min;
+    xTitle = "Time (min)";
+  } else if (maxTime >= sec) {
+    divisor = sec;
+    xTitle = "Time (s)";
+  } else {
+    divisor = 1;
+    xTitle = "Time (ms)";
+  }
+}
+
+/* limits the number of ticks in the Y-axis to oneInEvery */
+function numTicks(axis, oneInEvery) {
+  if (axis.shapes.length > 0) {
+    var del = 0;
+    if (oneInEvery > 1) {
+      axis.shapes.selectAll("text").each(function(d) {
+        if (del % oneInEvery !== 0) {
+          this.remove();
+          axis.shapes.selectAll("line").each(function(d2) {
+            if (d === d2) {
+              this.remove();
+            }
+          });
+        }
+        del += 1;
+      });
+    }
+  }
+}
+
+window.onresize = function() {
+  if ($(".expand-visualization-arrow").hasClass("arrow-open")) {
+    chart.draw(0, true);
+  }
+};

core/src/main/resources/org/apache/spark/ui/static/webui.css

@@ -149,7 +149,7 @@ pre {
   border: none;
 }
 
-span.expand-additional-metrics {
+span.expand-additional-metrics, span.expand-visualization {
   cursor: pointer;
 }
 

core/src/main/scala/org/apache/spark/ui/UIUtils.scala

@@ -165,6 +165,9 @@ private[spark] object UIUtils extends Logging {
     <script src={prependBaseUri("/static/initialize-tooltips.js")}></script>
     <script src={prependBaseUri("/static/table.js")}></script>
     <script src={prependBaseUri("/static/additional-metrics.js")}></script>
+    <script src={prependBaseUri("/static/d3.min.js")}></script>
+    <script src={prependBaseUri("/static/dimple.min.js")}></script>
+    <script src={prependBaseUri("/static/jobs-graph.js")}></script>
   }
 
   /** Returns a spark page with correctly formatted headers */

core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala

@@ -21,6 +21,7 @@ import java.util.Date
 import javax.servlet.http.HttpServletRequest
 
 import scala.xml.{Elem, Node, Unparsed}
+import scala.util.parsing.json.{JSONArray, JSONObject}
 
 import org.apache.commons.lang3.StringEscapeUtils
 
@@ -58,6 +59,7 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
 
       val stageData = stageDataOption.get
       val tasks = stageData.taskData.values.toSeq.sortBy(_.taskInfo.launchTime)
+      val graphData = scala.collection.mutable.Map[String, JSONArray]()
 
       val numCompleted = tasks.count(_.taskInfo.finished)
       val accumulables = listener.stageIdToData((stageId, stageAttemptId)).accumulables
@@ -234,6 +236,8 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
           val deserializationTimes = validTasks.map { case TaskUIData(_, metrics, _) =>
             metrics.get.executorDeserializeTime.toDouble
           }
+          graphData("Task Deserialization Time") = JSONArray(deserializationTimes.toList)
+
           val deserializationQuantiles =
             <td>
               <span data-toggle="tooltip" title={ToolTips.TASK_DESERIALIZATION_TIME}
@@ -250,6 +254,8 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
           val gcTimes = validTasks.map { case TaskUIData(_, metrics, _) =>
             metrics.get.jvmGCTime.toDouble
           }
+          graphData("GC Time") = JSONArray(gcTimes.toList)
+
           val gcQuantiles =
             <td>
               <span data-toggle="tooltip"
@@ -260,6 +266,8 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
           val serializationTimes = validTasks.map { case TaskUIData(_, metrics, _) =>
             metrics.get.resultSerializationTime.toDouble
           }
+          graphData("Result Serialization Time") = JSONArray(serializationTimes.toList)
+
           val serializationQuantiles =
             <td>
               <span data-toggle="tooltip"
@@ -271,6 +279,8 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
           val gettingResultTimes = validTasks.map { case TaskUIData(info, _, _) =>
             getGettingResultTime(info).toDouble
           }
+          graphData("Getting Result Time") = JSONArray(gettingResultTimes.toList)
+
           val gettingResultQuantiles =
             <td>
               <span data-toggle="tooltip"
@@ -285,6 +295,18 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
           val schedulerDelays = validTasks.map { case TaskUIData(info, metrics, _) =>
             getSchedulerDelay(info, metrics.get).toDouble
           }
+          graphData("Scheduler Delay") = JSONArray(schedulerDelays.toList)
+
+          val launchTimes = validTasks.map { case TaskUIData(info, metrics, _) =>
+            info.launchTime
+          }
+          graphData("Launch Time") = JSONArray(launchTimes.toList)
+
+          val durations = validTasks.map { case TaskUIData(info, metrics, _) => if (info.status == "RUNNING") 
+            info.timeRunning(System.currentTimeMillis()) else metrics.map(_.executorRunTime).getOrElse(1L)
+          }
+          graphData("Duration") = JSONArray(durations.toList)
+
           val schedulerDelayTitle = <td><span data-toggle="tooltip"
             title={ToolTips.SCHEDULER_DELAY} data-placement="right">Scheduler Delay</span></td>
           val schedulerDelayQuantiles = schedulerDelayTitle +:
@@ -431,13 +453,28 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
       val maybeAccumulableTable: Seq[Node] =
         if (accumulables.size > 0) { <h4>Accumulators</h4> ++ accumulableTable } else Seq()
 
+      val graphJSON = JSONObject(graphData.toMap)
+
+      val showVisualization =
+        <div>
+          <span class="expand-visualization" onclick="render();">
+            <span class="expand-visualization-arrow arrow-closed"></span>
+            <strong>Show Visualization</strong>
+          </span>
+          <div id="chartContainer" class="container"></div>
+          <script type="text/javascript">
+            {Unparsed(s"function render() {renderJobsGraphs(${graphJSON})}")}
+          </script>
+        </div>
+
       val content =
         summary ++
         showAdditionalMetrics ++
         <h4>Summary Metrics for {numCompleted} Completed Tasks</h4> ++
         <div>{summaryTable.getOrElse("No tasks have reported metrics yet.")}</div> ++
         <h4>Aggregated Metrics by Executor</h4> ++ executorTable.toNodeSeq ++
         maybeAccumulableTable ++
+        showVisualization ++
         <h4>Tasks</h4> ++ taskTable
 
       UIUtils.headerSparkPage("Details for Stage %d".format(stageId), content, parent)