Skip to content

Commit

Permalink
Make the X axis use the same range
Browse files Browse the repository at this point in the history
  • Loading branch information
@zsxwing
zsxwing committed Apr 23, 2015
1 parent 881c907 commit d78672a
Show file tree
Hide file tree
Showing 2 changed files with 68 additions and 47 deletions.
14 changes: 12 additions & 2 deletions core/src/main/resources/org/apache/spark/ui/static/streaming-page.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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;
}
101 changes: 56 additions & 45 deletions streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingPage.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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>
}
}

Expand All @@ -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>
}
}

Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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(
Expand All @@ -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(
Expand All @@ -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(
Expand All @@ -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
Expand All @@ -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">
Expand All @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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>
Expand All @@ -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
}
Expand Down Expand Up @@ -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>
}
}

0 comments on commit d78672a

Please sign in to comment.