Make the X axis use the same range

Author: zsxwing

core/src/main/resources/org/apache/spark/ui/static/streaming-page.js

@@ -16,6 +16,10 @@
  */
 
 
+var globalMinX = 0;
+var globalMaxX = 0;
+var binCount = 10;
+
 // An invisible div to show details of a point in the graph
 var graphTooltip = d3.select("body").append("div")
     .style("position", "absolute")
@@ -137,14 +141,14 @@ function drawDistribution(id, values, minY, maxY, unitY) {
     var height = 150 - margin.top - margin.bottom;
 
     //var binCount = values.length > 100 ? 100 : values.length;
-    var binCount = 10;
     var formatBinValue = d3.format(",.2f");
 
     var y = d3.scale.linear().domain([minY, maxY]).range([height, 0]);
     var data = d3.layout.histogram().range([minY, maxY]).bins(binCount)(values);
 
     var x = d3.scale.linear()
-        .domain([0, d3.max(data, function(d) { return d.y; })])
+        .domain([globalMinX, globalMaxX])
+        //.domain([0, d3.max(data, function(d) { return d.y; })])
         .range([0, width]);
 
     var xAxis = d3.svg.axis().scale(x).orient("bottom").ticks(5);
@@ -197,3 +201,9 @@ function drawDistribution(id, values, minY, maxY, unitY) {
                       hideGraphTooltip();
                   });
 }
+
+function prepareDistribution(values, minY, maxY) {
+    var data = d3.layout.histogram().range([minY, maxY]).bins(binCount)(values);
+    var maxBarSize = d3.max(data, function(d) { return d.y; });
+    globalMaxX = maxBarSize > globalMaxX? maxBarSize : globalMaxX;
+}

streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingPage.scala

@@ -37,13 +37,14 @@ import org.apache.spark.util.Distribution
 private[ui] case class TimelineUIData(divId: String, data: Seq[(Long, _)], minX: Long, maxX: Long,
     minY: Long, maxY: Long, unitY: String) {
 
-  def toHtmlAndJs: (Seq[Node], String) = {
+  def toHtml(jsCollector: JsCollector): Seq[Node] = {
     val jsForData = data.map { case (x, y) =>
       s"""{"x": $x, "y": $y}"""
     }.mkString("[", ",", "]")
-
-    (<div id={divId}></div>,
+    jsCollector.addStatement(
       s"drawTimeline('#$divId', $jsForData, $minX, $maxX, $minY, $maxY, '$unitY');")
+
+    <div id={divId}></div>
   }
 }
 
@@ -57,11 +58,12 @@ private[ui] case class TimelineUIData(divId: String, data: Seq[(Long, _)], minX:
 private[ui] case class DistributionUIData(
     divId: String, data: Seq[_], minY: Long, maxY: Long, unitY: String) {
 
-  def toHtmlAndJs: (Seq[Node], String) = {
+  def toHtml(jsCollector: JsCollector): Seq[Node] = {
     val jsForData = data.mkString("[", ",", "]")
+    jsCollector.addPreparedStatement(s"prepareDistribution($jsForData, $minY, $maxY);")
+    jsCollector.addStatement(s"drawDistribution('#$divId', $jsForData, $minY, $maxY, '$unitY');")
 
-    (<div id={divId}></div>,
-      s"drawDistribution('#$divId', $jsForData, $minY, $maxY, '$unitY');")
+    <div id={divId}></div>
   }
 }
 
@@ -149,7 +151,7 @@ private[ui] class StreamingPage(parent: StreamingTab)
       batchInfo.totalDelay.map(batchInfo.batchTime.milliseconds -> _)
     })
 
-    val jsCollector = ArrayBuffer[String]()
+    val jsCollector = new JsCollector
 
     // Use the max value of "schedulingDelay", "processingTime", and "totalDelay" to make the
     // Y axis ranges same.
@@ -179,17 +181,15 @@ private[ui] class StreamingPage(parent: StreamingTab)
         maxBatchTime,
         minEventRate,
         maxEventRate,
-        "events/sec").toHtmlAndJs
-    jsCollector += timelineDataForEventRateOfAllReceivers._2
+        "events/sec").toHtml(jsCollector)
 
     val distributionDataForEventRateOfAllReceivers =
       DistributionUIData(
         "all-receiver-events-distribution",
         eventRateForAllReceivers.data.map(_._2),
         minEventRate,
         maxEventRate,
-        "events/sec").toHtmlAndJs
-    jsCollector += distributionDataForEventRateOfAllReceivers._2
+        "events/sec").toHtml(jsCollector)
 
     val timelineDataForSchedulingDelay =
       TimelineUIData(
@@ -199,17 +199,15 @@ private[ui] class StreamingPage(parent: StreamingTab)
         maxBatchTime,
         minTime,
         maxTime,
-        "ms").toHtmlAndJs
-    jsCollector += timelineDataForSchedulingDelay._2
+        "ms").toHtml(jsCollector)
 
     val distributionDataForSchedulingDelay =
       DistributionUIData(
         "scheduling-delay-distribution",
         schedulingDelay.data.map(_._2),
         minTime,
         maxTime,
-        "ms").toHtmlAndJs
-    jsCollector += distributionDataForSchedulingDelay._2
+        "ms").toHtml(jsCollector)
 
     val timelineDataForProcessingTime =
       TimelineUIData(
@@ -219,17 +217,15 @@ private[ui] class StreamingPage(parent: StreamingTab)
         maxBatchTime,
         minTime,
         maxTime,
-        "ms").toHtmlAndJs
-    jsCollector += timelineDataForProcessingTime._2
+        "ms").toHtml(jsCollector)
 
     val distributionDataForProcessingTime =
       DistributionUIData(
         "processing-time-distribution",
         processingTime.data.map(_._2),
         minTime,
         maxTime,
-        "ms").toHtmlAndJs
-    jsCollector += distributionDataForProcessingTime._2
+        "ms").toHtml(jsCollector)
 
     val timelineDataForTotalDelay =
       TimelineUIData(
@@ -239,17 +235,15 @@ private[ui] class StreamingPage(parent: StreamingTab)
         maxBatchTime,
         minTime,
         maxTime,
-        "ms").toHtmlAndJs
-    jsCollector += timelineDataForTotalDelay._2
+        "ms").toHtml(jsCollector)
 
     val distributionDataForTotalDelay =
       DistributionUIData(
         "total-delay-distribution",
         totalDelay.data.map(_._2),
         minTime,
         maxTime,
-        "ms").toHtmlAndJs
-    jsCollector += distributionDataForTotalDelay._2
+        "ms").toHtml(jsCollector)
 
     val table =
       // scalastyle:off
