Skip to content

Commit 04c7500

Browse files
zsxwingzsxwing
committed
Make the server and client use the same timezone
1 parent fed8219 commit 04c7500

File tree

3 files changed

+78
-31
lines changed

3 files changed

+78
-31
lines changed

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

Lines changed: 19 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,9 @@
1616
*/
1717

1818

19-
var globalMinX = 0;
20-
var globalMaxX = 0;
19+
var timelineMarginLeft = 50;
20+
var distributionMinX = 0;
21+
var distributionMaxX = 0;
2122
var binCount = 10;
2223

2324
// An invisible div to show details of a point in the graph
@@ -53,19 +54,18 @@ function hideGraphTooltip() {
5354
*/
5455
function drawTimeline(id, data, minX, maxX, minY, maxY, unitY) {
5556
d3.select(d3.select(id).node().parentNode).style("padding", "8px 0 8px 8px").style("border-right", "0px solid white");
56-
var margin = {top: 20, right: 27, bottom: 30, left: 50};
57+
var margin = {top: 20, right: 27, bottom: 30, left: timelineMarginLeft};
5758
var width = 500 - margin.left - margin.right;
5859
var height = 150 - margin.top - margin.bottom;
5960

60-
var x = d3.time.scale().domain([minX, maxX]).range([0, width]);
61+
var x = d3.scale.linear().domain([minX, maxX]).range([0, width]);
6162
var y = d3.scale.linear().domain([minY, maxY]).range([height, 0]);
6263

63-
var timeFormat = d3.time.format("%H:%M:%S")
64-
var xAxis = d3.svg.axis().scale(x).orient("bottom").tickFormat(timeFormat);
64+
var xAxis = d3.svg.axis().scale(x).orient("bottom").tickFormat(function(d) { return timeFormat[d]; });
6565
var yAxis = d3.svg.axis().scale(y).orient("left").ticks(5);
6666

6767
var line = d3.svg.line()
68-
.x(function(d) { return x(new Date(d.x)); })
68+
.x(function(d) { return x(d.x); })
6969
.y(function(d) { return y(d.y); });
7070

7171
var svg = d3.select(id).append("svg")
@@ -106,7 +106,7 @@ function drawTimeline(id, data, minX, maxX, minY, maxY, unitY) {
106106
.attr("cy", function(d) { return y(d.y); })
107107
.attr("r", function(d) { return 3; })
108108
.on('mouseover', function(d) {
109-
var tip = d.y + " " + unitY + " at " + timeFormat(new Date(d.x));
109+
var tip = d.y + " " + unitY + " at " + timeFormat[d.x];
110110
showGraphTooltip(tip, d3.event.pageX + 5, d3.event.pageY - 25);
111111
// show the point
112112
d3.select(this)
@@ -137,15 +137,13 @@ function drawDistribution(id, values, minY, maxY, unitY) {
137137
var width = 300 - margin.left - margin.right;
138138
var height = 150 - margin.top - margin.bottom;
139139

140-
//var binCount = values.length > 100 ? 100 : values.length;
141140
var formatBinValue = d3.format(",.2f");
142141

143142
var y = d3.scale.linear().domain([minY, maxY]).range([height, 0]);
144143
var data = d3.layout.histogram().range([minY, maxY]).bins(binCount)(values);
145144

146145
var x = d3.scale.linear()
147-
.domain([globalMinX, globalMaxX])
148-
//.domain([0, d3.max(data, function(d) { return d.y; })])
146+
.domain([distributionMinX, distributionMaxX])
149147
.range([0, width]);
150148

151149
var xAxis = d3.svg.axis().scale(x).orient("top").ticks(5);
@@ -165,9 +163,6 @@ function drawDistribution(id, values, minY, maxY, unitY) {
165163
svg.append("g")
166164
.attr("class", "y axis")
167165
.call(yAxis)
168-
// .append("text")
169-
// .attr("transform", "translate(0," + (-3) + ")")
170-
// .text(unitY);
171166

172167
var bar = svg.selectAll(".bar")
173168
.data(data)
@@ -199,8 +194,17 @@ function drawDistribution(id, values, minY, maxY, unitY) {
199194
});
200195
}
201196

197+
function prepareTimeline(minY, maxY) {
198+
var y = d3.scale.linear().domain([0, maxY]).tickFormat(5);
199+
console.log(y(maxY));
200+
var numOfChars = y(maxY).length;
201+
var maxPx = numOfChars * 8 + 10;
202+
// Make sure we have enough space to show the ticks in the y axis of timeline
203+
timelineMarginLeft = maxPx > timelineMarginLeft? maxPx : timelineMarginLeft;
204+
}
205+
202206
function prepareDistribution(values, minY, maxY) {
203207
var data = d3.layout.histogram().range([minY, maxY]).bins(binCount)(values);
204208
var maxBarSize = d3.max(data, function(d) { return d.y; });
205-
globalMaxX = maxBarSize > globalMaxX? maxBarSize : globalMaxX;
209+
distributionMaxX = maxBarSize > distributionMaxX? maxBarSize : distributionMaxX;
206210
}

‎streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -196,7 +196,7 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext)
196196
retainedBatches.lastOption
197197
}
198198

199-
def retainedBatches: Seq[BatchInfo] = {
199+
def retainedBatches: Seq[BatchInfo] = synchronized {
200200
(waitingBatchInfos.values.toSeq ++
201201
runningBatchInfos.values.toSeq ++ completedBatchInfos).sortBy(_.batchTime)(Time.ordering)
202202
}

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

Lines changed: 58 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717

1818
package org.apache.spark.streaming.ui
1919

20+
import java.text.SimpleDateFormat
21+
import java.util.Date
2022
import javax.servlet.http.HttpServletRequest
2123

2224
import scala.collection.mutable.ArrayBuffer
@@ -30,6 +32,8 @@ import org.apache.spark.util.Distribution
3032
/**
3133
* @param divId the `id` used in the html `div` tag
3234
* @param data the data for the timeline graph
35+
* @param minX the min value of X axis
36+
* @param maxX the max value of X axis
3337
* @param minY the min value of Y axis
3438
* @param maxY the max value of Y axis
3539
* @param unitY the unit of Y axis
@@ -41,6 +45,7 @@ private[ui] case class TimelineUIData(divId: String, data: Seq[(Long, _)], minX:
4145
val jsForData = data.map { case (x, y) =>
4246
s"""{"x": $x, "y": $y}"""
4347
}.mkString("[", ",", "]")
48+
jsCollector.addPreparedStatement(s"prepareTimeline($minY, $maxY);")
4449
jsCollector.addStatement(
4550
s"drawTimeline('#$divId', $jsForData, $minX, $maxX, $minY, $maxY, '$unitY');")
4651

@@ -67,17 +72,21 @@ private[ui] case class DistributionUIData(
6772
}
6873
}
6974

70-
private[ui] case class LongStreamingUIData(data: Seq[(Long, Long)]) {
75+
private[ui] case class MillisecondsStatUIData(data: Seq[(Long, Long)]) {
7176

7277
val avg: Option[Long] = if (data.isEmpty) None else Some(data.map(_._2).sum / data.size)
7378

79+
val formattedAvg: String = StreamingPage.formatDurationOption(avg)
80+
7481
val max: Option[Long] = if (data.isEmpty) None else Some(data.map(_._2).max)
7582
}
7683

77-
private[ui] case class DoubleStreamingUIData(data: Seq[(Long, Double)]) {
84+
private[ui] case class DoubleStatUIData(data: Seq[(Long, Double)]) {
7885

7986
val avg: Option[Double] = if (data.isEmpty) None else Some(data.map(_._2).sum / data.size)
8087

88+
val formattedAvg: String = avg.map(_.formatted("%.2f")).getOrElse("-")
89+
8190
val max: Option[Double] = if (data.isEmpty) None else Some(data.map(_._2).max)
8291
}
8392

@@ -89,7 +98,6 @@ private[ui] class StreamingPage(parent: StreamingTab)
8998

9099
private val listener = parent.listener
91100
private val startTime = System.currentTimeMillis()
92-
private val emptyCell = "-"
93101

94102
/** Render the page */
95103
def render(request: HttpServletRequest): Seq[Node] = {
@@ -104,6 +112,9 @@ private[ui] class StreamingPage(parent: StreamingTab)
104112
UIUtils.headerSparkPage("Streaming Statistics", content, parent, Some(5000))
105113
}
106114

115+
/**
116+
* Generate html that will load css/js files for StreamingPage
117+
*/
107118
private def generateLoadResources(): Seq[Node] = {
108119
// scalastyle:off
109120
<script src={UIUtils.prependBaseUri("/static/d3.min.js")}></script>
@@ -130,24 +141,34 @@ private[ui] class StreamingPage(parent: StreamingTab)
130141
</div>
131142
}
132143

144+
private def generateTimeMap(times: Seq[Long]): Seq[Node] = {
145+
val dateFormat = new SimpleDateFormat("HH:mm:ss")
146+
val js = "var timeFormat = {};\n" + times.map { time =>
147+
val formattedTime = dateFormat.format(new Date(time))
148+
s"timeFormat[$time] = '$formattedTime';"
149+
}.mkString("\n")
150+
151+
<script>{Unparsed(js)}</script>
152+
}
153+
133154
private def generateStatTable(): Seq[Node] = {
134155
val batchInfos = listener.retainedBatches
135156

136157
val batchTimes = batchInfos.map(_.batchTime.milliseconds)
137158
val minBatchTime = if (batchTimes.isEmpty) startTime else batchTimes.min
138159
val maxBatchTime = if (batchTimes.isEmpty) startTime else batchTimes.max
139160

140-
val eventRateForAllReceivers = DoubleStreamingUIData(batchInfos.map { batchInfo =>
161+
val eventRateForAllReceivers = DoubleStatUIData(batchInfos.map { batchInfo =>
141162
(batchInfo.batchTime.milliseconds, batchInfo.numRecords * 1000.0 / listener.batchDuration)
142163
})
143164

144-
val schedulingDelay = LongStreamingUIData(batchInfos.flatMap { batchInfo =>
165+
val schedulingDelay = MillisecondsStatUIData(batchInfos.flatMap { batchInfo =>
145166
batchInfo.schedulingDelay.map(batchInfo.batchTime.milliseconds -> _)
146167
})
147-
val processingTime = LongStreamingUIData(batchInfos.flatMap { batchInfo =>
168+
val processingTime = MillisecondsStatUIData(batchInfos.flatMap { batchInfo =>
148169
batchInfo.processingDelay.map(batchInfo.batchTime.milliseconds -> _)
149170
})
150-
val totalDelay = LongStreamingUIData(batchInfos.flatMap { batchInfo =>
171+
val totalDelay = MillisecondsStatUIData(batchInfos.flatMap { batchInfo =>
151172
batchInfo.totalDelay.map(batchInfo.batchTime.milliseconds -> _)
152173
})
153174

@@ -167,6 +188,7 @@ private[ui] class StreamingPage(parent: StreamingTab)
167188
val maxEventRate = eventRateForAllReceivers.max.map(_.ceil.toLong).getOrElse(0L)
168189
val minEventRate = 0L
169190

191+
// JavaScript to show/hide the receiver sub table.
170192
val triangleJs =
171193
s"""$$('#inputs-table').toggle('collapsed');
172194
|if ($$(this).html() == '$BLACK_RIGHT_TRIANGLE_HTML')
@@ -258,7 +280,7 @@ private[ui] class StreamingPage(parent: StreamingTab)
258280
<span onclick={Unparsed(triangleJs)}>{Unparsed(BLACK_RIGHT_TRIANGLE_HTML)}</span>
259281
<strong>Input Rate</strong>
260282
</div>
261-
<div>Avg: {eventRateForAllReceivers.avg.map(_.formatted("%.2f")).getOrElse(emptyCell)} events/sec</div>
283+
<div>Avg: {eventRateForAllReceivers.formattedAvg} events/sec</div>
262284
</td>
263285
<td class="timeline">{timelineDataForEventRateOfAllReceivers}</td>
264286
<td class="distribution">{distributionDataForEventRateOfAllReceivers}</td>
@@ -271,23 +293,23 @@ private[ui] class StreamingPage(parent: StreamingTab)
271293
<tr>
272294
<td style="vertical-align: middle;">
273295
<div><strong>Scheduling Delay</strong></div>
274-
<div>Avg: {formatDurationOption(schedulingDelay.avg)}</div>
296+
<div>Avg: {schedulingDelay.formattedAvg}</div>
275297
</td>
276298
<td class="timeline">{timelineDataForSchedulingDelay}</td>
277299
<td class="distribution">{distributionDataForSchedulingDelay}</td>
278300
</tr>
279301
<tr>
280302
<td style="vertical-align: middle;">
281303
<div><strong>Processing Time</strong></div>
282-
<div>Avg: {formatDurationOption(processingTime.avg)}</div>
304+
<div>Avg: {processingTime.formattedAvg}</div>
283305
</td>
284306
<td class="timeline">{timelineDataForProcessingTime}</td>
285307
<td class="distribution">{distributionDataForProcessingTime}</td>
286308
</tr>
287309
<tr>
288310
<td style="vertical-align: middle;">
289311
<div><strong>Total Delay</strong></div>
290-
<div>Avg: {formatDurationOption(totalDelay.avg)}</div>
312+
<div>Avg: {totalDelay.formattedAvg}</div>
291313
</td>
292314
<td class="timeline">{timelineDataForTotalDelay}</td>
293315
<td class="distribution">{distributionDataForTotalDelay}</td>
@@ -296,7 +318,7 @@ private[ui] class StreamingPage(parent: StreamingTab)
296318
</table>
297319
// scalastyle:on
298320

299-
table ++ jsCollector.toHtml
321+
generateTimeMap(batchTimes) ++ table ++ jsCollector.toHtml
300322
}
301323

302324
private def generateInputReceiversTable(
@@ -366,7 +388,6 @@ private[ui] class StreamingPage(parent: StreamingTab)
366388
maxY,
367389
"events/sec").toHtml(jsCollector)
368390

369-
// scalastyle:off
370391
<tr>
371392
<td rowspan="2" style="vertical-align: middle;">
372393
<div>
@@ -385,7 +406,6 @@ private[ui] class StreamingPage(parent: StreamingTab)
385406
</td>
386407
<td class="distribution">{distributionForEventsRate}</td>
387408
</tr>
388-
// scalastyle:on
389409
}
390410

391411
/**
@@ -417,13 +437,33 @@ private[ui] class StreamingPage(parent: StreamingTab)
417437
}
418438
}
419439

420-
private object StreamingPage {
440+
private[ui] object StreamingPage {
421441
val BLACK_RIGHT_TRIANGLE_HTML = "&#9654;"
422442
val BLACK_DOWN_TRIANGLE_HTML = "&#9660;"
443+
444+
val emptyCell = "-"
445+
446+
/**
447+
* Returns a human-readable string representing a duration such as "5 second 35 ms"
448+
*/
449+
def formatDurationOption(msOption: Option[Long]): String = {
450+
msOption.map(formatDurationVerbose).getOrElse(emptyCell)
451+
}
423452
}
424453

454+
/**
455+
* A helper class that allows the user to add JavaScript statements which will be executed when the
456+
* DOM has finished loading.
457+
*/
425458
private[ui] class JsCollector {
459+
/**
460+
* JavaScript statements that will execute before `statements`
461+
*/
426462
private val preparedStatements = ArrayBuffer[String]()
463+
464+
/**
465+
* JavaScript statements that will execute after `preparedStatements`
466+
*/
427467
private val statements = ArrayBuffer[String]()
428468

429469
def addPreparedStatement(js: String): Unit = {
@@ -434,6 +474,9 @@ private[ui] class JsCollector {
434474
statements += js
435475
}
436476

477+
/**
478+
* Generate a html snippet that will execute all scripts when the DOM has finished loading.
479+
*/
437480
def toHtml: Seq[Node] = {
438481
val js =
439482
s"""

0 commit comments

Comments
 (0)