From 0c7b2ebd0b2ce1ed615c06ef76c707538b5d6da4 Mon Sep 17 00:00:00 2001 From: zsxwing Date: Thu, 9 Apr 2015 11:56:58 +0800 Subject: [PATCH 01/13] Add BatchPage to display details of a batch --- .../org/apache/spark/ui/jobs/UIData.scala | 2 +- .../spark/streaming/dstream/DStream.scala | 2 +- .../spark/streaming/scheduler/BatchInfo.scala | 7 + .../spark/streaming/scheduler/Job.scala | 38 ++- .../streaming/scheduler/JobScheduler.scala | 11 +- .../spark/streaming/scheduler/JobSet.scala | 2 +- .../apache/spark/streaming/ui/BatchPage.scala | 224 ++++++++++++++++++ .../ui/StreamingJobProgressListener.scala | 60 ++++- .../spark/streaming/ui/StreamingTab.scala | 4 +- .../spark/streaming/UISeleniumSuite.scala | 21 ++ 10 files changed, 352 insertions(+), 19 deletions(-) create mode 100644 streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/UIData.scala b/core/src/main/scala/org/apache/spark/ui/jobs/UIData.scala index 711a3697bda15..935c8a4f80e7b 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/UIData.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/UIData.scala @@ -24,7 +24,7 @@ import org.apache.spark.util.collection.OpenHashSet import scala.collection.mutable.HashMap -private[jobs] object UIData { +private[spark] object UIData { class ExecutorSummary { var taskTime : Long = 0 diff --git a/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala b/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala index 24f99a2b929f5..83d41f5762444 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala @@ -626,7 +626,7 @@ abstract class DStream[T: ClassTag] ( println("Time: " + time) println("-------------------------------------------") firstNum.take(num).foreach(println) - if (firstNum.size > num) println("...") + if (firstNum.length > num) println("...") println() } } diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/BatchInfo.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/BatchInfo.scala index 92dc113f397ca..9d21be50a566f 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/BatchInfo.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/BatchInfo.scala @@ -58,4 +58,11 @@ case class BatchInfo( */ def totalDelay: Option[Long] = schedulingDelay.zip(processingDelay) .map(x => x._1 + x._2).headOption + + /** + * The number of recorders received by the receivers in this batch. + */ + def numRecords: Long = receivedBlockInfo.map { case (_, infos) => + infos.map(_.numRecords).sum + }.sum } diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/Job.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/Job.scala index 30cf87f5b7dd1..064315177e06f 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/Job.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/Job.scala @@ -25,15 +25,43 @@ import scala.util.Try */ private[streaming] class Job(val time: Time, func: () => _) { - var id: String = _ - var result: Try[_] = null + private var _id: String = _ + private var _outputOpId: Int = _ + private var isSet = false + private var _result: Try[_] = null def run() { - result = Try(func()) + _result = Try(func()) } - def setId(number: Int) { - id = "streaming job " + time + "." + number + def result: Try[_] = { + if (_result == null) { + throw new IllegalStateException("Cannot access result before job finishes") + } + _result + } + + def id: String = { + if (!isSet) { + throw new IllegalStateException("Cannot access id before calling setId") + } + _id + } + + def outputOpId: Int = { + if (!isSet) { + throw new IllegalStateException("Cannot access number before calling setId") + } + _outputOpId + } + + def setOutputOpId(outputOpId: Int) { + if (isSet) { + throw new IllegalStateException("Cannot call setOutputOpId more than once") + } + isSet = true + _id = "streaming job " + time + "." + outputOpId + _outputOpId = outputOpId } override def toString: String = id diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala index 95f1857b4c377..fa4daa9ade385 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala @@ -170,8 +170,10 @@ class JobScheduler(val ssc: StreamingContext) extends Logging { ssc.waiter.notifyError(e) } - private class JobHandler(job: Job) extends Runnable { + private class JobHandler(job: Job) extends Runnable with Logging { def run() { + ssc.sc.setLocalProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, job.time.milliseconds.toString) + ssc.sc.setLocalProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, job.outputOpId.toString) eventActor ! JobStarted(job) // Disable checks for existing output directories in jobs launched by the streaming scheduler, // since we may need to write output to an existing directory during checkpoint recovery; @@ -179,7 +181,14 @@ class JobScheduler(val ssc: StreamingContext) extends Logging { PairRDDFunctions.disableOutputSpecValidation.withValue(true) { job.run() } + ssc.sc.setLocalProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, null) + ssc.sc.setLocalProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, null) eventActor ! JobCompleted(job) } } } + +private[streaming] object JobScheduler { + private[streaming] val BATCH_TIME_PROPERTY_KEY = "spark.streaming.internal.batchTime" + private[streaming] val OUTPUT_OP_ID_PROPERTY_KEY = "spark.streaming.internal.outputOpId" +} diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobSet.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobSet.scala index 5b134877d0b2d..24b3794236ea5 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobSet.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobSet.scala @@ -35,7 +35,7 @@ case class JobSet( private var processingStartTime = -1L // when the first job of this jobset started processing private var processingEndTime = -1L // when the last job of this jobset finished processing - jobs.zipWithIndex.foreach { case (job, i) => job.setId(i) } + jobs.zipWithIndex.foreach { case (job, i) => job.setOutputOpId(i) } incompleteJobs ++= jobs def handleJobStart(job: Job) { diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala new file mode 100644 index 0000000000000..a539fdb9d6b16 --- /dev/null +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala @@ -0,0 +1,224 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.streaming.ui + +import javax.servlet.http.HttpServletRequest + +import org.apache.commons.lang3.StringEscapeUtils +import org.apache.spark.streaming.Time +import org.apache.spark.ui.{UIUtils, WebUIPage} +import org.apache.spark.streaming.ui.StreamingJobProgressListener.{JobId, OutputOpId} +import org.apache.spark.ui.jobs.UIData.JobUIData + +import scala.xml.{NodeSeq, Node} + +class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { + private val streaminglistener = parent.listener + private val sparkListener = parent.ssc.sc.jobProgressListener + + private def columns: Seq[Node] = { + Output Op Id + Description + Duration + Job Id + Duration + Stages: Succeeded/Total + Tasks (for all stages): Succeeded/Total + Last Error + } + + private def makeOutputOpIdRow(outputOpId: OutputOpId, jobs: Seq[JobUIData]): Seq[Node] = { + val jobDurations = jobs.map(job => { + job.submissionTime.map { start => + val end = job.completionTime.getOrElse(System.currentTimeMillis()) + end - start + } + }) + val formattedOutputOpDuration = + if (jobDurations.exists(_ == None)) { + // If any job does not finish, set "formattedOutputOpDuration" to "-" + "-" + } else { + UIUtils.formatDuration(jobDurations.flatMap(x => x).sum) + } + + def makeJobRow(job: JobUIData, isFirstRow: Boolean): Seq[Node] = { + val lastStageInfo = Option(job.stageIds) + .filter(_.nonEmpty) + .flatMap { ids => sparkListener.stageIdToInfo.get(ids.max) } + val lastStageData = lastStageInfo.flatMap { s => + sparkListener.stageIdToData.get((s.stageId, s.attemptId)) + } + + val lastStageName = lastStageInfo.map(_.name).getOrElse("(Unknown Stage Name)") + val lastStageDescription = lastStageData.flatMap(_.description).getOrElse("") + val duration: Option[Long] = { + job.submissionTime.map { start => + val end = job.completionTime.getOrElse(System.currentTimeMillis()) + end - start + } + } + val lastFailureReason = job.stageIds.sorted.reverse.flatMap(sparkListener.stageIdToInfo.get). + dropWhile(_.failureReason == None).take(1). // get the first info that contains failure + flatMap(info => info.failureReason).headOption.getOrElse("") + val formattedDuration = duration.map(d => UIUtils.formatDuration(d)).getOrElse("-") + val detailUrl = s"${UIUtils.prependBaseUri(parent.basePath)}/jobs/job?id=${job.jobId}" + + {if(isFirstRow) { + {outputOpId} + + + {lastStageDescription} + {lastStageName} + + {formattedOutputOpDuration}} + } + + + {job.jobId}{job.jobGroup.map(id => s"($id)").getOrElse("")} + + + + {formattedDuration} + + + {job.completedStageIndices.size}/{job.stageIds.size - job.numSkippedStages} + {if (job.numFailedStages > 0) s"(${job.numFailedStages} failed)"} + {if (job.numSkippedStages > 0) s"(${job.numSkippedStages} skipped)"} + + + {UIUtils.makeProgressBar(started = job.numActiveTasks, completed = job.numCompletedTasks, + failed = job.numFailedTasks, skipped = job.numSkippedTasks, + total = job.numTasks - job.numSkippedTasks)} + + {failureReasonCell(lastFailureReason)} + + } + + makeJobRow(jobs.head, true) ++ (jobs.tail.map(job => makeJobRow(job, false)).flatMap(x => x)) + } + + private def failureReasonCell(failureReason: String): Seq[Node] = { + val isMultiline = failureReason.indexOf('\n') >= 0 + // Display the first line by default + val failureReasonSummary = StringEscapeUtils.escapeHtml4( + if (isMultiline) { + failureReason.substring(0, failureReason.indexOf('\n')) + } else { + failureReason + }) + val details = if (isMultiline) { + // scalastyle:off + + +details + ++ + + // scalastyle:on + } else { + "" + } + {failureReasonSummary}{details} + } + + private def jobsTable(jobInfos: Seq[(OutputOpId, JobId)]): Seq[Node] = { + def getJobData(jobId: JobId): Option[JobUIData] = { + sparkListener.activeJobs.get(jobId).orElse { + sparkListener.completedJobs.find(_.jobId == jobId).orElse { + sparkListener.failedJobs.find(_.jobId == jobId) + } + } + } + + // Group jobInfos by OutputOpId firstly, then sort them. + // E.g., [(0, 1), (1, 3), (0, 2), (1, 4)] => [(0, [1, 2]), (1, [3, 4])] + val outputOpIdWithJobIds: Seq[(OutputOpId, Seq[JobId])] = + jobInfos.groupBy(_._1).toSeq.sortBy(_._1). // sorted by OutputOpId + map { case (outputOpId, jobs) => + (outputOpId, jobs.map(_._2).sortBy(x => x).toSeq)} // sort JobIds for each OutputOpId + sparkListener.synchronized { + val outputOpIdWithJobs: Seq[(OutputOpId, Seq[JobUIData])] = outputOpIdWithJobIds.map { + case (outputOpId, jobIds) => + // Filter out JobIds that don't exist in sparkListener + (outputOpId, jobIds.flatMap(getJobData)) + } + + + + {columns} + + + {outputOpIdWithJobs.map { case (outputOpId, jobs) => makeOutputOpIdRow(outputOpId, jobs)}} + +
+ } + } + + def render(request: HttpServletRequest): Seq[Node] = { + val batchTime = Option(request.getParameter("id")).map(id => Time(id.toLong)).getOrElse { + throw new IllegalArgumentException(s"Missing id parameter") + } + val formattedBatchTime = UIUtils.formatDate(batchTime.milliseconds) + val (batchInfo, jobInfos) = streaminglistener.synchronized { + val _batchInfo = streaminglistener.getBatchInfo(batchTime).getOrElse { + throw new IllegalArgumentException(s"Batch $formattedBatchTime does not exist") + } + val _jobInfos = streaminglistener.getJobInfos(batchTime) + (_batchInfo, _jobInfos) + } + + val formattedSchedulingDelay = + batchInfo.schedulingDelay.map(UIUtils.formatDuration).getOrElse("-") + val formattedProcessingTime = + batchInfo.processingDelay.map(UIUtils.formatDuration).getOrElse("-") + val formattedTotalDelay = batchInfo.totalDelay.map(UIUtils.formatDuration).getOrElse("-") + + val summary: NodeSeq = +
+ +
+ + val content = summary ++ jobInfos.map(jobsTable).getOrElse { +
Cannot find any job for Batch {formattedBatchTime}
+ } + UIUtils.headerSparkPage(s"Details of batch at $formattedBatchTime", content, parent) + } +} diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala index 84f80e638f638..33bf21ee9e162 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala @@ -17,8 +17,11 @@ package org.apache.spark.streaming.ui -import scala.collection.mutable.{Queue, HashMap} +import java.util.Properties +import scala.collection.mutable.{ArrayBuffer, Queue, HashMap} + +import org.apache.spark.scheduler._ import org.apache.spark.streaming.{Time, StreamingContext} import org.apache.spark.streaming.scheduler._ import org.apache.spark.streaming.scheduler.StreamingListenerReceiverStarted @@ -29,7 +32,9 @@ import org.apache.spark.util.Distribution private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) - extends StreamingListener { + extends StreamingListener with SparkListener { + + import StreamingJobProgressListener._ private val waitingBatchInfos = new HashMap[Time, BatchInfo] private val runningBatchInfos = new HashMap[Time, BatchInfo] @@ -40,6 +45,8 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) private var totalProcessedRecords = 0L private val receiverInfos = new HashMap[Int, ReceiverInfo] + private val batchTimeToJobIds = new HashMap[Time, ArrayBuffer[(OutputOpId, JobId)]] + val batchDuration = ssc.graph.batchDuration.milliseconds override def onReceiverStarted(receiverStarted: StreamingListenerReceiverStarted) { @@ -70,9 +77,7 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) runningBatchInfos(batchStarted.batchInfo.batchTime) = batchStarted.batchInfo waitingBatchInfos.remove(batchStarted.batchInfo.batchTime) - batchStarted.batchInfo.receivedBlockInfo.foreach { case (_, infos) => - totalReceivedRecords += infos.map(_.numRecords).sum - } + totalReceivedRecords += batchStarted.batchInfo.numRecords } override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = { @@ -80,12 +85,32 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) waitingBatchInfos.remove(batchCompleted.batchInfo.batchTime) runningBatchInfos.remove(batchCompleted.batchInfo.batchTime) completedBatchInfos.enqueue(batchCompleted.batchInfo) - if (completedBatchInfos.size > batchInfoLimit) completedBatchInfos.dequeue() + if (completedBatchInfos.size > batchInfoLimit) { + val removedBatch = completedBatchInfos.dequeue() + batchTimeToJobIds.remove(removedBatch.batchTime) + } totalCompletedBatches += 1L - batchCompleted.batchInfo.receivedBlockInfo.foreach { case (_, infos) => - totalProcessedRecords += infos.map(_.numRecords).sum - } + totalProcessedRecords += batchCompleted.batchInfo.numRecords + } + } + + override def onJobStart(jobStart: SparkListenerJobStart): Unit = synchronized { + getBatchTimeAndOutputOpId(jobStart.properties).foreach { case (batchTime, outputOpId) => + batchTimeToJobIds.getOrElseUpdate(batchTime, ArrayBuffer[(OutputOpId, JobId)]()) += + outputOpId -> jobStart.jobId + } + } + + private def getBatchTimeAndOutputOpId(properties: Properties): Option[(Time, Int)] = { + val batchTime = properties.getProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY) + if (batchTime == null) { + // Not submitted from JobScheduler + None + } else { + val outputOpId = properties.getProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY) + assert(outputOpId != null) + Some(Time(batchTime.toLong) -> outputOpId.toInt) } } @@ -180,4 +205,21 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) private def extractDistribution(getMetric: BatchInfo => Option[Long]): Option[Distribution] = { Distribution(completedBatchInfos.flatMap(getMetric(_)).map(_.toDouble)) } + + def getBatchInfo(batchTime: Time): Option[BatchInfo] = synchronized { + waitingBatchInfos.get(batchTime).orElse { + runningBatchInfos.get(batchTime).orElse { + completedBatchInfos.find(batch => batch.batchTime == batchTime) + } + } + } + + def getJobInfos(batchTime: Time): Option[Seq[(OutputOpId, JobId)]] = synchronized { + batchTimeToJobIds.get(batchTime).map(_.toList) + } +} + +private[streaming] object StreamingJobProgressListener { + type JobId = Int + type OutputOpId = Int } diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingTab.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingTab.scala index 9a860ea4a6c68..e4039639adbad 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingTab.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingTab.scala @@ -27,14 +27,16 @@ import StreamingTab._ * Spark Web UI tab that shows statistics of a streaming job. * This assumes the given SparkContext has enabled its SparkUI. */ -private[spark] class StreamingTab(ssc: StreamingContext) +private[spark] class StreamingTab(val ssc: StreamingContext) extends SparkUITab(getSparkUI(ssc), "streaming") with Logging { val parent = getSparkUI(ssc) val listener = ssc.progressListener ssc.addStreamingListener(listener) + ssc.sc.addSparkListener(listener) attachPage(new StreamingPage(this)) + attachPage(new BatchPage(this)) parent.attachTab(this) def detach() { diff --git a/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala index 998426ebb82e5..8f5fa8c672f33 100644 --- a/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala +++ b/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala @@ -17,6 +17,8 @@ package org.apache.spark.streaming +import scala.collection.mutable.Queue + import org.openqa.selenium.WebDriver import org.openqa.selenium.htmlunit.HtmlUnitDriver import org.scalatest._ @@ -60,8 +62,17 @@ class UISeleniumSuite ssc } + private def setupStreams(ssc: StreamingContext): Unit = { + val rdds = Queue(ssc.sc.parallelize(1 to 4, 4)) + val inputStream = ssc.queueStream(rdds) + inputStream.foreachRDD(rdd => rdd.foreach(_ => {})) + } + test("attaching and detaching a Streaming tab") { withStreamingContext(newSparkStreamingContext()) { ssc => + setupStreams(ssc) + ssc.start() + val sparkUI = ssc.sparkContext.ui.get eventually(timeout(10 seconds), interval(50 milliseconds)) { @@ -75,6 +86,16 @@ class UISeleniumSuite val statisticText = findAll(cssSelector("li strong")).map(_.text).toSeq statisticText should contain("Network receivers:") statisticText should contain("Batch interval:") + + // TODO add tests once SPARK-6796 is merged + + // Check a batch page without id + go to (sparkUI.appUIAddress.stripSuffix("/") + "/streaming/batch/") + webDriver.getPageSource should include ("Missing id parameter") + + // Check a non-exist batch + go to (sparkUI.appUIAddress.stripSuffix("/") + "/streaming/batch/?id=12345") + webDriver.getPageSource should include ("does not exist") } ssc.stop(false) From fc98a43b5903a8e3ee01ec6e34898e48f6303f67 Mon Sep 17 00:00:00 2001 From: zsxwing Date: Tue, 14 Apr 2015 11:40:34 +0800 Subject: [PATCH 02/13] Put clearing local properties to finally and remove redundant private[streaming] --- .../streaming/scheduler/JobScheduler.scala | 25 +++++++++++-------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala index fa4daa9ade385..491f922cbcdad 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala @@ -174,21 +174,24 @@ class JobScheduler(val ssc: StreamingContext) extends Logging { def run() { ssc.sc.setLocalProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, job.time.milliseconds.toString) ssc.sc.setLocalProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, job.outputOpId.toString) - eventActor ! JobStarted(job) - // Disable checks for existing output directories in jobs launched by the streaming scheduler, - // since we may need to write output to an existing directory during checkpoint recovery; - // see SPARK-4835 for more details. - PairRDDFunctions.disableOutputSpecValidation.withValue(true) { - job.run() + try { + eventActor ! JobStarted(job) + // Disable checks for existing output directories in jobs launched by the streaming + // scheduler, since we may need to write output to an existing directory during checkpoint + // recovery; see SPARK-4835 for more details. + PairRDDFunctions.disableOutputSpecValidation.withValue(true) { + job.run() + } + eventActor ! JobCompleted(job) + } finally { + ssc.sc.setLocalProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, null) + ssc.sc.setLocalProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, null) } - ssc.sc.setLocalProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, null) - ssc.sc.setLocalProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, null) - eventActor ! JobCompleted(job) } } } private[streaming] object JobScheduler { - private[streaming] val BATCH_TIME_PROPERTY_KEY = "spark.streaming.internal.batchTime" - private[streaming] val OUTPUT_OP_ID_PROPERTY_KEY = "spark.streaming.internal.outputOpId" + val BATCH_TIME_PROPERTY_KEY = "spark.streaming.internal.batchTime" + val OUTPUT_OP_ID_PROPERTY_KEY = "spark.streaming.internal.outputOpId" } From 0b226f9fd79d88ba3505553fdda8f6e774abf47b Mon Sep 17 00:00:00 2001 From: zsxwing Date: Tue, 14 Apr 2015 13:32:27 +0800 Subject: [PATCH 03/13] Change 'Last Error' to 'Error' --- .../main/scala/org/apache/spark/streaming/ui/BatchPage.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala index a539fdb9d6b16..4cf28708512eb 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala @@ -39,7 +39,7 @@ class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { Duration Stages: Succeeded/Total Tasks (for all stages): Succeeded/Total - Last Error + Error } private def makeOutputOpIdRow(outputOpId: OutputOpId, jobs: Seq[JobUIData]): Seq[Node] = { From 7168807f46d7ae7213b9f05071f5a1d25baa5b5e Mon Sep 17 00:00:00 2001 From: zsxwing Date: Tue, 14 Apr 2015 14:28:47 +0800 Subject: [PATCH 04/13] Limit the max width of the error message and fix nits in the UI --- .../scala/org/apache/spark/streaming/ui/BatchPage.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala index 4cf28708512eb..e876157fb4b00 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala @@ -135,7 +135,7 @@ class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { } else { "" } - {failureReasonSummary}{details} + {failureReasonSummary}{details} } private def jobsTable(jobInfos: Seq[(OutputOpId, JobId)]): Seq[Node] = { @@ -203,7 +203,7 @@ class BatchPage(parent: StreamingTab) extends WebUIPage("batch") {
  • Scheduling delay: - {formattedSchedulingDelay} records + {formattedSchedulingDelay}
  • Processing time: @@ -211,7 +211,7 @@ class BatchPage(parent: StreamingTab) extends WebUIPage("batch") {
  • Total delay: - {formattedTotalDelay} records + {formattedTotalDelay}
    • From 15bdf9b61d6cfda76a14131090949370a255a8b6 Mon Sep 17 00:00:00 2001 From: zsxwing Date: Wed, 15 Apr 2015 16:28:53 +0800 Subject: [PATCH 05/13] Add batch links and unit tests --- .../spark/streaming/ui/AllBatchesTable.scala | 6 +++- .../spark/streaming/UISeleniumSuite.scala | 31 +++++++++++++++++-- 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala index df1c0a10704c3..a6fe4a3b49c54 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala @@ -42,7 +42,11 @@ private[ui] abstract class BatchTableBase(tableId: String) { val processingTime = batch.processingDelay val formattedProcessingTime = processingTime.map(UIUtils.formatDuration).getOrElse("-") - {formattedBatchTime} + + + {formattedBatchTime} + + {eventCount.toString} events {formattedSchedulingDelay} diff --git a/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala index 7b7735213a858..1ca18af864ec2 100644 --- a/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala +++ b/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala @@ -88,8 +88,8 @@ class UISeleniumSuite statisticText should contain("Batch interval:") val h4Text = findAll(cssSelector("h4")).map(_.text).toSeq - h4Text should contain("Active Batches (0)") - h4Text should contain("Completed Batches (last 0 out of 0)") + h4Text.exists(_.matches("Active Batches \\(\\d+\\)")) should be (true) + h4Text.exists(_.matches("Completed Batches \\(last \\d+ out of \\d+\\)")) should be (true) findAll(cssSelector("""#active-batches-table th""")).map(_.text).toSeq should be { List("Batch Time", "Input Size", "Scheduling Delay", "Processing Time", "Status") @@ -98,7 +98,32 @@ class UISeleniumSuite List("Batch Time", "Input Size", "Scheduling Delay", "Processing Time", "Total Delay") } - // TODO add tests once SPARK-6796 is merged + val batchLinks = + findAll(cssSelector("""#completed-batches-table a""")).flatMap(_.attribute("href")).toSeq + batchLinks.size should be >= 1 + + // Check a normal batch page + go to (batchLinks.last) // Last should be the first batch, so it will have some jobs + val summaryText = findAll(cssSelector("li strong")).map(_.text).toSeq + summaryText should contain ("Batch Duration:") + summaryText should contain ("Input data size:") + summaryText should contain ("Scheduling delay:") + summaryText should contain ("Processing time:") + summaryText should contain ("Total delay:") + + findAll(cssSelector("""#batch-job-table th""")).map(_.text).toSeq should be { + List("Output Op Id", "Description", "Duration", "Job Id", "Duration", + "Stages: Succeeded/Total", "Tasks (for all stages): Succeeded/Total", "Error") + } + val jobLinks = + findAll(cssSelector("""#batch-job-table a""")).flatMap(_.attribute("href")).toSeq + jobLinks.size should be >= (1) + + // Check the job link in the batch page is right + go to (jobLinks(0)) + val jobDetails = findAll(cssSelector("li strong")).map(_.text).toSeq + jobDetails should contain("Status:") + jobDetails should contain("Completed Stages:") // Check a batch page without id go to (sparkUI.appUIAddress.stripSuffix("/") + "/streaming/batch/") From 77a69aeb1b096cdb0803e887870373007dd0bf0b Mon Sep 17 00:00:00 2001 From: zsxwing Date: Wed, 22 Apr 2015 22:19:50 +0800 Subject: [PATCH 06/13] Refactor codes as per TD's comments --- .../spark/streaming/scheduler/Job.scala | 8 +- .../apache/spark/streaming/ui/BatchPage.scala | 225 +++++++++++------- .../ui/StreamingJobProgressListener.scala | 26 +- 3 files changed, 158 insertions(+), 101 deletions(-) diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/Job.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/Job.scala index 064315177e06f..3c481bf3491f9 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/Job.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/Job.scala @@ -41,6 +41,9 @@ class Job(val time: Time, func: () => _) { _result } + /** + * @return the global unique id of this Job. + */ def id: String = { if (!isSet) { throw new IllegalStateException("Cannot access id before calling setId") @@ -48,6 +51,9 @@ class Job(val time: Time, func: () => _) { _id } + /** + * @return the output op id of this Job. Each Job has a unique output op id in the same JobSet. + */ def outputOpId: Int = { if (!isSet) { throw new IllegalStateException("Cannot access number before calling setId") @@ -60,7 +66,7 @@ class Job(val time: Time, func: () => _) { throw new IllegalStateException("Cannot call setOutputOpId more than once") } isSet = true - _id = "streaming job " + time + "." + outputOpId + _id = s"streaming job $time.$outputOpId" _outputOpId = outputOpId } diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala index e876157fb4b00..f88ac66b8d77c 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala @@ -19,16 +19,24 @@ package org.apache.spark.streaming.ui import javax.servlet.http.HttpServletRequest +import scala.collection.mutable.{ArrayBuffer, Map} +import scala.xml.{NodeSeq, Node} + import org.apache.commons.lang3.StringEscapeUtils + import org.apache.spark.streaming.Time +import org.apache.spark.streaming.scheduler.BatchInfo import org.apache.spark.ui.{UIUtils, WebUIPage} -import org.apache.spark.streaming.ui.StreamingJobProgressListener.{JobId, OutputOpId} +import org.apache.spark.streaming.ui.StreamingJobProgressListener.{SparkJobId, OutputOpId} import org.apache.spark.ui.jobs.UIData.JobUIData -import scala.xml.{NodeSeq, Node} +private[ui] case class BatchUIData( + var batchInfo: BatchInfo = null, + outputOpIdToSparkJobIds: Map[OutputOpId, ArrayBuffer[SparkJobId]] = Map()) { +} -class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { - private val streaminglistener = parent.listener +private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { + private val streamingListener = parent.listener private val sparkListener = parent.ssc.sc.jobProgressListener private def columns: Seq[Node] = { @@ -42,75 +50,101 @@ class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { Error } - private def makeOutputOpIdRow(outputOpId: OutputOpId, jobs: Seq[JobUIData]): Seq[Node] = { - val jobDurations = jobs.map(job => { - job.submissionTime.map { start => - val end = job.completionTime.getOrElse(System.currentTimeMillis()) - end - start - } - }) - val formattedOutputOpDuration = - if (jobDurations.exists(_ == None)) { - // If any job does not finish, set "formattedOutputOpDuration" to "-" - "-" - } else { - UIUtils.formatDuration(jobDurations.flatMap(x => x).sum) - } + /** + * Generate a row for a Spark Job. Because duplicated output op infos needs to be collapsed into + * one cell, we use "rowspan" for the first row of a output op. + */ + def generateJobRow( + outputOpId: OutputOpId, + formattedOutputOpDuration: String, + numSparkJobRowsInOutputOp: Int, + isFirstRow: Boolean, + sparkJob: JobUIData): Seq[Node] = { + val lastStageInfo = Option(sparkJob.stageIds) + .filter(_.nonEmpty) + .flatMap { ids => sparkListener.stageIdToInfo.get(ids.max) } + val lastStageData = lastStageInfo.flatMap { s => + sparkListener.stageIdToData.get((s.stageId, s.attemptId)) + } - def makeJobRow(job: JobUIData, isFirstRow: Boolean): Seq[Node] = { - val lastStageInfo = Option(job.stageIds) - .filter(_.nonEmpty) - .flatMap { ids => sparkListener.stageIdToInfo.get(ids.max) } - val lastStageData = lastStageInfo.flatMap { s => - sparkListener.stageIdToData.get((s.stageId, s.attemptId)) + val lastStageName = lastStageInfo.map(_.name).getOrElse("(Unknown Stage Name)") + val lastStageDescription = lastStageData.flatMap(_.description).getOrElse("") + val duration: Option[Long] = { + sparkJob.submissionTime.map { start => + val end = sparkJob.completionTime.getOrElse(System.currentTimeMillis()) + end - start } + } + val lastFailureReason = + sparkJob.stageIds.sorted.reverse.flatMap(sparkListener.stageIdToInfo.get). + dropWhile(_.failureReason == None).take(1). // get the first info that contains failure + flatMap(info => info.failureReason).headOption.getOrElse("") + val formattedDuration = duration.map(d => UIUtils.formatDuration(d)).getOrElse("-") + val detailUrl = s"${UIUtils.prependBaseUri(parent.basePath)}/jobs/job?id=${sparkJob.jobId}" - val lastStageName = lastStageInfo.map(_.name).getOrElse("(Unknown Stage Name)") - val lastStageDescription = lastStageData.flatMap(_.description).getOrElse("") - val duration: Option[Long] = { - job.submissionTime.map { start => - val end = job.completionTime.getOrElse(System.currentTimeMillis()) - end - start - } - } - val lastFailureReason = job.stageIds.sorted.reverse.flatMap(sparkListener.stageIdToInfo.get). - dropWhile(_.failureReason == None).take(1). // get the first info that contains failure - flatMap(info => info.failureReason).headOption.getOrElse("") - val formattedDuration = duration.map(d => UIUtils.formatDuration(d)).getOrElse("-") - val detailUrl = s"${UIUtils.prependBaseUri(parent.basePath)}/jobs/job?id=${job.jobId}" - - {if(isFirstRow) { - {outputOpId} - + // In the first row, output op id and its information needs to be shown. In other rows, these + // cells will be taken up due to "rowspan". + val prefixCells = + if (isFirstRow) { + {outputOpId.toString} + {lastStageDescription} {lastStageName} - {formattedOutputOpDuration}} + {formattedOutputOpDuration} + } else { + Nil + } + + + {prefixCells} + + + {sparkJob.jobId}{sparkJob.jobGroup.map(id => s"($id)").getOrElse("")} + + + + {formattedDuration} + + + {sparkJob.completedStageIndices.size}/{sparkJob.stageIds.size - sparkJob.numSkippedStages} + {if (sparkJob.numFailedStages > 0) s"(${sparkJob.numFailedStages} failed)"} + {if (sparkJob.numSkippedStages > 0) s"(${sparkJob.numSkippedStages} skipped)"} + + + { + UIUtils.makeProgressBar( + started = sparkJob.numActiveTasks, + completed = sparkJob.numCompletedTasks, + failed = sparkJob.numFailedTasks, + skipped = sparkJob.numSkippedTasks, + total = sparkJob.numTasks - sparkJob.numSkippedTasks) } - - - {job.jobId}{job.jobGroup.map(id => s"($id)").getOrElse("")} - - - - {formattedDuration} - - - {job.completedStageIndices.size}/{job.stageIds.size - job.numSkippedStages} - {if (job.numFailedStages > 0) s"(${job.numFailedStages} failed)"} - {if (job.numSkippedStages > 0) s"(${job.numSkippedStages} skipped)"} - - - {UIUtils.makeProgressBar(started = job.numActiveTasks, completed = job.numCompletedTasks, - failed = job.numFailedTasks, skipped = job.numSkippedTasks, - total = job.numTasks - job.numSkippedTasks)} - - {failureReasonCell(lastFailureReason)} - - } + + {failureReasonCell(lastFailureReason)} + + } - makeJobRow(jobs.head, true) ++ (jobs.tail.map(job => makeJobRow(job, false)).flatMap(x => x)) + private def generateOutputOpIdRow( + outputOpId: OutputOpId, sparkJobs: Seq[JobUIData]): Seq[Node] = { + val sparkjobDurations = sparkJobs.map(sparkJob => { + sparkJob.submissionTime.map { start => + val end = sparkJob.completionTime.getOrElse(System.currentTimeMillis()) + end - start + } + }) + val formattedOutputOpDuration = + if (sparkjobDurations.exists(_ == None)) { + // If any job does not finish, set "formattedOutputOpDuration" to "-" + "-" + } else { + UIUtils.formatDuration(sparkjobDurations.flatMap(x => x).sum) + } + generateJobRow(outputOpId, formattedOutputOpDuration, sparkJobs.size, true, sparkJobs.head) ++ + sparkJobs.tail.map { sparkJob => + generateJobRow(outputOpId, formattedOutputOpDuration, sparkJobs.size, false, sparkJob) + }.flatMap(x => x) } private def failureReasonCell(failureReason: String): Seq[Node] = { @@ -138,26 +172,29 @@ class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { {failureReasonSummary}{details} } - private def jobsTable(jobInfos: Seq[(OutputOpId, JobId)]): Seq[Node] = { - def getJobData(jobId: JobId): Option[JobUIData] = { - sparkListener.activeJobs.get(jobId).orElse { - sparkListener.completedJobs.find(_.jobId == jobId).orElse { - sparkListener.failedJobs.find(_.jobId == jobId) - } + private def getJobData(sparkJobId: SparkJobId): Option[JobUIData] = { + sparkListener.activeJobs.get(sparkJobId).orElse { + sparkListener.completedJobs.find(_.jobId == sparkJobId).orElse { + sparkListener.failedJobs.find(_.jobId == sparkJobId) } } + } - // Group jobInfos by OutputOpId firstly, then sort them. - // E.g., [(0, 1), (1, 3), (0, 2), (1, 4)] => [(0, [1, 2]), (1, [3, 4])] - val outputOpIdWithJobIds: Seq[(OutputOpId, Seq[JobId])] = - jobInfos.groupBy(_._1).toSeq.sortBy(_._1). // sorted by OutputOpId + /** + * Generate the job table for the batch. + */ + private def generateJobTable(batchUIData: BatchUIData): Seq[Node] = { + val outputOpIdWithSparkJobIds: Seq[(OutputOpId, Seq[SparkJobId])] = { + batchUIData.outputOpIdToSparkJobIds.toSeq.sortBy(_._1). // sorted by OutputOpId map { case (outputOpId, jobs) => - (outputOpId, jobs.map(_._2).sortBy(x => x).toSeq)} // sort JobIds for each OutputOpId + (outputOpId, jobs.sorted.toSeq) // sort JobIds for each OutputOpId + } + } sparkListener.synchronized { - val outputOpIdWithJobs: Seq[(OutputOpId, Seq[JobUIData])] = outputOpIdWithJobIds.map { - case (outputOpId, jobIds) => - // Filter out JobIds that don't exist in sparkListener - (outputOpId, jobIds.flatMap(getJobData)) + val outputOpIdWithJobs: Seq[(OutputOpId, Seq[JobUIData])] = outputOpIdWithSparkJobIds.map { + case (outputOpId, sparkJobIds) => + // Filter out spark Job ids that don't exist in sparkListener + (outputOpId, sparkJobIds.flatMap(getJobData)) } @@ -165,7 +202,11 @@ class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { {columns} - {outputOpIdWithJobs.map { case (outputOpId, jobs) => makeOutputOpIdRow(outputOpId, jobs)}} + { + outputOpIdWithJobs.map { + case (outputOpId, jobs) => generateOutputOpIdRow(outputOpId, jobs) + } + }
    } @@ -176,13 +217,11 @@ class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { throw new IllegalArgumentException(s"Missing id parameter") } val formattedBatchTime = UIUtils.formatDate(batchTime.milliseconds) - val (batchInfo, jobInfos) = streaminglistener.synchronized { - val _batchInfo = streaminglistener.getBatchInfo(batchTime).getOrElse { - throw new IllegalArgumentException(s"Batch $formattedBatchTime does not exist") - } - val _jobInfos = streaminglistener.getJobInfos(batchTime) - (_batchInfo, _jobInfos) + + val batchUIData = streamingListener.getBatchUIData(batchTime).getOrElse { + throw new IllegalArgumentException(s"Batch $formattedBatchTime does not exist") } + val batchInfo = batchUIData.batchInfo val formattedSchedulingDelay = batchInfo.schedulingDelay.map(UIUtils.formatDuration).getOrElse("-") @@ -195,7 +234,7 @@ class BatchPage(parent: StreamingTab) extends WebUIPage("batch") {
    • Batch Duration: - {UIUtils.formatDuration(streaminglistener.batchDuration)} + {UIUtils.formatDuration(streamingListener.batchDuration)}
    • Input data size: @@ -216,9 +255,15 @@ class BatchPage(parent: StreamingTab) extends WebUIPage("batch") {
    - val content = summary ++ jobInfos.map(jobsTable).getOrElse { -
    Cannot find any job for Batch {formattedBatchTime}
    - } + val jobTable = + if (batchUIData.outputOpIdToSparkJobIds.isEmpty) { +
    Cannot find any job for Batch {formattedBatchTime}.
    + } else { + generateJobTable(batchUIData) + } + + val content = summary ++ jobTable + UIUtils.headerSparkPage(s"Details of batch at $formattedBatchTime", content, parent) } } diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala index a7dd41f0443c4..d9f9602bb13c9 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala @@ -34,8 +34,6 @@ import org.apache.spark.util.Distribution private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) extends StreamingListener with SparkListener { - import StreamingJobProgressListener._ - private val waitingBatchInfos = new HashMap[Time, BatchInfo] private val runningBatchInfos = new HashMap[Time, BatchInfo] private val completedBatchInfos = new Queue[BatchInfo] @@ -45,7 +43,7 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) private var totalProcessedRecords = 0L private val receiverInfos = new HashMap[Int, ReceiverInfo] - private val batchTimeToJobIds = new HashMap[Time, ArrayBuffer[(OutputOpId, JobId)]] + private val batchTimeToBatchUIData = new HashMap[Time, BatchUIData] val batchDuration = ssc.graph.batchDuration.milliseconds @@ -87,7 +85,7 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) completedBatchInfos.enqueue(batchCompleted.batchInfo) if (completedBatchInfos.size > batchInfoLimit) { val removedBatch = completedBatchInfos.dequeue() - batchTimeToJobIds.remove(removedBatch.batchTime) + batchTimeToBatchUIData.remove(removedBatch.batchTime) } totalCompletedBatches += 1L @@ -97,8 +95,12 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) override def onJobStart(jobStart: SparkListenerJobStart): Unit = synchronized { getBatchTimeAndOutputOpId(jobStart.properties).foreach { case (batchTime, outputOpId) => - batchTimeToJobIds.getOrElseUpdate(batchTime, ArrayBuffer[(OutputOpId, JobId)]()) += - outputOpId -> jobStart.jobId + val batchUIData = batchTimeToBatchUIData.getOrElseUpdate(batchTime, BatchUIData()) + // Because onJobStart and onBatchXXX messages are processed in different threads, + // we may not be able to get the corresponding BatchInfo now. So here we only set + // batchUIData.outputOpIdToSparkJobIds, batchUIData.batchInfo will be set in "getBatchUIData". + batchUIData.outputOpIdToSparkJobIds. + getOrElseUpdate(outputOpId, ArrayBuffer()) += jobStart.jobId } } @@ -206,7 +208,7 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) Distribution(completedBatchInfos.flatMap(getMetric(_)).map(_.toDouble)) } - def getBatchInfo(batchTime: Time): Option[BatchInfo] = synchronized { + private def getBatchInfo(batchTime: Time): Option[BatchInfo] = { waitingBatchInfos.get(batchTime).orElse { runningBatchInfos.get(batchTime).orElse { completedBatchInfos.find(batch => batch.batchTime == batchTime) @@ -214,12 +216,16 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) } } - def getJobInfos(batchTime: Time): Option[Seq[(OutputOpId, JobId)]] = synchronized { - batchTimeToJobIds.get(batchTime).map(_.toList) + def getBatchUIData(batchTime: Time): Option[BatchUIData] = synchronized { + for (batchInfo <- getBatchInfo(batchTime)) yield { + val batchUIData = batchTimeToBatchUIData.getOrElse(batchTime, BatchUIData(batchInfo)) + batchUIData.batchInfo = batchInfo + batchUIData + } } } private[streaming] object StreamingJobProgressListener { - type JobId = Int + type SparkJobId = Int type OutputOpId = Int } From 1282b10a4a44b8f67e15689dc31c4d6798a712af Mon Sep 17 00:00:00 2001 From: zsxwing Date: Mon, 27 Apr 2015 00:14:16 -0700 Subject: [PATCH 07/13] Handle some corner cases and add tests for StreamingJobProgressListener --- .../apache/spark/streaming/ui/BatchPage.scala | 16 +-- .../spark/streaming/ui/BatchUIData.scala | 31 +++++ .../ui/StreamingJobProgressListener.scala | 57 ++++++--- .../StreamingJobProgressListenerSuite.scala | 108 ++++++++++++++++++ 4 files changed, 185 insertions(+), 27 deletions(-) create mode 100644 streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala index f88ac66b8d77c..e7c38ce9f4f01 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala @@ -30,10 +30,6 @@ import org.apache.spark.ui.{UIUtils, WebUIPage} import org.apache.spark.streaming.ui.StreamingJobProgressListener.{SparkJobId, OutputOpId} import org.apache.spark.ui.jobs.UIData.JobUIData -private[ui] case class BatchUIData( - var batchInfo: BatchInfo = null, - outputOpIdToSparkJobIds: Map[OutputOpId, ArrayBuffer[SparkJobId]] = Map()) { -} private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { private val streamingListener = parent.listener @@ -184,18 +180,12 @@ private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { * Generate the job table for the batch. */ private def generateJobTable(batchUIData: BatchUIData): Seq[Node] = { - val outputOpIdWithSparkJobIds: Seq[(OutputOpId, Seq[SparkJobId])] = { - batchUIData.outputOpIdToSparkJobIds.toSeq.sortBy(_._1). // sorted by OutputOpId - map { case (outputOpId, jobs) => - (outputOpId, jobs.sorted.toSeq) // sort JobIds for each OutputOpId - } - } sparkListener.synchronized { - val outputOpIdWithJobs: Seq[(OutputOpId, Seq[JobUIData])] = outputOpIdWithSparkJobIds.map { - case (outputOpId, sparkJobIds) => + val outputOpIdWithJobs: Seq[(OutputOpId, Seq[JobUIData])] = + batchUIData.outputOpIdToSparkJobIds.map { case (outputOpId, sparkJobIds) => // Filter out spark Job ids that don't exist in sparkListener (outputOpId, sparkJobIds.flatMap(getJobData)) - } + } diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala new file mode 100644 index 0000000000000..be0a49a13f859 --- /dev/null +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + +package org.apache.spark.streaming.ui + +import org.apache.spark.streaming.scheduler.BatchInfo +import org.apache.spark.streaming.ui.StreamingJobProgressListener._ + +/** + * The data in outputOpIdToSparkJobIds are sorted by `OutputOpId` in ascending order, and for each + * `OutputOpId`, the corresponding SparkJobId`s are sorted in ascending order. + */ +private[ui] case class BatchUIData( + batchInfo: BatchInfo, + outputOpIdToSparkJobIds: Seq[(OutputOpId, Seq[SparkJobId])]) { +} diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala index d9f9602bb13c9..c9d1a87825696 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala @@ -17,6 +17,8 @@ package org.apache.spark.streaming.ui +import java.util.LinkedHashMap +import java.util.{Map => JMap} import java.util.Properties import scala.collection.mutable.{ArrayBuffer, Queue, HashMap} @@ -34,6 +36,8 @@ import org.apache.spark.util.Distribution private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) extends StreamingListener with SparkListener { + import StreamingJobProgressListener._ + private val waitingBatchInfos = new HashMap[Time, BatchInfo] private val runningBatchInfos = new HashMap[Time, BatchInfo] private val completedBatchInfos = new Queue[BatchInfo] @@ -43,7 +47,28 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) private var totalProcessedRecords = 0L private val receiverInfos = new HashMap[Int, ReceiverInfo] - private val batchTimeToBatchUIData = new HashMap[Time, BatchUIData] + // Because onJobStart and onBatchXXX messages are processed in different threads, + // we may not be able to get the corresponding BatchInfo when receiving onJobStart. So here we + // cannot use a map of (Time, BatchUIData). + private[ui] val batchTimeToOutputOpIdToSparkJobIds = + new LinkedHashMap[Time, OutputOpIdToSparkJobIds] { + override def removeEldestEntry(p1: JMap.Entry[Time, OutputOpIdToSparkJobIds]): Boolean = { + // If a lot of "onBatchCompleted"s happen before "onJobStart" (image if + // SparkContext.listenerBus is very slow), "batchTimeToOutputOpIdToSparkJobIds" + // may add some information for a removed batch when processing "onJobStart". It will be a + // memory leak. + // + // To avoid the memory leak, we control the size of "batchTimeToOutputOpIdToSparkJobIds" and + // evict the eldest one. + // + // Note: if "onJobStart" happens before "onBatchSubmitted", the size of + // "batchTimeToOutputOpIdToSparkJobIds" may be greater than the number of the retained + // batches temporarily, so here we use "10" to handle such case. This is not a perfect + // solution, but at least it can handle most of cases. + size() > waitingBatchInfos.size + runningBatchInfos.size + completedBatchInfos.size + 10 + } + } + val batchDuration = ssc.graph.batchDuration.milliseconds @@ -85,7 +110,7 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) completedBatchInfos.enqueue(batchCompleted.batchInfo) if (completedBatchInfos.size > batchInfoLimit) { val removedBatch = completedBatchInfos.dequeue() - batchTimeToBatchUIData.remove(removedBatch.batchTime) + batchTimeToOutputOpIdToSparkJobIds.remove(removedBatch.batchTime) } totalCompletedBatches += 1L @@ -95,12 +120,12 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) override def onJobStart(jobStart: SparkListenerJobStart): Unit = synchronized { getBatchTimeAndOutputOpId(jobStart.properties).foreach { case (batchTime, outputOpId) => - val batchUIData = batchTimeToBatchUIData.getOrElseUpdate(batchTime, BatchUIData()) - // Because onJobStart and onBatchXXX messages are processed in different threads, - // we may not be able to get the corresponding BatchInfo now. So here we only set - // batchUIData.outputOpIdToSparkJobIds, batchUIData.batchInfo will be set in "getBatchUIData". - batchUIData.outputOpIdToSparkJobIds. - getOrElseUpdate(outputOpId, ArrayBuffer()) += jobStart.jobId + var outputOpIdToSparkJobIds = batchTimeToOutputOpIdToSparkJobIds.get(batchTime) + if (outputOpIdToSparkJobIds == null) { + outputOpIdToSparkJobIds = new OutputOpIdToSparkJobIds() + batchTimeToOutputOpIdToSparkJobIds.put(batchTime, outputOpIdToSparkJobIds) + } + outputOpIdToSparkJobIds.getOrElseUpdate(outputOpId, ArrayBuffer()) += jobStart.jobId } } @@ -116,9 +141,7 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) } } - def numReceivers: Int = synchronized { - ssc.graph.getReceiverInputStreams().size - } + def numReceivers: Int = ssc.graph.getReceiverInputStreams().size def numTotalCompletedBatches: Long = synchronized { totalCompletedBatches @@ -218,9 +241,14 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) def getBatchUIData(batchTime: Time): Option[BatchUIData] = synchronized { for (batchInfo <- getBatchInfo(batchTime)) yield { - val batchUIData = batchTimeToBatchUIData.getOrElse(batchTime, BatchUIData(batchInfo)) - batchUIData.batchInfo = batchInfo - batchUIData + // outputOpIdToSparkJobIds is a sorted copy of the original one so that the caller can feel + // free to use the data in BatchUIData. + val outputOpIdToSparkJobIds = Option(batchTimeToOutputOpIdToSparkJobIds.get(batchTime)). + getOrElse(Map.empty).toSeq.sortWith(_._1 < _._1). // sorted by OutputOpId + map { case (outputOpId, jobs) => + (outputOpId, jobs.sorted.toSeq) // sort JobIds for each OutputOpId + } + BatchUIData(batchInfo, outputOpIdToSparkJobIds) } } } @@ -228,4 +256,5 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) private[streaming] object StreamingJobProgressListener { type SparkJobId = Int type OutputOpId = Int + private type OutputOpIdToSparkJobIds = HashMap[OutputOpId, ArrayBuffer[SparkJobId]] } diff --git a/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala index 94b1985116feb..8cf7d7f861aca 100644 --- a/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala +++ b/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala @@ -17,8 +17,11 @@ package org.apache.spark.streaming.ui +import java.util.Properties + import org.scalatest.Matchers +import org.apache.spark.scheduler.SparkListenerJobStart import org.apache.spark.streaming.dstream.DStream import org.apache.spark.streaming.scheduler._ import org.apache.spark.streaming.{Duration, Time, Milliseconds, TestSuiteBase} @@ -64,6 +67,48 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { listener.numTotalProcessedRecords should be (0) listener.numTotalReceivedRecords should be (600) + // onJobStart + val properties1 = new Properties() + properties1.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, Time(1000).milliseconds.toString) + properties1.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, 0.toString) + val jobStart1 = SparkListenerJobStart(jobId = 0, + 0L, // unused + Nil, // unused + properties1) + listener.onJobStart(jobStart1) + + val properties2 = new Properties() + properties2.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, Time(1000).milliseconds.toString) + properties2.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, 0.toString) + val jobStart2 = SparkListenerJobStart(jobId = 1, + 0L, // unused + Nil, // unused + properties2) + listener.onJobStart(jobStart2) + + val properties3 = new Properties() + properties3.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, Time(1000).milliseconds.toString) + properties3.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, 1.toString) + val jobStart3 = SparkListenerJobStart(jobId = 0, + 0L, // unused + Nil, // unused + properties3) + listener.onJobStart(jobStart3) + + val properties4 = new Properties() + properties4.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, Time(1000).milliseconds.toString) + properties4.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, 1.toString) + val jobStart4 = SparkListenerJobStart(jobId = 1, + 0L, // unused + Nil, // unused + properties4) + listener.onJobStart(jobStart4) + + val batchUIData = listener.getBatchUIData(Time(1000)) + assert(batchUIData != None) + assert(batchUIData.get.batchInfo === batchInfoStarted) + assert(batchUIData.get.outputOpIdToSparkJobIds === Seq(0 -> Seq(0, 1), 1 -> Seq(0, 1))) + // onBatchCompleted val batchInfoCompleted = BatchInfo(Time(1000), receivedBlockInfo, 1000, Some(2000), None) listener.onBatchCompleted(StreamingListenerBatchCompleted(batchInfoCompleted)) @@ -116,4 +161,67 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { listener.retainedCompletedBatches.size should be (limit) listener.numTotalCompletedBatches should be(limit + 10) } + + test("disorder onJobStart and onBatchXXX") { + val ssc = setupStreams(input, operation) + val limit = ssc.conf.getInt("spark.streaming.ui.retainedBatches", 100) + val listener = new StreamingJobProgressListener(ssc) + + // fulfill completedBatchInfos + for(i <- 0 until limit) { + val batchInfoCompleted = + BatchInfo(Time(1000 + i * 100), Map.empty, 1000 + i * 100, Some(2000 + i * 100), None) + val properties = new Properties() + properties.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, (1000 + i * 100).toString) + properties.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, "0") + val jobStart = SparkListenerJobStart(jobId = 1, + 0L, // unused + Nil, // unused + properties) + listener.onBatchCompleted(StreamingListenerBatchCompleted(batchInfoCompleted)) + listener.onJobStart(jobStart) + } + + // onJobStart happens before onBatchSubmitted + val properties = new Properties() + properties.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, (1000 + limit * 100).toString) + properties.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, "0") + val jobStart = SparkListenerJobStart(jobId = 0, + 0L, // unused + Nil, // unused + properties) + listener.onJobStart(jobStart) + + val batchInfoSubmitted = + BatchInfo(Time(1000 + limit * 100), Map.empty, (1000 + limit * 100), None, None) + listener.onBatchSubmitted(StreamingListenerBatchSubmitted(batchInfoSubmitted)) + + // We still can see the info retrieved from onJobStart + listener.getBatchUIData(Time(1000 + limit * 100)) should be + Some(BatchUIData(batchInfoSubmitted, Seq((0, Seq(0))))) + + + // A lot of "onBatchCompleted"s happen before "onJobStart" + for(i <- limit + 1 to limit * 2) { + val batchInfoCompleted = + BatchInfo(Time(1000 + i * 100), Map.empty, 1000 + i * 100, Some(2000 + i * 100), None) + listener.onBatchCompleted(StreamingListenerBatchCompleted(batchInfoCompleted)) + } + + for(i <- limit + 1 to limit * 2) { + val properties = new Properties() + properties.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, (1000 + i * 100).toString) + properties.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, "0") + val jobStart = SparkListenerJobStart(jobId = 1, + 0L, // unused + Nil, // unused + properties) + listener.onJobStart(jobStart) + } + + // We should not leak memory + listener.batchTimeToOutputOpIdToSparkJobIds.size() should be <= + (listener.waitingBatches.size + listener.runningBatches.size + + listener.retainedCompletedBatches.size + 10) + } } From 72f8e7e61c5cedc362ee8c3ce292216bbdc021ca Mon Sep 17 00:00:00 2001 From: zsxwing Date: Mon, 27 Apr 2015 01:15:41 -0700 Subject: [PATCH 08/13] Add unit tests for BatchPage --- .../apache/spark/streaming/ui/BatchPage.scala | 4 +- .../spark/streaming/UISeleniumSuite.scala | 41 +++++++++++++++++-- 2 files changed, 40 insertions(+), 5 deletions(-) diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala index e7c38ce9f4f01..1589a91fd7574 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala @@ -80,9 +80,10 @@ private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { // In the first row, output op id and its information needs to be shown. In other rows, these // cells will be taken up due to "rowspan". + // scalastyle:off val prefixCells = if (isFirstRow) { - + {prefixCells} diff --git a/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala index 1ca18af864ec2..405a4e90ff093 100644 --- a/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala +++ b/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala @@ -65,7 +65,18 @@ class UISeleniumSuite private def setupStreams(ssc: StreamingContext): Unit = { val rdds = Queue(ssc.sc.parallelize(1 to 4, 4)) val inputStream = ssc.queueStream(rdds) - inputStream.foreachRDD(rdd => rdd.foreach(_ => {})) + inputStream.foreachRDD { rdd => + rdd.foreach(_ => {}) + rdd.foreach(_ => {}) + } + inputStream.foreachRDD { rdd => + rdd.foreach(_ => {}) + try { + rdd.foreach(_ => throw new RuntimeException("Oops")) + } catch { + case e: SparkException if e.getMessage.contains("Oops") => + } + } } test("attaching and detaching a Streaming tab") { @@ -115,9 +126,31 @@ class UISeleniumSuite List("Output Op Id", "Description", "Duration", "Job Id", "Duration", "Stages: Succeeded/Total", "Tasks (for all stages): Succeeded/Total", "Error") } - val jobLinks = - findAll(cssSelector("""#batch-job-table a""")).flatMap(_.attribute("href")).toSeq - jobLinks.size should be >= (1) + + // Check we have 2 output op ids + val outputOpIds = findAll(cssSelector(".output-op-id-cell")).toSeq + outputOpIds.map(_.attribute("rowspan")) should be (List(Some("2"), Some("2"))) + outputOpIds.map(_.text) should be (List("0", "1")) + + // Check job ids + val jobIdCells = findAll(cssSelector( """#batch-job-table a""")).toSeq + jobIdCells.map(_.text) should be (List("0", "1", "2", "3")) + + val jobLinks = jobIdCells.flatMap(_.attribute("href")) + jobLinks.size should be (4) + + // Check stage progress + findAll(cssSelector(""".stage-progress-cell""")).map(_.text).toSeq should be + (List("1/1", "1/1", "1/1", "0/1 (1 failed)")) + + // Check job progress + findAll(cssSelector(""".progress-cell""")).map(_.text).toSeq should be + (List("1/1", "1/1", "1/1", "0/1 (1 failed)")) + + // Check stacktrack + val errorCells = findAll(cssSelector(""".stacktrace-details""")).map(_.text).toSeq + errorCells should have size 1 + errorCells(0) should include("java.lang.RuntimeException: Oops") // Check the job link in the batch page is right go to (jobLinks(0)) From cb62e4fe27763a23f7c925fc7086d3f606dc7034 Mon Sep 17 00:00:00 2001 From: zsxwing Date: Mon, 27 Apr 2015 13:46:56 -0700 Subject: [PATCH 09/13] Use Seq[(OutputOpId, SparkJobId)] to store the id relations --- .../apache/spark/streaming/ui/BatchPage.scala | 10 +++++-- .../spark/streaming/ui/BatchUIData.scala | 6 +---- .../ui/StreamingJobProgressListener.scala | 27 ++++++++----------- .../StreamingJobProgressListenerSuite.scala | 6 ++--- 4 files changed, 23 insertions(+), 26 deletions(-) diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala index 1589a91fd7574..77beebe2a2a56 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala @@ -182,9 +182,15 @@ private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { * Generate the job table for the batch. */ private def generateJobTable(batchUIData: BatchUIData): Seq[Node] = { + val outputOpIdToSparkJobIds = batchUIData.outputOpIdSparkJobIdPairs.groupBy(_._1).toSeq. + sortBy(_._1). // sorted by OutputOpId + map { case (outputOpId, outputOpIdAndSparkJobIdPairs) => + // sort SparkJobIds for each OutputOpId + (outputOpId, outputOpIdAndSparkJobIdPairs.map(_._2).sorted) + } sparkListener.synchronized { val outputOpIdWithJobs: Seq[(OutputOpId, Seq[JobUIData])] = - batchUIData.outputOpIdToSparkJobIds.map { case (outputOpId, sparkJobIds) => + outputOpIdToSparkJobIds.map { case (outputOpId, sparkJobIds) => // Filter out spark Job ids that don't exist in sparkListener (outputOpId, sparkJobIds.flatMap(getJobData)) } @@ -248,7 +254,7 @@ private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { val jobTable = - if (batchUIData.outputOpIdToSparkJobIds.isEmpty) { + if (batchUIData.outputOpIdSparkJobIdPairs.isEmpty) {
    Cannot find any job for Batch {formattedBatchTime}.
    } else { generateJobTable(batchUIData) diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala index be0a49a13f859..a7da139d4e4b3 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala @@ -21,11 +21,7 @@ package org.apache.spark.streaming.ui import org.apache.spark.streaming.scheduler.BatchInfo import org.apache.spark.streaming.ui.StreamingJobProgressListener._ -/** - * The data in outputOpIdToSparkJobIds are sorted by `OutputOpId` in ascending order, and for each - * `OutputOpId`, the corresponding SparkJobId`s are sorted in ascending order. - */ private[ui] case class BatchUIData( batchInfo: BatchInfo, - outputOpIdToSparkJobIds: Seq[(OutputOpId, Seq[SparkJobId])]) { + outputOpIdSparkJobIdPairs: Seq[(OutputOpId, SparkJobId)]) { } diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala index c9d1a87825696..a3f190c4889ae 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala @@ -50,9 +50,10 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) // Because onJobStart and onBatchXXX messages are processed in different threads, // we may not be able to get the corresponding BatchInfo when receiving onJobStart. So here we // cannot use a map of (Time, BatchUIData). - private[ui] val batchTimeToOutputOpIdToSparkJobIds = - new LinkedHashMap[Time, OutputOpIdToSparkJobIds] { - override def removeEldestEntry(p1: JMap.Entry[Time, OutputOpIdToSparkJobIds]): Boolean = { + private[ui] val batchTimeToOutputOpIdSparkJobIdPair = + new LinkedHashMap[Time, ArrayBuffer[(OutputOpId, SparkJobId)]] { + override def removeEldestEntry(p1: JMap.Entry[Time, ArrayBuffer[(OutputOpId, SparkJobId)]]): + Boolean = { // If a lot of "onBatchCompleted"s happen before "onJobStart" (image if // SparkContext.listenerBus is very slow), "batchTimeToOutputOpIdToSparkJobIds" // may add some information for a removed batch when processing "onJobStart". It will be a @@ -110,7 +111,7 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) completedBatchInfos.enqueue(batchCompleted.batchInfo) if (completedBatchInfos.size > batchInfoLimit) { val removedBatch = completedBatchInfos.dequeue() - batchTimeToOutputOpIdToSparkJobIds.remove(removedBatch.batchTime) + batchTimeToOutputOpIdSparkJobIdPair.remove(removedBatch.batchTime) } totalCompletedBatches += 1L @@ -120,12 +121,12 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) override def onJobStart(jobStart: SparkListenerJobStart): Unit = synchronized { getBatchTimeAndOutputOpId(jobStart.properties).foreach { case (batchTime, outputOpId) => - var outputOpIdToSparkJobIds = batchTimeToOutputOpIdToSparkJobIds.get(batchTime) + var outputOpIdToSparkJobIds = batchTimeToOutputOpIdSparkJobIdPair.get(batchTime) if (outputOpIdToSparkJobIds == null) { - outputOpIdToSparkJobIds = new OutputOpIdToSparkJobIds() - batchTimeToOutputOpIdToSparkJobIds.put(batchTime, outputOpIdToSparkJobIds) + outputOpIdToSparkJobIds = ArrayBuffer[(OutputOpId, SparkJobId)]() + batchTimeToOutputOpIdSparkJobIdPair.put(batchTime, outputOpIdToSparkJobIds) } - outputOpIdToSparkJobIds.getOrElseUpdate(outputOpId, ArrayBuffer()) += jobStart.jobId + outputOpIdToSparkJobIds += (outputOpId -> jobStart.jobId) } } @@ -241,13 +242,8 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) def getBatchUIData(batchTime: Time): Option[BatchUIData] = synchronized { for (batchInfo <- getBatchInfo(batchTime)) yield { - // outputOpIdToSparkJobIds is a sorted copy of the original one so that the caller can feel - // free to use the data in BatchUIData. - val outputOpIdToSparkJobIds = Option(batchTimeToOutputOpIdToSparkJobIds.get(batchTime)). - getOrElse(Map.empty).toSeq.sortWith(_._1 < _._1). // sorted by OutputOpId - map { case (outputOpId, jobs) => - (outputOpId, jobs.sorted.toSeq) // sort JobIds for each OutputOpId - } + val outputOpIdToSparkJobIds = Option(batchTimeToOutputOpIdSparkJobIdPair.get(batchTime)). + map(ArrayBuffer(_: _*)).getOrElse(Seq.empty) BatchUIData(batchInfo, outputOpIdToSparkJobIds) } } @@ -256,5 +252,4 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) private[streaming] object StreamingJobProgressListener { type SparkJobId = Int type OutputOpId = Int - private type OutputOpIdToSparkJobIds = HashMap[OutputOpId, ArrayBuffer[SparkJobId]] } diff --git a/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala index 8cf7d7f861aca..31b599a5ba28a 100644 --- a/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala +++ b/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala @@ -107,7 +107,7 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { val batchUIData = listener.getBatchUIData(Time(1000)) assert(batchUIData != None) assert(batchUIData.get.batchInfo === batchInfoStarted) - assert(batchUIData.get.outputOpIdToSparkJobIds === Seq(0 -> Seq(0, 1), 1 -> Seq(0, 1))) + assert(batchUIData.get.outputOpIdSparkJobIdPairs === Seq((0, 0), (0, 1), (1, 0), (1, 1))) // onBatchCompleted val batchInfoCompleted = BatchInfo(Time(1000), receivedBlockInfo, 1000, Some(2000), None) @@ -198,7 +198,7 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { // We still can see the info retrieved from onJobStart listener.getBatchUIData(Time(1000 + limit * 100)) should be - Some(BatchUIData(batchInfoSubmitted, Seq((0, Seq(0))))) + Some(BatchUIData(batchInfoSubmitted, Seq(0 -> 0))) // A lot of "onBatchCompleted"s happen before "onJobStart" @@ -220,7 +220,7 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { } // We should not leak memory - listener.batchTimeToOutputOpIdToSparkJobIds.size() should be <= + listener.batchTimeToOutputOpIdSparkJobIdPair.size() should be <= (listener.waitingBatches.size + listener.runningBatches.size + listener.retainedCompletedBatches.size + 10) } From 087ba986de3dc6c16c49ca42a8c61439fdd2201e Mon Sep 17 00:00:00 2001 From: zsxwing Date: Tue, 28 Apr 2015 20:58:18 -0700 Subject: [PATCH 10/13] Refactor BatchInfo to store only necessary fields --- .../spark/streaming/scheduler/BatchInfo.scala | 7 -- .../spark/streaming/ui/AllBatchesTable.scala | 20 ++- .../apache/spark/streaming/ui/BatchPage.scala | 17 ++- .../spark/streaming/ui/BatchUIData.scala | 75 ++++++++++- .../ui/StreamingJobProgressListener.scala | 119 +++++++++--------- .../StreamingJobProgressListenerSuite.scala | 36 ++++-- 6 files changed, 172 insertions(+), 102 deletions(-) diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/BatchInfo.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/BatchInfo.scala index 9d21be50a566f..92dc113f397ca 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/BatchInfo.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/BatchInfo.scala @@ -58,11 +58,4 @@ case class BatchInfo( */ def totalDelay: Option[Long] = schedulingDelay.zip(processingDelay) .map(x => x._1 + x._2).headOption - - /** - * The number of recorders received by the receivers in this batch. - */ - def numRecords: Long = receivedBlockInfo.map { case (_, infos) => - infos.map(_.numRecords).sum - }.sum } diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala index a6fe4a3b49c54..e219e27785533 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala @@ -19,7 +19,6 @@ package org.apache.spark.streaming.ui import scala.xml.Node -import org.apache.spark.streaming.scheduler.BatchInfo import org.apache.spark.ui.UIUtils private[ui] abstract class BatchTableBase(tableId: String) { @@ -31,12 +30,10 @@ private[ui] abstract class BatchTableBase(tableId: String) {     } - protected def baseRow(batch: BatchInfo): Seq[Node] = { + protected def baseRow(batch: BatchUIData): Seq[Node] = { val batchTime = batch.batchTime.milliseconds val formattedBatchTime = UIUtils.formatDate(batch.batchTime.milliseconds) - val eventCount = batch.receivedBlockInfo.values.map { - receivers => receivers.map(_.numRecords).sum - }.sum + val eventCount = batch.numRecords val schedulingDelay = batch.schedulingDelay val formattedSchedulingDelay = schedulingDelay.map(UIUtils.formatDuration).getOrElse("-") val processingTime = batch.processingDelay @@ -77,8 +74,9 @@ private[ui] abstract class BatchTableBase(tableId: String) { protected def renderRows: Seq[Node] } -private[ui] class ActiveBatchTable(runningBatches: Seq[BatchInfo], waitingBatches: Seq[BatchInfo]) - extends BatchTableBase("active-batches-table") { +private[ui] class ActiveBatchTable( + runningBatches: Seq[BatchUIData], + waitingBatches: Seq[BatchUIData]) extends BatchTableBase("active-batches-table") { override protected def columns: Seq[Node] = super.columns ++ @@ -89,16 +87,16 @@ private[ui] class ActiveBatchTable(runningBatches: Seq[BatchInfo], waitingBatche runningBatches.flatMap(batch => {runningBatchRow(batch)}) } - private def runningBatchRow(batch: BatchInfo): Seq[Node] = { + private def runningBatchRow(batch: BatchUIData): Seq[Node] = { baseRow(batch) ++ } - private def waitingBatchRow(batch: BatchInfo): Seq[Node] = { + private def waitingBatchRow(batch: BatchUIData): Seq[Node] = { baseRow(batch) ++ } } -private[ui] class CompletedBatchTable(batches: Seq[BatchInfo]) +private[ui] class CompletedBatchTable(batches: Seq[BatchUIData]) extends BatchTableBase("completed-batches-table") { override protected def columns: Seq[Node] = super.columns ++ @@ -107,7 +105,7 @@ private[ui] class CompletedBatchTable(batches: Seq[BatchInfo]) batches.flatMap(batch => {completedBatchRow(batch)}) } - private def completedBatchRow(batch: BatchInfo): Seq[Node] = { + private def completedBatchRow(batch: BatchUIData): Seq[Node] = { val totalDelay = batch.totalDelay val formattedTotalDelay = totalDelay.map(UIUtils.formatDuration).getOrElse("-") baseRow(batch) ++ diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala index 77beebe2a2a56..2da9a29e2529e 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala @@ -19,13 +19,11 @@ package org.apache.spark.streaming.ui import javax.servlet.http.HttpServletRequest -import scala.collection.mutable.{ArrayBuffer, Map} import scala.xml.{NodeSeq, Node} import org.apache.commons.lang3.StringEscapeUtils import org.apache.spark.streaming.Time -import org.apache.spark.streaming.scheduler.BatchInfo import org.apache.spark.ui.{UIUtils, WebUIPage} import org.apache.spark.streaming.ui.StreamingJobProgressListener.{SparkJobId, OutputOpId} import org.apache.spark.ui.jobs.UIData.JobUIData @@ -182,11 +180,11 @@ private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { * Generate the job table for the batch. */ private def generateJobTable(batchUIData: BatchUIData): Seq[Node] = { - val outputOpIdToSparkJobIds = batchUIData.outputOpIdSparkJobIdPairs.groupBy(_._1).toSeq. + val outputOpIdToSparkJobIds = batchUIData.outputOpIdSparkJobIdPairs.groupBy(_.outputOpId).toSeq. sortBy(_._1). // sorted by OutputOpId - map { case (outputOpId, outputOpIdAndSparkJobIdPairs) => + map { case (outputOpId, outputOpIdAndSparkJobIds) => // sort SparkJobIds for each OutputOpId - (outputOpId, outputOpIdAndSparkJobIdPairs.map(_._2).sorted) + (outputOpId, outputOpIdAndSparkJobIds.map(_.sparkJobId).sorted) } sparkListener.synchronized { val outputOpIdWithJobs: Seq[(OutputOpId, Seq[JobUIData])] = @@ -219,13 +217,12 @@ private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { val batchUIData = streamingListener.getBatchUIData(batchTime).getOrElse { throw new IllegalArgumentException(s"Batch $formattedBatchTime does not exist") } - val batchInfo = batchUIData.batchInfo val formattedSchedulingDelay = - batchInfo.schedulingDelay.map(UIUtils.formatDuration).getOrElse("-") + batchUIData.schedulingDelay.map(UIUtils.formatDuration).getOrElse("-") val formattedProcessingTime = - batchInfo.processingDelay.map(UIUtils.formatDuration).getOrElse("-") - val formattedTotalDelay = batchInfo.totalDelay.map(UIUtils.formatDuration).getOrElse("-") + batchUIData.processingDelay.map(UIUtils.formatDuration).getOrElse("-") + val formattedTotalDelay = batchUIData.totalDelay.map(UIUtils.formatDuration).getOrElse("-") val summary: NodeSeq =
    @@ -236,7 +233,7 @@ private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") {
  • Input data size: - {batchInfo.numRecords} records + {batchUIData.numRecords} records
  • Scheduling delay: diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala index a7da139d4e4b3..c2ee869e5e5a7 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala @@ -18,10 +18,79 @@ package org.apache.spark.streaming.ui +import org.apache.spark.streaming.Time import org.apache.spark.streaming.scheduler.BatchInfo import org.apache.spark.streaming.ui.StreamingJobProgressListener._ -private[ui] case class BatchUIData( - batchInfo: BatchInfo, - outputOpIdSparkJobIdPairs: Seq[(OutputOpId, SparkJobId)]) { +private[ui] case class OutputOpIdAndSparkJobId(outputOpId: OutputOpId, sparkJobId: SparkJobId) + +private[ui] class BatchUIData( + val batchTime: Time, + val receiverNumRecords: Map[Int, Long], + val submissionTime: Long, + val processingStartTime: Option[Long], + val processingEndTime: Option[Long]) { + + var outputOpIdSparkJobIdPairs: Seq[OutputOpIdAndSparkJobId] = Seq.empty + + /** + * Time taken for the first job of this batch to start processing from the time this batch + * was submitted to the streaming scheduler. Essentially, it is + * `processingStartTime` - `submissionTime`. + */ + def schedulingDelay: Option[Long] = processingStartTime.map(_ - submissionTime) + + /** + * Time taken for the all jobs of this batch to finish processing from the time they started + * processing. Essentially, it is `processingEndTime` - `processingStartTime`. + */ + def processingDelay: Option[Long] = { + for (start <- processingStartTime; + end <- processingEndTime) + yield end - start + } + + /** + * Time taken for all the jobs of this batch to finish processing from the time they + * were submitted. Essentially, it is `processingDelay` + `schedulingDelay`. + */ + def totalDelay: Option[Long] = processingEndTime.map(_ - submissionTime) + + /** + * The number of recorders received by the receivers in this batch. + */ + def numRecords: Long = receiverNumRecords.map(_._2).sum + + def canEqual(other: Any): Boolean = other.isInstanceOf[BatchUIData] + + override def equals(other: Any): Boolean = other match { + case that: BatchUIData => + (that canEqual this) && + outputOpIdSparkJobIdPairs == that.outputOpIdSparkJobIdPairs && + batchTime == that.batchTime && + receiverNumRecords == that.receiverNumRecords && + submissionTime == that.submissionTime && + processingStartTime == that.processingStartTime && + processingEndTime == that.processingEndTime + case _ => false + } + + override def hashCode(): Int = { + val state = Seq(outputOpIdSparkJobIdPairs, batchTime, receiverNumRecords, submissionTime, + processingStartTime, processingEndTime) + state.map(_.hashCode()).foldLeft(0)((a, b) => 31 * a + b) + } +} + +private[ui] object BatchUIData { + + def apply(batchInfo: BatchInfo): BatchUIData = { + new BatchUIData( + batchInfo.batchTime, + batchInfo.receivedBlockInfo.mapValues(_.map(_.numRecords).sum), + batchInfo.submissionTime, + batchInfo.processingStartTime, + batchInfo.processingEndTime + ) + } } diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala index a3f190c4889ae..6218cf03d25e0 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala @@ -21,14 +21,13 @@ import java.util.LinkedHashMap import java.util.{Map => JMap} import java.util.Properties -import scala.collection.mutable.{ArrayBuffer, Queue, HashMap} +import scala.collection.mutable.{ArrayBuffer, Queue, HashMap, SynchronizedBuffer} import org.apache.spark.scheduler._ import org.apache.spark.streaming.{Time, StreamingContext} import org.apache.spark.streaming.scheduler._ import org.apache.spark.streaming.scheduler.StreamingListenerReceiverStarted import org.apache.spark.streaming.scheduler.StreamingListenerBatchStarted -import org.apache.spark.streaming.scheduler.BatchInfo import org.apache.spark.streaming.scheduler.StreamingListenerBatchSubmitted import org.apache.spark.util.Distribution @@ -36,24 +35,22 @@ import org.apache.spark.util.Distribution private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) extends StreamingListener with SparkListener { - import StreamingJobProgressListener._ - - private val waitingBatchInfos = new HashMap[Time, BatchInfo] - private val runningBatchInfos = new HashMap[Time, BatchInfo] - private val completedBatchInfos = new Queue[BatchInfo] - private val batchInfoLimit = ssc.conf.getInt("spark.streaming.ui.retainedBatches", 100) + private val waitingBatchUIDatas = new HashMap[Time, BatchUIData] + private val runningBatchUIDatas = new HashMap[Time, BatchUIData] + private val completedBatchUIDatas = new Queue[BatchUIData] + private val batchUIDataLimit = ssc.conf.getInt("spark.streaming.ui.retainedBatches", 100) private var totalCompletedBatches = 0L private var totalReceivedRecords = 0L private var totalProcessedRecords = 0L private val receiverInfos = new HashMap[Int, ReceiverInfo] // Because onJobStart and onBatchXXX messages are processed in different threads, - // we may not be able to get the corresponding BatchInfo when receiving onJobStart. So here we + // we may not be able to get the corresponding BatchUIData when receiving onJobStart. So here we // cannot use a map of (Time, BatchUIData). private[ui] val batchTimeToOutputOpIdSparkJobIdPair = - new LinkedHashMap[Time, ArrayBuffer[(OutputOpId, SparkJobId)]] { - override def removeEldestEntry(p1: JMap.Entry[Time, ArrayBuffer[(OutputOpId, SparkJobId)]]): - Boolean = { + new LinkedHashMap[Time, SynchronizedBuffer[OutputOpIdAndSparkJobId]] { + override def removeEldestEntry( + p1: JMap.Entry[Time, SynchronizedBuffer[OutputOpIdAndSparkJobId]]): Boolean = { // If a lot of "onBatchCompleted"s happen before "onJobStart" (image if // SparkContext.listenerBus is very slow), "batchTimeToOutputOpIdToSparkJobIds" // may add some information for a removed batch when processing "onJobStart". It will be a @@ -66,7 +63,8 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) // "batchTimeToOutputOpIdToSparkJobIds" may be greater than the number of the retained // batches temporarily, so here we use "10" to handle such case. This is not a perfect // solution, but at least it can handle most of cases. - size() > waitingBatchInfos.size + runningBatchInfos.size + completedBatchInfos.size + 10 + size() > + waitingBatchUIDatas.size + runningBatchUIDatas.size + completedBatchUIDatas.size + 10 } } @@ -93,29 +91,32 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) override def onBatchSubmitted(batchSubmitted: StreamingListenerBatchSubmitted): Unit = { synchronized { - waitingBatchInfos(batchSubmitted.batchInfo.batchTime) = batchSubmitted.batchInfo + waitingBatchUIDatas(batchSubmitted.batchInfo.batchTime) = + BatchUIData(batchSubmitted.batchInfo) } } override def onBatchStarted(batchStarted: StreamingListenerBatchStarted): Unit = synchronized { - runningBatchInfos(batchStarted.batchInfo.batchTime) = batchStarted.batchInfo - waitingBatchInfos.remove(batchStarted.batchInfo.batchTime) + val batchUIData = BatchUIData(batchStarted.batchInfo) + runningBatchUIDatas(batchStarted.batchInfo.batchTime) = BatchUIData(batchStarted.batchInfo) + waitingBatchUIDatas.remove(batchStarted.batchInfo.batchTime) - totalReceivedRecords += batchStarted.batchInfo.numRecords + totalReceivedRecords += batchUIData.numRecords } override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = { synchronized { - waitingBatchInfos.remove(batchCompleted.batchInfo.batchTime) - runningBatchInfos.remove(batchCompleted.batchInfo.batchTime) - completedBatchInfos.enqueue(batchCompleted.batchInfo) - if (completedBatchInfos.size > batchInfoLimit) { - val removedBatch = completedBatchInfos.dequeue() + waitingBatchUIDatas.remove(batchCompleted.batchInfo.batchTime) + runningBatchUIDatas.remove(batchCompleted.batchInfo.batchTime) + val batchUIData = BatchUIData(batchCompleted.batchInfo) + completedBatchUIDatas.enqueue(batchUIData) + if (completedBatchUIDatas.size > batchUIDataLimit) { + val removedBatch = completedBatchUIDatas.dequeue() batchTimeToOutputOpIdSparkJobIdPair.remove(removedBatch.batchTime) } totalCompletedBatches += 1L - totalProcessedRecords += batchCompleted.batchInfo.numRecords + totalProcessedRecords += batchUIData.numRecords } } @@ -123,10 +124,12 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) getBatchTimeAndOutputOpId(jobStart.properties).foreach { case (batchTime, outputOpId) => var outputOpIdToSparkJobIds = batchTimeToOutputOpIdSparkJobIdPair.get(batchTime) if (outputOpIdToSparkJobIds == null) { - outputOpIdToSparkJobIds = ArrayBuffer[(OutputOpId, SparkJobId)]() + outputOpIdToSparkJobIds = + new ArrayBuffer[OutputOpIdAndSparkJobId]() + with SynchronizedBuffer[OutputOpIdAndSparkJobId] batchTimeToOutputOpIdSparkJobIdPair.put(batchTime, outputOpIdToSparkJobIds) } - outputOpIdToSparkJobIds += (outputOpId -> jobStart.jobId) + outputOpIdToSparkJobIds += OutputOpIdAndSparkJobId(outputOpId, jobStart.jobId) } } @@ -157,19 +160,19 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) } def numUnprocessedBatches: Long = synchronized { - waitingBatchInfos.size + runningBatchInfos.size + waitingBatchUIDatas.size + runningBatchUIDatas.size } - def waitingBatches: Seq[BatchInfo] = synchronized { - waitingBatchInfos.values.toSeq + def waitingBatches: Seq[BatchUIData] = synchronized { + waitingBatchUIDatas.values.toSeq } - def runningBatches: Seq[BatchInfo] = synchronized { - runningBatchInfos.values.toSeq + def runningBatches: Seq[BatchUIData] = synchronized { + runningBatchUIDatas.values.toSeq } - def retainedCompletedBatches: Seq[BatchInfo] = synchronized { - completedBatchInfos.toSeq + def retainedCompletedBatches: Seq[BatchUIData] = synchronized { + completedBatchUIDatas.toSeq } def processingDelayDistribution: Option[Distribution] = synchronized { @@ -185,15 +188,11 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) } def receivedRecordsDistributions: Map[Int, Option[Distribution]] = synchronized { - val latestBatchInfos = retainedBatches.reverse.take(batchInfoLimit) - val latestBlockInfos = latestBatchInfos.map(_.receivedBlockInfo) + val latestBatches = retainedBatches.reverse.take(batchUIDataLimit) (0 until numReceivers).map { receiverId => - val blockInfoOfParticularReceiver = latestBlockInfos.map { batchInfo => - batchInfo.get(receiverId).getOrElse(Array.empty) - } - val recordsOfParticularReceiver = blockInfoOfParticularReceiver.map { blockInfo => - // calculate records per second for each batch - blockInfo.map(_.numRecords).sum.toDouble * 1000 / batchDuration + val recordsOfParticularReceiver = latestBatches.map { batch => + // calculate records per second for each batch + batch.receiverNumRecords.get(receiverId).sum.toDouble * 1000 / batchDuration } val distributionOption = Distribution(recordsOfParticularReceiver) (receiverId, distributionOption) @@ -201,10 +200,10 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) } def lastReceivedBatchRecords: Map[Int, Long] = synchronized { - val lastReceivedBlockInfoOption = lastReceivedBatch.map(_.receivedBlockInfo) + val lastReceivedBlockInfoOption = lastReceivedBatch.map(_.receiverNumRecords) lastReceivedBlockInfoOption.map { lastReceivedBlockInfo => (0 until numReceivers).map { receiverId => - (receiverId, lastReceivedBlockInfo(receiverId).map(_.numRecords).sum) + (receiverId, lastReceivedBlockInfo.getOrElse(receiverId, 0L)) }.toMap }.getOrElse { (0 until numReceivers).map(receiverId => (receiverId, 0L)).toMap @@ -215,37 +214,35 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) receiverInfos.get(receiverId) } - def lastCompletedBatch: Option[BatchInfo] = synchronized { - completedBatchInfos.sortBy(_.batchTime)(Time.ordering).lastOption + def lastCompletedBatch: Option[BatchUIData] = synchronized { + completedBatchUIDatas.sortBy(_.batchTime)(Time.ordering).lastOption } - def lastReceivedBatch: Option[BatchInfo] = synchronized { + def lastReceivedBatch: Option[BatchUIData] = synchronized { retainedBatches.lastOption } - private def retainedBatches: Seq[BatchInfo] = { - (waitingBatchInfos.values.toSeq ++ - runningBatchInfos.values.toSeq ++ completedBatchInfos).sortBy(_.batchTime)(Time.ordering) + private def retainedBatches: Seq[BatchUIData] = { + (waitingBatchUIDatas.values.toSeq ++ + runningBatchUIDatas.values.toSeq ++ completedBatchUIDatas).sortBy(_.batchTime)(Time.ordering) } - private def extractDistribution(getMetric: BatchInfo => Option[Long]): Option[Distribution] = { - Distribution(completedBatchInfos.flatMap(getMetric(_)).map(_.toDouble)) + private def extractDistribution(getMetric: BatchUIData => Option[Long]): Option[Distribution] = { + Distribution(completedBatchUIDatas.flatMap(getMetric(_)).map(_.toDouble)) } - private def getBatchInfo(batchTime: Time): Option[BatchInfo] = { - waitingBatchInfos.get(batchTime).orElse { - runningBatchInfos.get(batchTime).orElse { - completedBatchInfos.find(batch => batch.batchTime == batchTime) + def getBatchUIData(batchTime: Time): Option[BatchUIData] = synchronized { + val batchUIData = waitingBatchUIDatas.get(batchTime).orElse { + runningBatchUIDatas.get(batchTime).orElse { + completedBatchUIDatas.find(batch => batch.batchTime == batchTime) } } - } - - def getBatchUIData(batchTime: Time): Option[BatchUIData] = synchronized { - for (batchInfo <- getBatchInfo(batchTime)) yield { - val outputOpIdToSparkJobIds = Option(batchTimeToOutputOpIdSparkJobIdPair.get(batchTime)). - map(ArrayBuffer(_: _*)).getOrElse(Seq.empty) - BatchUIData(batchInfo, outputOpIdToSparkJobIds) + batchUIData.foreach { _batchUIData => + val outputOpIdToSparkJobIds = + Option(batchTimeToOutputOpIdSparkJobIdPair.get(batchTime)).getOrElse(Seq.empty) + _batchUIData.outputOpIdSparkJobIdPairs = outputOpIdToSparkJobIds } + batchUIData } } diff --git a/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala index 31b599a5ba28a..d29ecd9ee20d2 100644 --- a/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala +++ b/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala @@ -19,6 +19,7 @@ package org.apache.spark.streaming.ui import java.util.Properties +import org.apache.spark.streaming.ui.OutputOpIdAndSparkJobId import org.scalatest.Matchers import org.apache.spark.scheduler.SparkListenerJobStart @@ -46,7 +47,7 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { // onBatchSubmitted val batchInfoSubmitted = BatchInfo(Time(1000), receivedBlockInfo, 1000, None, None) listener.onBatchSubmitted(StreamingListenerBatchSubmitted(batchInfoSubmitted)) - listener.waitingBatches should be (List(batchInfoSubmitted)) + listener.waitingBatches should be (List(BatchUIData(batchInfoSubmitted))) listener.runningBatches should be (Nil) listener.retainedCompletedBatches should be (Nil) listener.lastCompletedBatch should be (None) @@ -59,7 +60,7 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { val batchInfoStarted = BatchInfo(Time(1000), receivedBlockInfo, 1000, Some(2000), None) listener.onBatchStarted(StreamingListenerBatchStarted(batchInfoStarted)) listener.waitingBatches should be (Nil) - listener.runningBatches should be (List(batchInfoStarted)) + listener.runningBatches should be (List(BatchUIData(batchInfoStarted))) listener.retainedCompletedBatches should be (Nil) listener.lastCompletedBatch should be (None) listener.numUnprocessedBatches should be (1) @@ -105,17 +106,26 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { listener.onJobStart(jobStart4) val batchUIData = listener.getBatchUIData(Time(1000)) - assert(batchUIData != None) - assert(batchUIData.get.batchInfo === batchInfoStarted) - assert(batchUIData.get.outputOpIdSparkJobIdPairs === Seq((0, 0), (0, 1), (1, 0), (1, 1))) + batchUIData should not be None + batchUIData.get.batchTime should be (batchInfoStarted.batchTime) + batchUIData.get.schedulingDelay should be (batchInfoStarted.schedulingDelay) + batchUIData.get.processingDelay should be (batchInfoStarted.processingDelay) + batchUIData.get.totalDelay should be (batchInfoStarted.totalDelay) + batchUIData.get.receiverNumRecords should be (Map(0 -> 300L, 1 -> 300L)) + batchUIData.get.numRecords should be(600) + batchUIData.get.outputOpIdSparkJobIdPairs should be + Seq(OutputOpIdAndSparkJobId(0, 0), + OutputOpIdAndSparkJobId(0, 1), + OutputOpIdAndSparkJobId(1, 0), + OutputOpIdAndSparkJobId(1, 1)) // onBatchCompleted val batchInfoCompleted = BatchInfo(Time(1000), receivedBlockInfo, 1000, Some(2000), None) listener.onBatchCompleted(StreamingListenerBatchCompleted(batchInfoCompleted)) listener.waitingBatches should be (Nil) listener.runningBatches should be (Nil) - listener.retainedCompletedBatches should be (List(batchInfoCompleted)) - listener.lastCompletedBatch should be (Some(batchInfoCompleted)) + listener.retainedCompletedBatches should be (List(BatchUIData(batchInfoCompleted))) + listener.lastCompletedBatch should be (Some(BatchUIData(batchInfoCompleted))) listener.numUnprocessedBatches should be (0) listener.numTotalCompletedBatches should be (1) listener.numTotalProcessedRecords should be (600) @@ -197,9 +207,15 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { listener.onBatchSubmitted(StreamingListenerBatchSubmitted(batchInfoSubmitted)) // We still can see the info retrieved from onJobStart - listener.getBatchUIData(Time(1000 + limit * 100)) should be - Some(BatchUIData(batchInfoSubmitted, Seq(0 -> 0))) - + val batchUIData = listener.getBatchUIData(Time(1000 + limit * 100)) + batchUIData should not be None + batchUIData.get.batchTime should be (batchInfoSubmitted.batchTime) + batchUIData.get.schedulingDelay should be (batchInfoSubmitted.schedulingDelay) + batchUIData.get.processingDelay should be (batchInfoSubmitted.processingDelay) + batchUIData.get.totalDelay should be (batchInfoSubmitted.totalDelay) + batchUIData.get.receiverNumRecords should be (Map.empty) + batchUIData.get.numRecords should be (0) + batchUIData.get.outputOpIdSparkJobIdPairs should be (Seq(OutputOpIdAndSparkJobId(0, 0))) // A lot of "onBatchCompleted"s happen before "onJobStart" for(i <- limit + 1 to limit * 2) { From 9a3083d65575c1acd0b259cee1dc307d562d3ee0 Mon Sep 17 00:00:00 2001 From: zsxwing Date: Wed, 29 Apr 2015 07:26:03 -0700 Subject: [PATCH 11/13] Rename XxxDatas -> XxxData --- .../ui/StreamingJobProgressListener.scala | 46 +++++++++---------- 1 file changed, 23 insertions(+), 23 deletions(-) diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala index 6218cf03d25e0..34b55717a1db2 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala @@ -35,9 +35,9 @@ import org.apache.spark.util.Distribution private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) extends StreamingListener with SparkListener { - private val waitingBatchUIDatas = new HashMap[Time, BatchUIData] - private val runningBatchUIDatas = new HashMap[Time, BatchUIData] - private val completedBatchUIDatas = new Queue[BatchUIData] + private val waitingBatchUIData = new HashMap[Time, BatchUIData] + private val runningBatchUIData = new HashMap[Time, BatchUIData] + private val completedBatchUIData = new Queue[BatchUIData] private val batchUIDataLimit = ssc.conf.getInt("spark.streaming.ui.retainedBatches", 100) private var totalCompletedBatches = 0L private var totalReceivedRecords = 0L @@ -64,7 +64,7 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) // batches temporarily, so here we use "10" to handle such case. This is not a perfect // solution, but at least it can handle most of cases. size() > - waitingBatchUIDatas.size + runningBatchUIDatas.size + completedBatchUIDatas.size + 10 + waitingBatchUIData.size + runningBatchUIData.size + completedBatchUIData.size + 10 } } @@ -91,27 +91,27 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) override def onBatchSubmitted(batchSubmitted: StreamingListenerBatchSubmitted): Unit = { synchronized { - waitingBatchUIDatas(batchSubmitted.batchInfo.batchTime) = + waitingBatchUIData(batchSubmitted.batchInfo.batchTime) = BatchUIData(batchSubmitted.batchInfo) } } override def onBatchStarted(batchStarted: StreamingListenerBatchStarted): Unit = synchronized { val batchUIData = BatchUIData(batchStarted.batchInfo) - runningBatchUIDatas(batchStarted.batchInfo.batchTime) = BatchUIData(batchStarted.batchInfo) - waitingBatchUIDatas.remove(batchStarted.batchInfo.batchTime) + runningBatchUIData(batchStarted.batchInfo.batchTime) = BatchUIData(batchStarted.batchInfo) + waitingBatchUIData.remove(batchStarted.batchInfo.batchTime) totalReceivedRecords += batchUIData.numRecords } override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = { synchronized { - waitingBatchUIDatas.remove(batchCompleted.batchInfo.batchTime) - runningBatchUIDatas.remove(batchCompleted.batchInfo.batchTime) + waitingBatchUIData.remove(batchCompleted.batchInfo.batchTime) + runningBatchUIData.remove(batchCompleted.batchInfo.batchTime) val batchUIData = BatchUIData(batchCompleted.batchInfo) - completedBatchUIDatas.enqueue(batchUIData) - if (completedBatchUIDatas.size > batchUIDataLimit) { - val removedBatch = completedBatchUIDatas.dequeue() + completedBatchUIData.enqueue(batchUIData) + if (completedBatchUIData.size > batchUIDataLimit) { + val removedBatch = completedBatchUIData.dequeue() batchTimeToOutputOpIdSparkJobIdPair.remove(removedBatch.batchTime) } totalCompletedBatches += 1L @@ -160,19 +160,19 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) } def numUnprocessedBatches: Long = synchronized { - waitingBatchUIDatas.size + runningBatchUIDatas.size + waitingBatchUIData.size + runningBatchUIData.size } def waitingBatches: Seq[BatchUIData] = synchronized { - waitingBatchUIDatas.values.toSeq + waitingBatchUIData.values.toSeq } def runningBatches: Seq[BatchUIData] = synchronized { - runningBatchUIDatas.values.toSeq + runningBatchUIData.values.toSeq } def retainedCompletedBatches: Seq[BatchUIData] = synchronized { - completedBatchUIDatas.toSeq + completedBatchUIData.toSeq } def processingDelayDistribution: Option[Distribution] = synchronized { @@ -215,7 +215,7 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) } def lastCompletedBatch: Option[BatchUIData] = synchronized { - completedBatchUIDatas.sortBy(_.batchTime)(Time.ordering).lastOption + completedBatchUIData.sortBy(_.batchTime)(Time.ordering).lastOption } def lastReceivedBatch: Option[BatchUIData] = synchronized { @@ -223,18 +223,18 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext) } private def retainedBatches: Seq[BatchUIData] = { - (waitingBatchUIDatas.values.toSeq ++ - runningBatchUIDatas.values.toSeq ++ completedBatchUIDatas).sortBy(_.batchTime)(Time.ordering) + (waitingBatchUIData.values.toSeq ++ + runningBatchUIData.values.toSeq ++ completedBatchUIData).sortBy(_.batchTime)(Time.ordering) } private def extractDistribution(getMetric: BatchUIData => Option[Long]): Option[Distribution] = { - Distribution(completedBatchUIDatas.flatMap(getMetric(_)).map(_.toDouble)) + Distribution(completedBatchUIData.flatMap(getMetric(_)).map(_.toDouble)) } def getBatchUIData(batchTime: Time): Option[BatchUIData] = synchronized { - val batchUIData = waitingBatchUIDatas.get(batchTime).orElse { - runningBatchUIDatas.get(batchTime).orElse { - completedBatchUIDatas.find(batch => batch.batchTime == batchTime) + val batchUIData = waitingBatchUIData.get(batchTime).orElse { + runningBatchUIData.get(batchTime).orElse { + completedBatchUIData.find(batch => batch.batchTime == batchTime) } } batchUIData.foreach { _batchUIData => From b380cfbe2b18304e5f3117cabb0936fff6f450e1 Mon Sep 17 00:00:00 2001 From: zsxwing Date: Wed, 29 Apr 2015 07:22:00 -0700 Subject: [PATCH 12/13] Add createJobStart to eliminate duplicate codes --- .../spark/streaming/UISeleniumSuite.scala | 2 +- .../StreamingJobProgressListenerSuite.scala | 69 +++++-------------- 2 files changed, 20 insertions(+), 51 deletions(-) diff --git a/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala index 405a4e90ff093..8de43baabc21d 100644 --- a/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala +++ b/streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala @@ -147,7 +147,7 @@ class UISeleniumSuite findAll(cssSelector(""".progress-cell""")).map(_.text).toSeq should be (List("1/1", "1/1", "1/1", "0/1 (1 failed)")) - // Check stacktrack + // Check stacktrace val errorCells = findAll(cssSelector(""".stacktrace-details""")).map(_.text).toSeq errorCells should have size 1 errorCells(0) should include("java.lang.RuntimeException: Oops") diff --git a/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala index d29ecd9ee20d2..1a85b3ebff73d 100644 --- a/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala +++ b/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala @@ -32,6 +32,17 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { val input = (1 to 4).map(Seq(_)).toSeq val operation = (d: DStream[Int]) => d.map(x => x) + private def createJobStart( + batchTime: Time, outputOpId: Int, jobId: Int): SparkListenerJobStart = { + val properties = new Properties() + properties.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, batchTime.milliseconds.toString) + properties.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, outputOpId.toString) + SparkListenerJobStart(jobId = jobId, + 0L, // unused + Nil, // unused + properties) + } + override def batchDuration: Duration = Milliseconds(100) test("onBatchSubmitted, onBatchStarted, onBatchCompleted, " + @@ -69,40 +80,16 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { listener.numTotalReceivedRecords should be (600) // onJobStart - val properties1 = new Properties() - properties1.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, Time(1000).milliseconds.toString) - properties1.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, 0.toString) - val jobStart1 = SparkListenerJobStart(jobId = 0, - 0L, // unused - Nil, // unused - properties1) + val jobStart1 = createJobStart(Time(1000), outputOpId = 0, jobId = 0) listener.onJobStart(jobStart1) - val properties2 = new Properties() - properties2.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, Time(1000).milliseconds.toString) - properties2.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, 0.toString) - val jobStart2 = SparkListenerJobStart(jobId = 1, - 0L, // unused - Nil, // unused - properties2) + val jobStart2 = createJobStart(Time(1000), outputOpId = 0, jobId = 1) listener.onJobStart(jobStart2) - val properties3 = new Properties() - properties3.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, Time(1000).milliseconds.toString) - properties3.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, 1.toString) - val jobStart3 = SparkListenerJobStart(jobId = 0, - 0L, // unused - Nil, // unused - properties3) + val jobStart3 = createJobStart(Time(1000), outputOpId = 1, jobId = 0) listener.onJobStart(jobStart3) - val properties4 = new Properties() - properties4.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, Time(1000).milliseconds.toString) - properties4.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, 1.toString) - val jobStart4 = SparkListenerJobStart(jobId = 1, - 0L, // unused - Nil, // unused - properties4) + val jobStart4 = createJobStart(Time(1000), outputOpId = 1, jobId = 1) listener.onJobStart(jobStart4) val batchUIData = listener.getBatchUIData(Time(1000)) @@ -172,7 +159,7 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { listener.numTotalCompletedBatches should be(limit + 10) } - test("disorder onJobStart and onBatchXXX") { + test("out-of-order onJobStart and onBatchXXX") { val ssc = setupStreams(input, operation) val limit = ssc.conf.getInt("spark.streaming.ui.retainedBatches", 100) val listener = new StreamingJobProgressListener(ssc) @@ -181,25 +168,13 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { for(i <- 0 until limit) { val batchInfoCompleted = BatchInfo(Time(1000 + i * 100), Map.empty, 1000 + i * 100, Some(2000 + i * 100), None) - val properties = new Properties() - properties.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, (1000 + i * 100).toString) - properties.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, "0") - val jobStart = SparkListenerJobStart(jobId = 1, - 0L, // unused - Nil, // unused - properties) listener.onBatchCompleted(StreamingListenerBatchCompleted(batchInfoCompleted)) + val jobStart = createJobStart(Time(1000 + i * 100), outputOpId = 0, jobId = 1) listener.onJobStart(jobStart) } // onJobStart happens before onBatchSubmitted - val properties = new Properties() - properties.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, (1000 + limit * 100).toString) - properties.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, "0") - val jobStart = SparkListenerJobStart(jobId = 0, - 0L, // unused - Nil, // unused - properties) + val jobStart = createJobStart(Time(1000 + limit * 100), outputOpId = 0, jobId = 0) listener.onJobStart(jobStart) val batchInfoSubmitted = @@ -225,13 +200,7 @@ class StreamingJobProgressListenerSuite extends TestSuiteBase with Matchers { } for(i <- limit + 1 to limit * 2) { - val properties = new Properties() - properties.setProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, (1000 + i * 100).toString) - properties.setProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, "0") - val jobStart = SparkListenerJobStart(jobId = 1, - 0L, // unused - Nil, // unused - properties) + val jobStart = createJobStart(Time(1000 + i * 100), outputOpId = 0, jobId = 1) listener.onJobStart(jobStart) } From 0727d35a153da9f27516e1d3834046e5ed6a6088 Mon Sep 17 00:00:00 2001 From: zsxwing Date: Wed, 29 Apr 2015 07:39:20 -0700 Subject: [PATCH 13/13] Change BatchUIData to a case class --- .../spark/streaming/ui/BatchUIData.scala | 27 +++---------------- .../StreamingJobProgressListenerSuite.scala | 1 - 2 files changed, 3 insertions(+), 25 deletions(-) diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala index c2ee869e5e5a7..f45c291b7c0fe 100644 --- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala +++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchUIData.scala @@ -24,14 +24,13 @@ import org.apache.spark.streaming.ui.StreamingJobProgressListener._ private[ui] case class OutputOpIdAndSparkJobId(outputOpId: OutputOpId, sparkJobId: SparkJobId) -private[ui] class BatchUIData( +private[ui] case class BatchUIData( val batchTime: Time, val receiverNumRecords: Map[Int, Long], val submissionTime: Long, val processingStartTime: Option[Long], - val processingEndTime: Option[Long]) { - - var outputOpIdSparkJobIdPairs: Seq[OutputOpIdAndSparkJobId] = Seq.empty + val processingEndTime: Option[Long], + var outputOpIdSparkJobIdPairs: Seq[OutputOpIdAndSparkJobId] = Seq.empty) { /** * Time taken for the first job of this batch to start processing from the time this batch @@ -60,26 +59,6 @@ private[ui] class BatchUIData( * The number of recorders received by the receivers in this batch. */ def numRecords: Long = receiverNumRecords.map(_._2).sum - - def canEqual(other: Any): Boolean = other.isInstanceOf[BatchUIData] - - override def equals(other: Any): Boolean = other match { - case that: BatchUIData => - (that canEqual this) && - outputOpIdSparkJobIdPairs == that.outputOpIdSparkJobIdPairs && - batchTime == that.batchTime && - receiverNumRecords == that.receiverNumRecords && - submissionTime == that.submissionTime && - processingStartTime == that.processingStartTime && - processingEndTime == that.processingEndTime - case _ => false - } - - override def hashCode(): Int = { - val state = Seq(outputOpIdSparkJobIdPairs, batchTime, receiverNumRecords, submissionTime, - processingStartTime, processingEndTime) - state.map(_.hashCode()).foldLeft(0)((a, b) => 31 * a + b) - } } private[ui] object BatchUIData { diff --git a/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala index 1a85b3ebff73d..fa89536de4054 100644 --- a/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala +++ b/streaming/src/test/scala/org/apache/spark/streaming/ui/StreamingJobProgressListenerSuite.scala @@ -19,7 +19,6 @@ package org.apache.spark.streaming.ui import java.util.Properties -import org.apache.spark.streaming.ui.OutputOpIdAndSparkJobId import org.scalatest.Matchers import org.apache.spark.scheduler.SparkListenerJobStart
    • {outputOpId.toString}{outputOpId.toString} {lastStageDescription} @@ -92,6 +93,7 @@ private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") { } else { Nil } + // scalastyle:on
    Processing TimeStatus
    processingqueuedTotal Delay