@@ -266,8 +260,8 @@ private[ui] class StreamingPage(parent: StreamingTab)
             </div>
             <div>Avg: {eventRateForAllReceivers.avg.map(_.formatted("%.2f")).getOrElse(emptyCell)} events/sec</div>
           </td>
-          <td>{timelineDataForEventRateOfAllReceivers._1}</td>
-          <td>{distributionDataForEventRateOfAllReceivers._1}</td>
+          <td>{timelineDataForEventRateOfAllReceivers}</td>
+          <td>{distributionDataForEventRateOfAllReceivers}</td>
         </tr>
         <tr id="inputs-table" style="display: none;" >
           <td colspan="3">
@@ -279,39 +273,34 @@ private[ui] class StreamingPage(parent: StreamingTab)
             <div><strong>Scheduling Delay</strong></div>
             <div>Avg: {formatDurationOption(schedulingDelay.avg)}</div>
           </td>
-          <td>{timelineDataForSchedulingDelay._1}</td>
-          <td>{distributionDataForSchedulingDelay._1}</td>
+          <td>{timelineDataForSchedulingDelay}</td>
+          <td>{distributionDataForSchedulingDelay}</td>
         </tr>
         <tr>
           <td style="vertical-align: middle;">
             <div><strong>Processing Time</strong></div>
             <div>Avg: {formatDurationOption(processingTime.avg)}</div>
           </td>
-          <td>{timelineDataForProcessingTime._1}</td>
-          <td>{distributionDataForProcessingTime._1}</td>
+          <td>{timelineDataForProcessingTime}</td>
+          <td>{distributionDataForProcessingTime}</td>
         </tr>
         <tr>
           <td style="vertical-align: middle;">
             <div><strong>Total Delay</strong></div>
             <div>Avg: {formatDurationOption(totalDelay.avg)}</div>
           </td>
-          <td>{timelineDataForTotalDelay._1}</td>
-          <td>{distributionDataForTotalDelay._1}</td>
+          <td>{timelineDataForTotalDelay}</td>
+          <td>{distributionDataForTotalDelay}</td>
         </tr>
       </tbody>
     </table>
     // scalastyle:on
 
-    val js =
-      s"""
-         |$$(document).ready(function(){
-         |    ${jsCollector.mkString("\n")}
-         |});""".stripMargin
-    table ++ <script>{Unparsed(js)}</script>
+    table ++ jsCollector.toHtml
   }
 
   private def generateInputReceiversTable(
-      jsCollector: ArrayBuffer[String],
+      jsCollector: JsCollector,
       minX: Long,
       maxX: Long,
       minY: Long,
@@ -337,7 +326,7 @@ private[ui] class StreamingPage(parent: StreamingTab)
   }
 
   private def generateInputReceiverRow(
-      jsCollector: ArrayBuffer[String],
+      jsCollector: JsCollector,
       receiverId: Int,
       distribution: Option[Distribution],
       minX: Long,
@@ -367,17 +356,15 @@ private[ui] class StreamingPage(parent: StreamingTab)
         maxX,
         minY,
         maxY,
-        "events/sec").toHtmlAndJs
-    jsCollector += timelineForEventRate._2
+        "events/sec").toHtml(jsCollector)
 
     val distributionForEventsRate =
       DistributionUIData(
         s"receiver-$receiverId-events-distribution",
         receivedRecords.map(_._2),
         minY,
         maxY,
-        "events/sec").toHtmlAndJs
-    jsCollector += distributionForEventsRate._2
+        "events/sec").toHtml(jsCollector)
 
     // scalastyle:off
     <tr>
@@ -394,9 +381,9 @@ private[ui] class StreamingPage(parent: StreamingTab)
     </tr>
       <tr>
         <td colspan="3">
-          {timelineForEventRate._1}
+          {timelineForEventRate}
         </td>
-        <td>{distributionForEventsRate._1}</td>
+        <td>{distributionForEventsRate}</td>
       </tr>
     // scalastyle:on
   }
@@ -435,3 +422,27 @@ private object StreamingPage {
   val BLACK_DOWN_TRIANGLE_HTML = "&#9660;"
 }
 
+private[ui] class JsCollector {
+  private val preparedStatements = ArrayBuffer[String]()
+  private val statements = ArrayBuffer[String]()
+
+  def addPreparedStatement(js: String): Unit = {
+    preparedStatements += js
+  }
+
+  def addStatement(js: String): Unit = {
+    statements += js
+  }
+
+  def toHtml: Seq[Node] = {
+    val js =
+      s"""
+         |$$(document).ready(function(){
+         |    ${preparedStatements.mkString("\n")}
+         |    ${statements.mkString("\n")}
+         |});""".stripMargin
+
+   <script>{Unparsed(js)}</script>
+  }
+}
+