@@ -69,555 +68,521 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We
// if we find that it's okay.
private val MAX_TIMELINE_TASKS = parent.conf.getInt("spark.ui.timeline.tasks.maximum", 1000)
- private def getLocalitySummaryString(stageData: StageUIData): String = {
- val localities = stageData.taskData.values.map(_.taskInfo.taskLocality)
+ private def getLocalitySummaryString(stageData: StageData, taskList: Seq[TaskData]): String = {
+ val localities = taskList.map(_.taskLocality)
val localityCounts = localities.groupBy(identity).mapValues(_.size)
+ val names = Map(
+ TaskLocality.PROCESS_LOCAL.toString() -> "Process local",
+ TaskLocality.NODE_LOCAL.toString() -> "Node local",
+ TaskLocality.RACK_LOCAL.toString() -> "Rack local",
+ TaskLocality.ANY.toString() -> "Any")
val localityNamesAndCounts = localityCounts.toSeq.map { case (locality, count) =>
- val localityName = locality match {
- case TaskLocality.PROCESS_LOCAL => "Process local"
- case TaskLocality.NODE_LOCAL => "Node local"
- case TaskLocality.RACK_LOCAL => "Rack local"
- case TaskLocality.ANY => "Any"
- }
- s"$localityName: $count"
+ s"${names(locality)}: $count"
}
localityNamesAndCounts.sorted.mkString("; ")
}
def render(request: HttpServletRequest): Seq[Node] = {
- progressListener.synchronized {
- // stripXSS is called first to remove suspicious characters used in XSS attacks
- val parameterId = UIUtils.stripXSS(request.getParameter("id"))
- require(parameterId != null && parameterId.nonEmpty, "Missing id parameter")
-
- val parameterAttempt = UIUtils.stripXSS(request.getParameter("attempt"))
- require(parameterAttempt != null && parameterAttempt.nonEmpty, "Missing attempt parameter")
-
- val parameterTaskPage = UIUtils.stripXSS(request.getParameter("task.page"))
- val parameterTaskSortColumn = UIUtils.stripXSS(request.getParameter("task.sort"))
- val parameterTaskSortDesc = UIUtils.stripXSS(request.getParameter("task.desc"))
- val parameterTaskPageSize = UIUtils.stripXSS(request.getParameter("task.pageSize"))
- val parameterTaskPrevPageSize = UIUtils.stripXSS(request.getParameter("task.prevPageSize"))
-
- val taskPage = Option(parameterTaskPage).map(_.toInt).getOrElse(1)
- val taskSortColumn = Option(parameterTaskSortColumn).map { sortColumn =>
- UIUtils.decodeURLParameter(sortColumn)
- }.getOrElse("Index")
- val taskSortDesc = Option(parameterTaskSortDesc).map(_.toBoolean).getOrElse(false)
- val taskPageSize = Option(parameterTaskPageSize).map(_.toInt).getOrElse(100)
- val taskPrevPageSize = Option(parameterTaskPrevPageSize).map(_.toInt).getOrElse(taskPageSize)
-
- val stageId = parameterId.toInt
- val stageAttemptId = parameterAttempt.toInt
- val stageDataOption = progressListener.stageIdToData.get((stageId, stageAttemptId))
-
- val stageHeader = s"Details for Stage $stageId (Attempt $stageAttemptId)"
- if (stageDataOption.isEmpty) {
+ // stripXSS is called first to remove suspicious characters used in XSS attacks
+ val parameterId = UIUtils.stripXSS(request.getParameter("id"))
+ require(parameterId != null && parameterId.nonEmpty, "Missing id parameter")
+
+ val parameterAttempt = UIUtils.stripXSS(request.getParameter("attempt"))
+ require(parameterAttempt != null && parameterAttempt.nonEmpty, "Missing attempt parameter")
+
+ val parameterTaskPage = UIUtils.stripXSS(request.getParameter("task.page"))
+ val parameterTaskSortColumn = UIUtils.stripXSS(request.getParameter("task.sort"))
+ val parameterTaskSortDesc = UIUtils.stripXSS(request.getParameter("task.desc"))
+ val parameterTaskPageSize = UIUtils.stripXSS(request.getParameter("task.pageSize"))
+ val parameterTaskPrevPageSize = UIUtils.stripXSS(request.getParameter("task.prevPageSize"))
+
+ val taskPage = Option(parameterTaskPage).map(_.toInt).getOrElse(1)
+ val taskSortColumn = Option(parameterTaskSortColumn).map { sortColumn =>
+ UIUtils.decodeURLParameter(sortColumn)
+ }.getOrElse("Index")
+ val taskSortDesc = Option(parameterTaskSortDesc).map(_.toBoolean).getOrElse(false)
+ val taskPageSize = Option(parameterTaskPageSize).map(_.toInt).getOrElse(100)
+ val taskPrevPageSize = Option(parameterTaskPrevPageSize).map(_.toInt).getOrElse(taskPageSize)
+
+ val stageId = parameterId.toInt
+ val stageAttemptId = parameterAttempt.toInt
+
+ val stageHeader = s"Details for Stage $stageId (Attempt $stageAttemptId)"
+ val stageData = parent.store
+ .asOption(parent.store.stageAttempt(stageId, stageAttemptId, details = true))
+ .getOrElse {
val content =
No information to display for Stage {stageId} (Attempt {stageAttemptId})
return UIUtils.headerSparkPage(stageHeader, content, parent)
-
- }
- if (stageDataOption.get.taskData.isEmpty) {
- val content =
-
-
Summary Metrics No tasks have started yet
- Tasks No tasks have started yet
-
- return UIUtils.headerSparkPage(stageHeader, content, parent)
}
- val stageData = stageDataOption.get
- val tasks = stageData.taskData.values.toSeq.sortBy(_.taskInfo.launchTime)
- val numCompleted = stageData.numCompleteTasks
- val totalTasks = stageData.numActiveTasks +
- stageData.numCompleteTasks + stageData.numFailedTasks
- val totalTasksNumStr = if (totalTasks == tasks.size) {
- s"$totalTasks"
- } else {
- s"$totalTasks, showing ${tasks.size}"
- }
+ val tasks = stageData.tasks.getOrElse(Map.empty).values.toSeq
+ if (tasks.isEmpty) {
+ val content =
+
+
Summary Metrics No tasks have started yet
+ Tasks No tasks have started yet
+
+ return UIUtils.headerSparkPage(stageHeader, content, parent)
+ }
- val allAccumulables = progressListener.stageIdToData((stageId, stageAttemptId)).accumulables
- val externalAccumulables = allAccumulables.values.filter { acc => !acc.internal }
- val hasAccumulators = externalAccumulables.nonEmpty
+ val numCompleted = stageData.numCompleteTasks
+ val totalTasks = stageData.numActiveTasks + stageData.numCompleteTasks +
+ stageData.numFailedTasks + stageData.numKilledTasks
+ val totalTasksNumStr = if (totalTasks == tasks.size) {
+ s"$totalTasks"
+ } else {
+ s"$totalTasks, showing ${tasks.size}"
+ }
- val summary =
-
-
+ val externalAccumulables = stageData.accumulatorUpdates
+ val hasAccumulators = externalAccumulables.size > 0
+
+ val summary =
+
+
+
+ Total Time Across All Tasks:
+ {UIUtils.formatDuration(stageData.executorRunTime)}
+
+
+ Locality Level Summary:
+ {getLocalitySummaryString(stageData, tasks)}
+
+ {if (hasInput(stageData)) {
- Total Time Across All Tasks:
- {UIUtils.formatDuration(stageData.executorRunTime)}
+ Input Size / Records:
+ {s"${Utils.bytesToString(stageData.inputBytes)} / ${stageData.inputRecords}"}
+ }}
+ {if (hasOutput(stageData)) {
- Locality Level Summary:
- {getLocalitySummaryString(stageData)}
+ Output:
+ {s"${Utils.bytesToString(stageData.outputBytes)} / ${stageData.outputRecords}"}
- {if (stageData.hasInput) {
-
- Input Size / Records:
- {s"${Utils.bytesToString(stageData.inputBytes)} / ${stageData.inputRecords}"}
-
- }}
- {if (stageData.hasOutput) {
-
- Output:
- {s"${Utils.bytesToString(stageData.outputBytes)} / ${stageData.outputRecords}"}
-
- }}
- {if (stageData.hasShuffleRead) {
-
- Shuffle Read:
- {s"${Utils.bytesToString(stageData.shuffleReadTotalBytes)} / " +
- s"${stageData.shuffleReadRecords}"}
-
- }}
- {if (stageData.hasShuffleWrite) {
-
- Shuffle Write:
- {s"${Utils.bytesToString(stageData.shuffleWriteBytes)} / " +
- s"${stageData.shuffleWriteRecords}"}
-
- }}
- {if (stageData.hasBytesSpilled) {
-
- Shuffle Spill (Memory):
- {Utils.bytesToString(stageData.memoryBytesSpilled)}
-
-
- Shuffle Spill (Disk):
- {Utils.bytesToString(stageData.diskBytesSpilled)}
-
- }}
-
-
+ Shuffle Read:
+ {s"${Utils.bytesToString(stageData.shuffleReadBytes)} / " +
+ s"${stageData.shuffleReadRecords}"}
+
+ }}
+ {if (hasShuffleWrite(stageData)) {
+
+ Shuffle Write:
+ {s"${Utils.bytesToString(stageData.shuffleWriteBytes)} / " +
+ s"${stageData.shuffleWriteRecords}"}
+
+ }}
+ {if (hasBytesSpilled(stageData)) {
+
+ Shuffle Spill (Memory):
+ {Utils.bytesToString(stageData.memoryBytesSpilled)}
+
+
+ Shuffle Spill (Disk):
+ {Utils.bytesToString(stageData.diskBytesSpilled)}
+
+ }}
+
+
-
- Show Additional Metrics
-
-
-
-
- (De)select All
-
-
- Scheduler Delay
-
-
-
-
- Task Deserialization Time
-
-
- {if (stageData.hasShuffleRead) {
-
-
- Shuffle Read Blocked Time
-
-
-
-
- Shuffle Remote Reads
-
-
- }}
-
-
- Result Serialization Time
-
-
+ val showAdditionalMetrics =
+
-
- val dagViz = UIUtils.showDagVizForStage(
- stageId, operationGraphListener.getOperationGraphForStage(stageId))
-
- val accumulableHeaders: Seq[String] = Seq("Accumulable", "Value")
- def accumulableRow(acc: AccumulableInfo): Seq[Node] = {
- (acc.name, acc.value) match {
- case (Some(name), Some(value)) => {name} {value}
-
Error while rendering stage table:
-
- {Utils.exceptionString(e)}
-
-
{UIUtils.formatDuration(millis.toLong)}
- }
- }
- def getFormattedSizeQuantiles(data: Seq[Double]): Seq[Elem] = {
- getDistributionQuantiles(data).map(d => {Utils.bytesToString(d.toLong)} )
- }
-
- val deserializationTimes = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.executorDeserializeTime.toDouble
- }
- val deserializationQuantiles =
-
-
- Task Deserialization Time
-
- +: getFormattedTimeQuantiles(deserializationTimes)
-
- val serviceTimes = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.executorRunTime.toDouble
- }
- val serviceQuantiles = Duration +: getFormattedTimeQuantiles(serviceTimes)
-
- val gcTimes = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.jvmGCTime.toDouble
- }
- val gcQuantiles =
-
- GC Time
-
- +: getFormattedTimeQuantiles(gcTimes)
-
- val serializationTimes = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.resultSerializationTime.toDouble
- }
- val serializationQuantiles =
-
+ }}
+
- Result Serialization Time
+ Result Serialization Time
- +: getFormattedTimeQuantiles(serializationTimes)
-
- val gettingResultTimes = validTasks.map { taskUIData: TaskUIData =>
- getGettingResultTime(taskUIData.taskInfo, currentTime).toDouble
- }
- val gettingResultQuantiles =
-
+
+
- Getting Result Time
+ title={ToolTips.GETTING_RESULT_TIME} data-placement="right">
+ Getting Result Time
- +:
- getFormattedTimeQuantiles(gettingResultTimes)
-
- val peakExecutionMemory = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.peakExecutionMemory.toDouble
- }
- val peakExecutionMemoryQuantiles = {
-
+
+
- Peak Execution Memory
+ Peak Execution Memory
- +: getFormattedSizeQuantiles(peakExecutionMemory)
- }
-
- // The scheduler delay includes the network delay to send the task to the worker
- // machine and to send back the result (but not the time to fetch the task result,
- // if it needed to be fetched from the block manager on the worker).
- val schedulerDelays = validTasks.map { taskUIData: TaskUIData =>
- getSchedulerDelay(taskUIData.taskInfo, taskUIData.metrics.get, currentTime).toDouble
- }
- val schedulerDelayTitle = Scheduler Delay {s"${Utils.bytesToString(d.toLong)} / ${recordDist.next().toLong}"}
- )
- }
-
- val inputSizes = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.inputMetrics.bytesRead.toDouble
- }
-
- val inputRecords = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.inputMetrics.recordsRead.toDouble
- }
-
- val inputQuantiles = Input Size / Records +:
- getFormattedSizeQuantilesWithRecords(inputSizes, inputRecords)
+
+
+
+
{acc.name} {acc.value}
+
Error while rendering stage table:
+
+ {Utils.exceptionString(e)}
+
+
Output Size / Records +:
- getFormattedSizeQuantilesWithRecords(outputSizes, outputRecords)
+ val jsForScrollingDownToTaskTable =
+
- val shuffleReadBlockedTimes = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.shuffleReadMetrics.fetchWaitTime.toDouble
- }
- val shuffleReadBlockedQuantiles =
-
-
- Shuffle Read Blocked Time
-
- +:
- getFormattedTimeQuantiles(shuffleReadBlockedTimes)
+ val taskIdsInPage = if (taskTable == null) Set.empty[Long]
+ else taskTable.dataSource.slicedTaskIds
- val shuffleReadTotalSizes = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.shuffleReadMetrics.totalBytesRead.toDouble
- }
- val shuffleReadTotalRecords = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.shuffleReadMetrics.recordsRead.toDouble
- }
- val shuffleReadTotalQuantiles =
-
-
- Shuffle Read Size / Records
-
- +:
- getFormattedSizeQuantilesWithRecords(shuffleReadTotalSizes, shuffleReadTotalRecords)
+ // Excludes tasks which failed and have incomplete metrics
+ val validTasks = tasks.filter(t => t.status == "SUCCESS" && t.taskMetrics.isDefined)
- val shuffleReadRemoteSizes = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.shuffleReadMetrics.remoteBytesRead.toDouble
+ val summaryTable: Option[Seq[Node]] =
+ if (validTasks.size == 0) {
+ None
+ } else {
+ def getDistributionQuantiles(data: Seq[Double]): IndexedSeq[Double] = {
+ Distribution(data).get.getQuantiles()
+ }
+ def getFormattedTimeQuantiles(times: Seq[Double]): Seq[Node] = {
+ getDistributionQuantiles(times).map { millis =>
+ {UIUtils.formatDuration(millis.toLong)}
}
- val shuffleReadRemoteQuantiles =
-
-
- Shuffle Remote Reads
-
- +:
- getFormattedSizeQuantiles(shuffleReadRemoteSizes)
+ }
+ def getFormattedSizeQuantiles(data: Seq[Double]): Seq[Elem] = {
+ getDistributionQuantiles(data).map(d => {Utils.bytesToString(d.toLong)} )
+ }
- val shuffleWriteSizes = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.shuffleWriteMetrics.bytesWritten.toDouble
- }
+ val deserializationTimes = validTasks.map { task =>
+ task.taskMetrics.get.executorDeserializeTime.toDouble
+ }
+ val deserializationQuantiles =
+
+
+ Task Deserialization Time
+
+ +: getFormattedTimeQuantiles(deserializationTimes)
+
+ val serviceTimes = validTasks.map(_.taskMetrics.get.executorRunTime.toDouble)
+ val serviceQuantiles = Duration +: getFormattedTimeQuantiles(serviceTimes)
+
+ val gcTimes = validTasks.map(_.taskMetrics.get.jvmGcTime.toDouble)
+ val gcQuantiles =
+
+ GC Time
+
+ +: getFormattedTimeQuantiles(gcTimes)
+
+ val serializationTimes = validTasks.map(_.taskMetrics.get.resultSerializationTime.toDouble)
+ val serializationQuantiles =
+
+
+ Result Serialization Time
+
+ +: getFormattedTimeQuantiles(serializationTimes)
+
+ val gettingResultTimes = validTasks.map(getGettingResultTime(_, currentTime).toDouble)
+ val gettingResultQuantiles =
+
+
+ Getting Result Time
+
+ +:
+ getFormattedTimeQuantiles(gettingResultTimes)
+
+ val peakExecutionMemory = validTasks.map(_.taskMetrics.get.peakExecutionMemory.toDouble)
+ val peakExecutionMemoryQuantiles = {
+
+
+ Peak Execution Memory
+
+ +: getFormattedSizeQuantiles(peakExecutionMemory)
+ }
- val shuffleWriteRecords = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.shuffleWriteMetrics.recordsWritten.toDouble
- }
+ // The scheduler delay includes the network delay to send the task to the worker
+ // machine and to send back the result (but not the time to fetch the task result,
+ // if it needed to be fetched from the block manager on the worker).
+ val schedulerDelays = validTasks.map { task =>
+ getSchedulerDelay(task, task.taskMetrics.get, currentTime).toDouble
+ }
+ val schedulerDelayTitle = Scheduler Delay {s"${Utils.bytesToString(d.toLong)} / ${recordDist.next().toLong}"}
+ )
+ }
- val shuffleWriteQuantiles = Shuffle Write Size / Records +:
- getFormattedSizeQuantilesWithRecords(shuffleWriteSizes, shuffleWriteRecords)
+ val inputSizes = validTasks.map(_.taskMetrics.get.inputMetrics.bytesRead.toDouble)
+ val inputRecords = validTasks.map(_.taskMetrics.get.inputMetrics.recordsRead.toDouble)
+ val inputQuantiles = Input Size / Records +:
+ getFormattedSizeQuantilesWithRecords(inputSizes, inputRecords)
- val memoryBytesSpilledSizes = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.memoryBytesSpilled.toDouble
- }
- val memoryBytesSpilledQuantiles = Shuffle spill (memory) +:
- getFormattedSizeQuantiles(memoryBytesSpilledSizes)
+ val outputSizes = validTasks.map(_.taskMetrics.get.outputMetrics.bytesWritten.toDouble)
+ val outputRecords = validTasks.map(_.taskMetrics.get.outputMetrics.recordsWritten.toDouble)
+ val outputQuantiles = Output Size / Records +:
+ getFormattedSizeQuantilesWithRecords(outputSizes, outputRecords)
- val diskBytesSpilledSizes = validTasks.map { taskUIData: TaskUIData =>
- taskUIData.metrics.get.diskBytesSpilled.toDouble
- }
- val diskBytesSpilledQuantiles = Shuffle spill (disk) +:
- getFormattedSizeQuantiles(diskBytesSpilledSizes)
-
- val listings: Seq[Seq[Node]] = Seq(
- {serviceQuantiles} ,
- {schedulerDelayQuantiles} ,
-
- {deserializationQuantiles}
-
- {gcQuantiles} ,
-
- {serializationQuantiles}
- ,
- {gettingResultQuantiles} ,
-
- {peakExecutionMemoryQuantiles}
- ,
- if (stageData.hasInput) {inputQuantiles} else Nil,
- if (stageData.hasOutput) {outputQuantiles} else Nil,
- if (stageData.hasShuffleRead) {
-
- {shuffleReadBlockedQuantiles}
-
- {shuffleReadTotalQuantiles}
-
- {shuffleReadRemoteQuantiles}
-
- } else {
- Nil
- },
- if (stageData.hasShuffleWrite) {shuffleWriteQuantiles} else Nil,
- if (stageData.hasBytesSpilled) {memoryBytesSpilledQuantiles} else Nil,
- if (stageData.hasBytesSpilled) {diskBytesSpilledQuantiles} else Nil)
-
- val quantileHeaders = Seq("Metric", "Min", "25th percentile",
- "Median", "75th percentile", "Max")
- // The summary table does not use CSS to stripe rows, which doesn't work with hidden
- // rows (instead, JavaScript in table.js is used to stripe the non-hidden rows).
- Some(UIUtils.listingTable(
- quantileHeaders,
- identity[Seq[Node]],
- listings,
- fixedWidth = true,
- id = Some("task-summary-table"),
- stripeRowsWithCss = false))
+ val shuffleReadBlockedTimes = validTasks.map { task =>
+ task.taskMetrics.get.shuffleReadMetrics.fetchWaitTime.toDouble
+ }
+ val shuffleReadBlockedQuantiles =
+
+
+ Shuffle Read Blocked Time
+
+ +:
+ getFormattedTimeQuantiles(shuffleReadBlockedTimes)
+
+ val shuffleReadTotalSizes = validTasks.map { task =>
+ totalBytesRead(task.taskMetrics.get.shuffleReadMetrics).toDouble
+ }
+ val shuffleReadTotalRecords = validTasks.map { task =>
+ task.taskMetrics.get.shuffleReadMetrics.recordsRead.toDouble
+ }
+ val shuffleReadTotalQuantiles =
+
+
+ Shuffle Read Size / Records
+
+ +:
+ getFormattedSizeQuantilesWithRecords(shuffleReadTotalSizes, shuffleReadTotalRecords)
+
+ val shuffleReadRemoteSizes = validTasks.map { task =>
+ task.taskMetrics.get.shuffleReadMetrics.remoteBytesRead.toDouble
+ }
+ val shuffleReadRemoteQuantiles =
+
+
+ Shuffle Remote Reads
+
+ +:
+ getFormattedSizeQuantiles(shuffleReadRemoteSizes)
+
+ val shuffleWriteSizes = validTasks.map { task =>
+ task.taskMetrics.get.shuffleWriteMetrics.bytesWritten.toDouble
}
- val executorTable = new ExecutorTable(stageId, stageAttemptId, parent, store)
+ val shuffleWriteRecords = validTasks.map { task =>
+ task.taskMetrics.get.shuffleWriteMetrics.recordsWritten.toDouble
+ }
- val maybeAccumulableTable: Seq[Node] =
- if (hasAccumulators) { Accumulators ++ accumulableTable } else Seq.empty
+ val shuffleWriteQuantiles = Shuffle Write Size / Records +:
+ getFormattedSizeQuantilesWithRecords(shuffleWriteSizes, shuffleWriteRecords)
+
+ val memoryBytesSpilledSizes = validTasks.map(_.taskMetrics.get.memoryBytesSpilled.toDouble)
+ val memoryBytesSpilledQuantiles = Shuffle spill (memory) +:
+ getFormattedSizeQuantiles(memoryBytesSpilledSizes)
+
+ val diskBytesSpilledSizes = validTasks.map(_.taskMetrics.get.diskBytesSpilled.toDouble)
+ val diskBytesSpilledQuantiles = Shuffle spill (disk) +:
+ getFormattedSizeQuantiles(diskBytesSpilledSizes)
+
+ val listings: Seq[Seq[Node]] = Seq(
+ {serviceQuantiles} ,
+ {schedulerDelayQuantiles} ,
+
+ {deserializationQuantiles}
+
+ {gcQuantiles} ,
+
+ {serializationQuantiles}
+ ,
+ {gettingResultQuantiles} ,
+
+ {peakExecutionMemoryQuantiles}
+ ,
+ if (hasInput(stageData)) {inputQuantiles} else Nil,
+ if (hasOutput(stageData)) {outputQuantiles} else Nil,
+ if (hasShuffleRead(stageData)) {
+
+ {shuffleReadBlockedQuantiles}
+
+ {shuffleReadTotalQuantiles}
+
+ {shuffleReadRemoteQuantiles}
+
+ } else {
+ Nil
+ },
+ if (hasShuffleWrite(stageData)) {shuffleWriteQuantiles} else Nil,
+ if (hasBytesSpilled(stageData)) {memoryBytesSpilledQuantiles} else Nil,
+ if (hasBytesSpilled(stageData)) {diskBytesSpilledQuantiles} else Nil)
+
+ val quantileHeaders = Seq("Metric", "Min", "25th percentile",
+ "Median", "75th percentile", "Max")
+ // The summary table does not use CSS to stripe rows, which doesn't work with hidden
+ // rows (instead, JavaScript in table.js is used to stripe the non-hidden rows).
+ Some(UIUtils.listingTable(
+ quantileHeaders,
+ identity[Seq[Node]],
+ listings,
+ fixedWidth = true,
+ id = Some("task-summary-table"),
+ stripeRowsWithCss = false))
+ }
- val aggMetrics =
-
-
-
-
- {executorTable.toNodeSeq}
-
+ val executorTable = new ExecutorTable(stageData, parent.store)
+
+ val maybeAccumulableTable: Seq[Node] =
+ if (hasAccumulators) { Accumulators ++ accumulableTable } else Seq()
+
+ val aggMetrics =
+
+
+
+
+ {executorTable.toNodeSeq}
+
- val content =
- summary ++
- dagViz ++
- showAdditionalMetrics ++
- makeTimeline(
- // Only show the tasks in the table
- stageData.taskData.values.toSeq.filter(t => taskIdsInPage.contains(t.taskInfo.taskId)),
- currentTime) ++
- ++
- {summaryTable.getOrElse("No tasks have reported metrics yet.")}
++
- aggMetrics ++
- maybeAccumulableTable ++
- Tasks ({totalTasksNumStr}) ++
- taskTableHTML ++ jsForScrollingDownToTaskTable
- UIUtils.headerSparkPage(stageHeader, content, parent, showVisualization = true)
- }
+ val content =
+ summary ++
+ dagViz ++
+ showAdditionalMetrics ++
+ makeTimeline(
+ // Only show the tasks in the table
+ tasks.filter { t => taskIdsInPage.contains(t.taskId) },
+ currentTime) ++
+ ++
+ {summaryTable.getOrElse("No tasks have reported metrics yet.")}
++
+ aggMetrics ++
+ maybeAccumulableTable ++
+ Tasks ({totalTasksNumStr}) ++
+ taskTableHTML ++ jsForScrollingDownToTaskTable
+ UIUtils.headerSparkPage(stageHeader, content, parent, showVisualization = true)
}
- def makeTimeline(tasks: Seq[TaskUIData], currentTime: Long): Seq[Node] = {
+ def makeTimeline(tasks: Seq[TaskData], currentTime: Long): Seq[Node] = {
val executorsSet = new HashSet[(String, String)]
var minLaunchTime = Long.MaxValue
var maxFinishTime = Long.MinValue
val executorsArrayStr =
- tasks.sortBy(-_.taskInfo.launchTime).take(MAX_TIMELINE_TASKS).map { taskUIData =>
- val taskInfo = taskUIData.taskInfo
+ tasks.sortBy(-_.launchTime.getTime()).take(MAX_TIMELINE_TASKS).map { taskInfo =>
val executorId = taskInfo.executorId
val host = taskInfo.host
executorsSet += ((executorId, host))
- val launchTime = taskInfo.launchTime
- val finishTime = if (!taskInfo.running) taskInfo.finishTime else currentTime
+ val launchTime = taskInfo.launchTime.getTime()
+ val finishTime = taskInfo.duration.map(taskInfo.launchTime.getTime() + _)
+ .getOrElse(currentTime)
val totalExecutionTime = finishTime - launchTime
minLaunchTime = launchTime.min(minLaunchTime)
maxFinishTime = finishTime.max(maxFinishTime)
def toProportion(time: Long) = time.toDouble / totalExecutionTime * 100
- val metricsOpt = taskUIData.metrics
+ val metricsOpt = taskInfo.taskMetrics
val shuffleReadTime =
metricsOpt.map(_.shuffleReadMetrics.fetchWaitTime).getOrElse(0L)
val shuffleReadTimeProportion = toProportion(shuffleReadTime)
@@ -629,14 +594,14 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We
val serializationTimeProportion = toProportion(serializationTime)
val deserializationTime = metricsOpt.map(_.executorDeserializeTime).getOrElse(0L)
val deserializationTimeProportion = toProportion(deserializationTime)
- val gettingResultTime = getGettingResultTime(taskUIData.taskInfo, currentTime)
+ val gettingResultTime = getGettingResultTime(taskInfo, currentTime)
val gettingResultTimeProportion = toProportion(gettingResultTime)
val schedulerDelay =
metricsOpt.map(getSchedulerDelay(taskInfo, _, currentTime)).getOrElse(0L)
val schedulerDelayProportion = toProportion(schedulerDelay)
val executorOverhead = serializationTime + deserializationTime
- val executorRunTime = if (taskInfo.running) {
+ val executorRunTime = if (taskInfo.duration.isDefined) {
totalExecutionTime - executorOverhead - gettingResultTime
} else {
metricsOpt.map(_.executorRunTime).getOrElse(
@@ -663,7 +628,7 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We
serializationTimeProportionPos + serializationTimeProportion
val index = taskInfo.index
- val attempt = taskInfo.attemptNumber
+ val attempt = taskInfo.attempt
val svgTag =
if (totalExecutionTime == 0) {
@@ -705,7 +670,7 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We
|Status: ${taskInfo.status}{errorSummary}{details}
}
}
+
+private object ApiHelper {
+
+ def hasInput(stageData: StageData): Boolean = stageData.inputBytes > 0
+
+ def hasOutput(stageData: StageData): Boolean = stageData.outputBytes > 0
+
+ def hasShuffleRead(stageData: StageData): Boolean = stageData.shuffleReadBytes > 0
+
+ def hasShuffleWrite(stageData: StageData): Boolean = stageData.shuffleWriteBytes > 0
+
+ def hasBytesSpilled(stageData: StageData): Boolean = {
+ stageData.diskBytesSpilled > 0 || stageData.memoryBytesSpilled > 0
+ }
+
+ def totalBytesRead(metrics: ShuffleReadMetrics): Long = {
+ metrics.localBytesRead + metrics.remoteBytesRead
+ }
+
+}
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
index f0a12a28de06..18a4926f2f6c 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
@@ -26,19 +26,19 @@ import scala.xml._
import org.apache.commons.lang3.StringEscapeUtils
-import org.apache.spark.scheduler.StageInfo
+import org.apache.spark.status.AppStatusStore
+import org.apache.spark.status.api.v1
import org.apache.spark.ui._
-import org.apache.spark.ui.jobs.UIData.StageUIData
import org.apache.spark.util.Utils
private[ui] class StageTableBase(
+ store: AppStatusStore,
request: HttpServletRequest,
- stages: Seq[StageInfo],
+ stages: Seq[v1.StageData],
tableHeaderID: String,
stageTag: String,
basePath: String,
subPath: String,
- progressListener: JobProgressListener,
isFairScheduler: Boolean,
killEnabled: Boolean,
isFailedStage: Boolean) {
@@ -79,12 +79,12 @@ private[ui] class StageTableBase(
val toNodeSeq = try {
new StagePagedTable(
+ store,
stages,
tableHeaderID,
stageTag,
basePath,
subPath,
- progressListener,
isFairScheduler,
killEnabled,
currentTime,
@@ -106,13 +106,13 @@ private[ui] class StageTableBase(
}
private[ui] class StageTableRowData(
- val stageInfo: StageInfo,
- val stageData: Option[StageUIData],
+ val stage: v1.StageData,
+ val option: Option[v1.StageData],
val stageId: Int,
val attemptId: Int,
val schedulingPool: String,
val descriptionOption: Option[String],
- val submissionTime: Long,
+ val submissionTime: Date,
val formattedSubmissionTime: String,
val duration: Long,
val formattedDuration: String,
@@ -126,19 +126,20 @@ private[ui] class StageTableRowData(
val shuffleWriteWithUnit: String)
private[ui] class MissingStageTableRowData(
- stageInfo: StageInfo,
+ stageInfo: v1.StageData,
stageId: Int,
attemptId: Int) extends StageTableRowData(
- stageInfo, None, stageId, attemptId, "", None, 0, "", -1, "", 0, "", 0, "", 0, "", 0, "")
+ stageInfo, None, stageId, attemptId, "", None, new Date(0), "", -1, "", 0, "", 0, "", 0, "", 0,
+ "")
/** Page showing list of all ongoing and recently finished stages */
private[ui] class StagePagedTable(
- stages: Seq[StageInfo],
+ store: AppStatusStore,
+ stages: Seq[v1.StageData],
tableHeaderId: String,
stageTag: String,
basePath: String,
subPath: String,
- listener: JobProgressListener,
isFairScheduler: Boolean,
killEnabled: Boolean,
currentTime: Long,
@@ -164,8 +165,8 @@ private[ui] class StagePagedTable(
parameterOtherTable.mkString("&")
override val dataSource = new StageDataSource(
+ store,
stages,
- listener,
currentTime,
pageSize,
sortColumn,
@@ -274,10 +275,10 @@ private[ui] class StagePagedTable(
}
private def rowContent(data: StageTableRowData): Seq[Node] = {
- data.stageData match {
+ data.option match {
case None => missingStageRow(data.stageId)
case Some(stageData) =>
- val info = data.stageInfo
+ val info = data.stage
{if (data.attemptId > 0) {
{data.stageId} (retry {data.attemptId})
@@ -301,8 +302,8 @@ private[ui] class StagePagedTable(
{data.formattedDuration}
{UIUtils.makeProgressBar(started = stageData.numActiveTasks,
- completed = stageData.completedIndices.size, failed = stageData.numFailedTasks,
- skipped = 0, reasonToNumKilled = stageData.reasonToNumKilled, total = info.numTasks)}
+ completed = stageData.numCompleteTasks, failed = stageData.numFailedTasks,
+ skipped = 0, reasonToNumKilled = stageData.killedTasksSummary, total = info.numTasks)}
{data.inputReadWithUnit}
{data.outputWriteWithUnit}
@@ -318,7 +319,7 @@ private[ui] class StagePagedTable(
}
}
- private def failureReasonHtml(s: StageInfo): Seq[Node] = {
+ private def failureReasonHtml(s: v1.StageData): Seq[Node] = {
val failureReason = s.failureReason.getOrElse("")
val isMultiline = failureReason.indexOf('\n') >= 0
// Display the first line by default
@@ -344,7 +345,7 @@ private[ui] class StagePagedTable(
{failureReasonSummary}{details}
}
- private def makeDescription(s: StageInfo, descriptionOption: Option[String]): Seq[Node] = {
+ private def makeDescription(s: v1.StageData, descriptionOption: Option[String]): Seq[Node] = {
val basePathUri = UIUtils.prependBaseUri(basePath)
val killLink = if (killEnabled) {
@@ -368,8 +369,8 @@ private[ui] class StagePagedTable(
val nameLinkUri = s"$basePathUri/stages/stage?id=${s.stageId}&attempt=${s.attemptId}"
val nameLink = {s.name}
- val cachedRddInfos = s.rddInfos.filter(_.numCachedPartitions > 0)
- val details = if (s.details.nonEmpty) {
+ val cachedRddInfos = store.rddList().filter { rdd => s.rddIds.contains(rdd.id) }
+ val details = if (s.details != null && s.details.nonEmpty) {
+details
@@ -404,14 +405,14 @@ private[ui] class StagePagedTable(
}
private[ui] class StageDataSource(
- stages: Seq[StageInfo],
- listener: JobProgressListener,
+ store: AppStatusStore,
+ stages: Seq[v1.StageData],
currentTime: Long,
pageSize: Int,
sortColumn: String,
desc: Boolean) extends PagedDataSource[StageTableRowData](pageSize) {
- // Convert StageInfo to StageTableRowData which contains the final contents to show in the table
- // so that we can avoid creating duplicate contents during sorting the data
+ // Convert v1.StageData to StageTableRowData which contains the final contents to show in the
+ // table so that we can avoid creating duplicate contents during sorting the data
private val data = stages.map(stageRow).sorted(ordering(sortColumn, desc))
private var _slicedStageIds: Set[Int] = _
@@ -424,57 +425,46 @@ private[ui] class StageDataSource(
r
}
- private def stageRow(s: StageInfo): StageTableRowData = {
- val stageDataOption = listener.stageIdToData.get((s.stageId, s.attemptId))
+ private def stageRow(stageData: v1.StageData): StageTableRowData = {
+ val description = stageData.description.getOrElse("")
- if (stageDataOption.isEmpty) {
- return new MissingStageTableRowData(s, s.stageId, s.attemptId)
- }
- val stageData = stageDataOption.get
-
- val description = stageData.description
-
- val formattedSubmissionTime = s.submissionTime match {
- case Some(t) => UIUtils.formatDate(new Date(t))
+ val formattedSubmissionTime = stageData.submissionTime match {
+ case Some(t) => UIUtils.formatDate(t)
case None => "Unknown"
}
- val finishTime = s.completionTime.getOrElse(currentTime)
+ val finishTime = stageData.completionTime.map(_.getTime()).getOrElse(currentTime)
// The submission time for a stage is misleading because it counts the time
// the stage waits to be launched. (SPARK-10930)
- val taskLaunchTimes =
- stageData.taskData.values.map(_.taskInfo.launchTime).filter(_ > 0)
- val duration: Option[Long] =
- if (taskLaunchTimes.nonEmpty) {
- val startTime = taskLaunchTimes.min
- if (finishTime > startTime) {
- Some(finishTime - startTime)
- } else {
- Some(currentTime - startTime)
- }
+ val duration = stageData.firstTaskLaunchedTime.map { date =>
+ val time = date.getTime()
+ if (finishTime > time) {
+ finishTime - time
} else {
None
+ currentTime - time
}
+ }
val formattedDuration = duration.map(d => UIUtils.formatDuration(d)).getOrElse("Unknown")
val inputRead = stageData.inputBytes
val inputReadWithUnit = if (inputRead > 0) Utils.bytesToString(inputRead) else ""
val outputWrite = stageData.outputBytes
val outputWriteWithUnit = if (outputWrite > 0) Utils.bytesToString(outputWrite) else ""
- val shuffleRead = stageData.shuffleReadTotalBytes
+ val shuffleRead = stageData.shuffleReadBytes
val shuffleReadWithUnit = if (shuffleRead > 0) Utils.bytesToString(shuffleRead) else ""
val shuffleWrite = stageData.shuffleWriteBytes
val shuffleWriteWithUnit = if (shuffleWrite > 0) Utils.bytesToString(shuffleWrite) else ""
new StageTableRowData(
- s,
- stageDataOption,
- s.stageId,
- s.attemptId,
+ stageData,
+ Some(stageData),
+ stageData.stageId,
+ stageData.attemptId,
stageData.schedulingPool,
- description,
- s.submissionTime.getOrElse(0),
+ stageData.description,
+ stageData.submissionTime.getOrElse(new Date(0)),
formattedSubmissionTime,
duration.getOrElse(-1),
formattedDuration,
@@ -496,7 +486,7 @@ private[ui] class StageDataSource(
val ordering: Ordering[StageTableRowData] = sortColumn match {
case "Stage Id" => Ordering.by(_.stageId)
case "Pool Name" => Ordering.by(_.schedulingPool)
- case "Description" => Ordering.by(x => (x.descriptionOption, x.stageInfo.name))
+ case "Description" => Ordering.by(x => (x.descriptionOption, x.stage.name))
case "Submitted" => Ordering.by(_.submissionTime)
case "Duration" => Ordering.by(_.duration)
case "Input" => Ordering.by(_.inputRead)
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StagesTab.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StagesTab.scala
index 65446f967ad7..be05a963f0e6 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/StagesTab.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/StagesTab.scala
@@ -21,36 +21,42 @@ import javax.servlet.http.HttpServletRequest
import org.apache.spark.scheduler.SchedulingMode
import org.apache.spark.status.AppStatusStore
+import org.apache.spark.status.api.v1.StageStatus
import org.apache.spark.ui.{SparkUI, SparkUITab, UIUtils}
/** Web UI showing progress status of all stages in the given SparkContext. */
-private[ui] class StagesTab(val parent: SparkUI, store: AppStatusStore)
+private[ui] class StagesTab(val parent: SparkUI, val store: AppStatusStore)
extends SparkUITab(parent, "stages") {
val sc = parent.sc
val conf = parent.conf
val killEnabled = parent.killEnabled
- val progressListener = parent.jobProgressListener
- val operationGraphListener = parent.operationGraphListener
- val lastUpdateTime = parent.lastUpdateTime
attachPage(new AllStagesPage(this))
attachPage(new StagePage(this, store))
attachPage(new PoolPage(this))
- def isFairScheduler: Boolean = progressListener.schedulingMode == Some(SchedulingMode.FAIR)
+ def isFairScheduler: Boolean = {
+ store.environmentInfo().sparkProperties.toMap
+ .get("spark.scheduler.mode")
+ .map { mode => mode == SchedulingMode.FAIR }
+ .getOrElse(false)
+ }
def handleKillRequest(request: HttpServletRequest): Unit = {
if (killEnabled && parent.securityManager.checkModifyPermissions(request.getRemoteUser)) {
// stripXSS is called first to remove suspicious characters used in XSS attacks
val stageId = Option(UIUtils.stripXSS(request.getParameter("id"))).map(_.toInt)
stageId.foreach { id =>
- if (progressListener.activeStages.contains(id)) {
- sc.foreach(_.cancelStage(id, "killed via the Web UI"))
- // Do a quick pause here to give Spark time to kill the stage so it shows up as
- // killed after the refresh. Note that this will block the serving thread so the
- // time should be limited in duration.
- Thread.sleep(100)
+ store.asOption(store.lastStageAttempt(id)).foreach { stage =>
+ val status = stage.status
+ if (status == StageStatus.ACTIVE || status == StageStatus.PENDING) {
+ sc.foreach(_.cancelStage(id, "killed via the Web UI"))
+ // Do a quick pause here to give Spark time to kill the stage so it shows up as
+ // killed after the refresh. Note that this will block the serving thread so the
+ // time should be limited in duration.
+ Thread.sleep(100)
+ }
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/ui/scope/RDDOperationGraph.scala b/core/src/main/scala/org/apache/spark/ui/scope/RDDOperationGraph.scala
index bb763248cd7e..827a8637b9bd 100644
--- a/core/src/main/scala/org/apache/spark/ui/scope/RDDOperationGraph.scala
+++ b/core/src/main/scala/org/apache/spark/ui/scope/RDDOperationGraph.scala
@@ -35,20 +35,20 @@ import org.apache.spark.storage.StorageLevel
* nodes and children clusters. Additionally, a graph may also have edges that enter or exit
* the graph from nodes that belong to adjacent graphs.
*/
-private[ui] case class RDDOperationGraph(
+private[spark] case class RDDOperationGraph(
edges: Seq[RDDOperationEdge],
outgoingEdges: Seq[RDDOperationEdge],
incomingEdges: Seq[RDDOperationEdge],
rootCluster: RDDOperationCluster)
/** A node in an RDDOperationGraph. This represents an RDD. */
-private[ui] case class RDDOperationNode(id: Int, name: String, cached: Boolean, callsite: String)
+private[spark] case class RDDOperationNode(id: Int, name: String, cached: Boolean, callsite: String)
/**
* A directed edge connecting two nodes in an RDDOperationGraph.
* This represents an RDD dependency.
*/
-private[ui] case class RDDOperationEdge(fromId: Int, toId: Int)
+private[spark] case class RDDOperationEdge(fromId: Int, toId: Int)
/**
* A cluster that groups nodes together in an RDDOperationGraph.
@@ -56,7 +56,7 @@ private[ui] case class RDDOperationEdge(fromId: Int, toId: Int)
* This represents any grouping of RDDs, including operation scopes (e.g. textFile, flatMap),
* stages, jobs, or any higher level construct. A cluster may be nested inside of other clusters.
*/
-private[ui] class RDDOperationCluster(val id: String, private var _name: String) {
+private[spark] class RDDOperationCluster(val id: String, private var _name: String) {
private val _childNodes = new ListBuffer[RDDOperationNode]
private val _childClusters = new ListBuffer[RDDOperationCluster]
@@ -92,7 +92,7 @@ private[ui] class RDDOperationCluster(val id: String, private var _name: String)
}
}
-private[ui] object RDDOperationGraph extends Logging {
+private[spark] object RDDOperationGraph extends Logging {
val STAGE_CLUSTER_PREFIX = "stage_"
diff --git a/core/src/main/scala/org/apache/spark/ui/scope/RDDOperationGraphListener.scala b/core/src/main/scala/org/apache/spark/ui/scope/RDDOperationGraphListener.scala
deleted file mode 100644
index 37a12a864693..000000000000
--- a/core/src/main/scala/org/apache/spark/ui/scope/RDDOperationGraphListener.scala
+++ /dev/null
@@ -1,150 +0,0 @@
-/*
- * 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.ui.scope
-
-import scala.collection.mutable
-
-import org.apache.spark.SparkConf
-import org.apache.spark.scheduler._
-import org.apache.spark.ui.SparkUI
-
-/**
- * A SparkListener that constructs a DAG of RDD operations.
- */
-private[ui] class RDDOperationGraphListener(conf: SparkConf) extends SparkListener {
-
- // Note: the fate of jobs and stages are tied. This means when we clean up a job,
- // we always clean up all of its stages. Similarly, when we clean up a stage, we
- // always clean up its job (and, transitively, other stages in the same job).
- private[ui] val jobIdToStageIds = new mutable.HashMap[Int, Seq[Int]]
- private[ui] val jobIdToSkippedStageIds = new mutable.HashMap[Int, Seq[Int]]
- private[ui] val stageIdToJobId = new mutable.HashMap[Int, Int]
- private[ui] val stageIdToGraph = new mutable.HashMap[Int, RDDOperationGraph]
- private[ui] val completedStageIds = new mutable.HashSet[Int]
-
- // Keep track of the order in which these are inserted so we can remove old ones
- private[ui] val jobIds = new mutable.ArrayBuffer[Int]
- private[ui] val stageIds = new mutable.ArrayBuffer[Int]
-
- // How many root nodes to retain in DAG Graph
- private[ui] val retainedNodes =
- conf.getInt("spark.ui.dagGraph.retainedRootRDDs", Int.MaxValue)
-
- // How many jobs or stages to retain graph metadata for
- private val retainedJobs =
- conf.getInt("spark.ui.retainedJobs", SparkUI.DEFAULT_RETAINED_JOBS)
- private val retainedStages =
- conf.getInt("spark.ui.retainedStages", SparkUI.DEFAULT_RETAINED_STAGES)
-
- /**
- * Return the graph metadata for all stages in the given job.
- * An empty list is returned if one or more of its stages has been cleaned up.
- */
- def getOperationGraphForJob(jobId: Int): Seq[RDDOperationGraph] = synchronized {
- val skippedStageIds = jobIdToSkippedStageIds.getOrElse(jobId, Seq.empty)
- val graphs = jobIdToStageIds.getOrElse(jobId, Seq.empty)
- .flatMap { sid => stageIdToGraph.get(sid) }
- // Mark any skipped stages as such
- graphs.foreach { g =>
- val stageId = g.rootCluster.id.replaceAll(RDDOperationGraph.STAGE_CLUSTER_PREFIX, "").toInt
- if (skippedStageIds.contains(stageId) && !g.rootCluster.name.contains("skipped")) {
- g.rootCluster.setName(g.rootCluster.name + " (skipped)")
- }
- }
- graphs
- }
-
- /** Return the graph metadata for the given stage, or None if no such information exists. */
- def getOperationGraphForStage(stageId: Int): Option[RDDOperationGraph] = synchronized {
- stageIdToGraph.get(stageId)
- }
-
- /** On job start, construct a RDDOperationGraph for each stage in the job for display later. */
- override def onJobStart(jobStart: SparkListenerJobStart): Unit = synchronized {
- val jobId = jobStart.jobId
- val stageInfos = jobStart.stageInfos
-
- jobIds += jobId
- jobIdToStageIds(jobId) = jobStart.stageInfos.map(_.stageId).sorted
-
- stageInfos.foreach { stageInfo =>
- val stageId = stageInfo.stageId
- stageIds += stageId
- stageIdToJobId(stageId) = jobId
- stageIdToGraph(stageId) = RDDOperationGraph.makeOperationGraph(stageInfo, retainedNodes)
- trimStagesIfNecessary()
- }
-
- trimJobsIfNecessary()
- }
-
- /** Keep track of stages that have completed. */
- override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = synchronized {
- val stageId = stageCompleted.stageInfo.stageId
- if (stageIdToJobId.contains(stageId)) {
- // Note: Only do this if the stage has not already been cleaned up
- // Otherwise, we may never clean this stage from `completedStageIds`
- completedStageIds += stageCompleted.stageInfo.stageId
- }
- }
-
- /** On job end, find all stages in this job that are skipped and mark them as such. */
- override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit = synchronized {
- val jobId = jobEnd.jobId
- jobIdToStageIds.get(jobId).foreach { stageIds =>
- val skippedStageIds = stageIds.filter { sid => !completedStageIds.contains(sid) }
- // Note: Only do this if the job has not already been cleaned up
- // Otherwise, we may never clean this job from `jobIdToSkippedStageIds`
- jobIdToSkippedStageIds(jobId) = skippedStageIds
- }
- }
-
- /** Clean metadata for old stages if we have exceeded the number to retain. */
- private def trimStagesIfNecessary(): Unit = {
- if (stageIds.size >= retainedStages) {
- val toRemove = math.max(retainedStages / 10, 1)
- stageIds.take(toRemove).foreach { id => cleanStage(id) }
- stageIds.trimStart(toRemove)
- }
- }
-
- /** Clean metadata for old jobs if we have exceeded the number to retain. */
- private def trimJobsIfNecessary(): Unit = {
- if (jobIds.size >= retainedJobs) {
- val toRemove = math.max(retainedJobs / 10, 1)
- jobIds.take(toRemove).foreach { id => cleanJob(id) }
- jobIds.trimStart(toRemove)
- }
- }
-
- /** Clean metadata for the given stage, its job, and all other stages that belong to the job. */
- private[ui] def cleanStage(stageId: Int): Unit = {
- completedStageIds.remove(stageId)
- stageIdToGraph.remove(stageId)
- stageIdToJobId.remove(stageId).foreach { jobId => cleanJob(jobId) }
- }
-
- /** Clean metadata for the given job and all stages that belong to it. */
- private[ui] def cleanJob(jobId: Int): Unit = {
- jobIdToSkippedStageIds.remove(jobId)
- jobIdToStageIds.remove(jobId).foreach { stageIds =>
- stageIds.foreach { stageId => cleanStage(stageId) }
- }
- }
-
-}
diff --git a/core/src/test/resources/HistoryServerExpectations/complete_stage_list_json_expectation.json b/core/src/test/resources/HistoryServerExpectations/complete_stage_list_json_expectation.json
index 25c4fff77e0a..37b7d7269059 100644
--- a/core/src/test/resources/HistoryServerExpectations/complete_stage_list_json_expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/complete_stage_list_json_expectation.json
@@ -2,9 +2,12 @@
"status" : "COMPLETE",
"stageId" : 3,
"attemptId" : 0,
+ "numTasks" : 8,
"numActiveTasks" : 0,
"numCompleteTasks" : 8,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"executorRunTime" : 162,
"executorCpuTime" : 0,
"submissionTime" : "2015-02-03T16:43:07.191GMT",
@@ -23,14 +26,19 @@
"name" : "count at :17",
"details" : "org.apache.spark.rdd.RDD.count(RDD.scala:910)\n$line19.$read$$iwC$$iwC$$iwC$$iwC.(:17)\n$line19.$read$$iwC$$iwC$$iwC.(:22)\n$line19.$read$$iwC$$iwC.(:24)\n$line19.$read$$iwC.(:26)\n$line19.$read.(:28)\n$line19.$read$.(:32)\n$line19.$read$.()\n$line19.$eval$.(:7)\n$line19.$eval$.()\n$line19.$eval.$print()\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:606)\norg.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:852)\norg.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1125)\norg.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:674)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:705)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:669)",
"schedulingPool" : "default",
- "accumulatorUpdates" : [ ]
+ "rddIds" : [ 6, 5 ],
+ "accumulatorUpdates" : [ ],
+ "killedTasksSummary" : { }
}, {
"status" : "COMPLETE",
"stageId" : 1,
"attemptId" : 0,
+ "numTasks" : 8,
"numActiveTasks" : 0,
"numCompleteTasks" : 8,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"executorRunTime" : 3476,
"executorCpuTime" : 0,
"submissionTime" : "2015-02-03T16:43:05.829GMT",
@@ -49,14 +57,19 @@
"name" : "map at :14",
"details" : "org.apache.spark.rdd.RDD.map(RDD.scala:271)\n$line10.$read$$iwC$$iwC$$iwC$$iwC.(:14)\n$line10.$read$$iwC$$iwC$$iwC.(:19)\n$line10.$read$$iwC$$iwC.(:21)\n$line10.$read$$iwC.(:23)\n$line10.$read.(:25)\n$line10.$read$.(:29)\n$line10.$read$.()\n$line10.$eval$.(:7)\n$line10.$eval$.()\n$line10.$eval.$print()\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:606)\norg.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:852)\norg.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1125)\norg.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:674)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:705)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:669)",
"schedulingPool" : "default",
- "accumulatorUpdates" : [ ]
+ "rddIds" : [ 1, 0 ],
+ "accumulatorUpdates" : [ ],
+ "killedTasksSummary" : { }
}, {
"status" : "COMPLETE",
"stageId" : 0,
"attemptId" : 0,
+ "numTasks" : 8,
"numActiveTasks" : 0,
"numCompleteTasks" : 8,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"executorRunTime" : 4338,
"executorCpuTime" : 0,
"submissionTime" : "2015-02-03T16:43:04.228GMT",
@@ -75,5 +88,7 @@
"name" : "count at :15",
"details" : "org.apache.spark.rdd.RDD.count(RDD.scala:910)\n$line9.$read$$iwC$$iwC$$iwC$$iwC.(:15)\n$line9.$read$$iwC$$iwC$$iwC.(:20)\n$line9.$read$$iwC$$iwC.(:22)\n$line9.$read$$iwC.(:24)\n$line9.$read.(:26)\n$line9.$read$.(:30)\n$line9.$read$.()\n$line9.$eval$.(:7)\n$line9.$eval$.()\n$line9.$eval.$print()\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:606)\norg.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:852)\norg.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1125)\norg.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:674)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:705)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:669)",
"schedulingPool" : "default",
- "accumulatorUpdates" : [ ]
+ "rddIds" : [ 0 ],
+ "accumulatorUpdates" : [ ],
+ "killedTasksSummary" : { }
} ]
diff --git a/core/src/test/resources/HistoryServerExpectations/failed_stage_list_json_expectation.json b/core/src/test/resources/HistoryServerExpectations/failed_stage_list_json_expectation.json
index b86ba1e65de1..2fd55666fa01 100644
--- a/core/src/test/resources/HistoryServerExpectations/failed_stage_list_json_expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/failed_stage_list_json_expectation.json
@@ -2,14 +2,18 @@
"status" : "FAILED",
"stageId" : 2,
"attemptId" : 0,
+ "numTasks" : 8,
"numActiveTasks" : 0,
"numCompleteTasks" : 7,
"numFailedTasks" : 1,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 7,
"executorRunTime" : 278,
"executorCpuTime" : 0,
"submissionTime" : "2015-02-03T16:43:06.296GMT",
"firstTaskLaunchedTime" : "2015-02-03T16:43:06.296GMT",
"completionTime" : "2015-02-03T16:43:06.347GMT",
+ "failureReason" : "Job aborted due to stage failure: Task 3 in stage 2.0 failed 1 times, most recent failure: Lost task 3.0 in stage 2.0 (TID 19, localhost): java.lang.RuntimeException: got a 3, failing\n\tat $line11.$read$$iwC$$iwC$$iwC$$iwC$$anonfun$1.apply(:18)\n\tat $line11.$read$$iwC$$iwC$$iwC$$iwC$$anonfun$1.apply(:17)\n\tat scala.collection.Iterator$$anon$11.next(Iterator.scala:328)\n\tat org.apache.spark.util.Utils$.getIteratorSize(Utils.scala:1311)\n\tat org.apache.spark.rdd.RDD$$anonfun$count$1.apply(RDD.scala:910)\n\tat org.apache.spark.rdd.RDD$$anonfun$count$1.apply(RDD.scala:910)\n\tat org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1314)\n\tat org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1314)\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:56)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n\tat java.lang.Thread.run(Thread.java:745)\n\nDriver stacktrace:",
"inputBytes" : 0,
"inputRecords" : 0,
"outputBytes" : 0,
@@ -23,5 +27,7 @@
"name" : "count at :20",
"details" : "org.apache.spark.rdd.RDD.count(RDD.scala:910)\n$line11.$read$$iwC$$iwC$$iwC$$iwC.(:20)\n$line11.$read$$iwC$$iwC$$iwC.(:25)\n$line11.$read$$iwC$$iwC.(:27)\n$line11.$read$$iwC.(:29)\n$line11.$read.(:31)\n$line11.$read$.(:35)\n$line11.$read$.()\n$line11.$eval$.(:7)\n$line11.$eval$.()\n$line11.$eval.$print()\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:606)\norg.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:852)\norg.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1125)\norg.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:674)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:705)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:669)",
"schedulingPool" : "default",
- "accumulatorUpdates" : [ ]
+ "rddIds" : [ 3, 2 ],
+ "accumulatorUpdates" : [ ],
+ "killedTasksSummary" : { }
} ]
diff --git a/core/src/test/resources/HistoryServerExpectations/job_list_from_multi_attempt_app_json_1__expectation.json b/core/src/test/resources/HistoryServerExpectations/job_list_from_multi_attempt_app_json_1__expectation.json
index c108fa61a431..2f275c7bfe2f 100644
--- a/core/src/test/resources/HistoryServerExpectations/job_list_from_multi_attempt_app_json_1__expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/job_list_from_multi_attempt_app_json_1__expectation.json
@@ -8,8 +8,11 @@
"numCompletedTasks" : 8,
"numSkippedTasks" : 0,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"numActiveStages" : 0,
"numCompletedStages" : 1,
"numSkippedStages" : 0,
- "numFailedStages" : 0
+ "numFailedStages" : 0,
+ "killedTasksSummary" : { }
} ]
diff --git a/core/src/test/resources/HistoryServerExpectations/job_list_from_multi_attempt_app_json_2__expectation.json b/core/src/test/resources/HistoryServerExpectations/job_list_from_multi_attempt_app_json_2__expectation.json
index c108fa61a431..2f275c7bfe2f 100644
--- a/core/src/test/resources/HistoryServerExpectations/job_list_from_multi_attempt_app_json_2__expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/job_list_from_multi_attempt_app_json_2__expectation.json
@@ -8,8 +8,11 @@
"numCompletedTasks" : 8,
"numSkippedTasks" : 0,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"numActiveStages" : 0,
"numCompletedStages" : 1,
"numSkippedStages" : 0,
- "numFailedStages" : 0
+ "numFailedStages" : 0,
+ "killedTasksSummary" : { }
} ]
diff --git a/core/src/test/resources/HistoryServerExpectations/job_list_json_expectation.json b/core/src/test/resources/HistoryServerExpectations/job_list_json_expectation.json
index 3d7407004d26..71bf8706307c 100644
--- a/core/src/test/resources/HistoryServerExpectations/job_list_json_expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/job_list_json_expectation.json
@@ -8,10 +8,13 @@
"numCompletedTasks" : 8,
"numSkippedTasks" : 0,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"numActiveStages" : 0,
"numCompletedStages" : 1,
"numSkippedStages" : 0,
- "numFailedStages" : 0
+ "numFailedStages" : 0,
+ "killedTasksSummary" : { }
}, {
"jobId" : 1,
"name" : "count at :20",
@@ -22,10 +25,13 @@
"numCompletedTasks" : 15,
"numSkippedTasks" : 0,
"numFailedTasks" : 1,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 15,
"numActiveStages" : 0,
"numCompletedStages" : 1,
"numSkippedStages" : 0,
- "numFailedStages" : 1
+ "numFailedStages" : 1,
+ "killedTasksSummary" : { }
}, {
"jobId" : 0,
"name" : "count at :15",
@@ -36,8 +42,11 @@
"numCompletedTasks" : 8,
"numSkippedTasks" : 0,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"numActiveStages" : 0,
"numCompletedStages" : 1,
"numSkippedStages" : 0,
- "numFailedStages" : 0
+ "numFailedStages" : 0,
+ "killedTasksSummary" : { }
} ]
diff --git a/core/src/test/resources/HistoryServerExpectations/one_job_json_expectation.json b/core/src/test/resources/HistoryServerExpectations/one_job_json_expectation.json
index 10c7e1c0b36f..1eae5f3d5beb 100644
--- a/core/src/test/resources/HistoryServerExpectations/one_job_json_expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/one_job_json_expectation.json
@@ -8,8 +8,11 @@
"numCompletedTasks" : 8,
"numSkippedTasks" : 0,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"numActiveStages" : 0,
"numCompletedStages" : 1,
"numSkippedStages" : 0,
- "numFailedStages" : 0
+ "numFailedStages" : 0,
+ "killedTasksSummary" : { }
}
diff --git a/core/src/test/resources/HistoryServerExpectations/one_stage_attempt_json_expectation.json b/core/src/test/resources/HistoryServerExpectations/one_stage_attempt_json_expectation.json
index 6fb40f6f1713..31093a661663 100644
--- a/core/src/test/resources/HistoryServerExpectations/one_stage_attempt_json_expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/one_stage_attempt_json_expectation.json
@@ -2,9 +2,12 @@
"status" : "COMPLETE",
"stageId" : 1,
"attemptId" : 0,
+ "numTasks" : 8,
"numActiveTasks" : 0,
"numCompleteTasks" : 8,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"executorRunTime" : 3476,
"executorCpuTime" : 0,
"submissionTime" : "2015-02-03T16:43:05.829GMT",
@@ -23,14 +26,15 @@
"name" : "map at :14",
"details" : "org.apache.spark.rdd.RDD.map(RDD.scala:271)\n$line10.$read$$iwC$$iwC$$iwC$$iwC.(:14)\n$line10.$read$$iwC$$iwC$$iwC.(:19)\n$line10.$read$$iwC$$iwC.(:21)\n$line10.$read$$iwC.(:23)\n$line10.$read.(:25)\n$line10.$read$.(:29)\n$line10.$read$.()\n$line10.$eval$.(:7)\n$line10.$eval$.()\n$line10.$eval.$print()\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:606)\norg.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:852)\norg.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1125)\norg.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:674)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:705)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:669)",
"schedulingPool" : "default",
+ "rddIds" : [ 1, 0 ],
"accumulatorUpdates" : [ ],
"tasks" : {
- "8" : {
- "taskId" : 8,
- "index" : 0,
+ "10" : {
+ "taskId" : 10,
+ "index" : 2,
"attempt" : 0,
- "launchTime" : "2015-02-03T16:43:05.829GMT",
- "duration" : 435,
+ "launchTime" : "2015-02-03T16:43:05.830GMT",
+ "duration" : 456,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -38,15 +42,16 @@
"speculative" : false,
"accumulatorUpdates" : [ ],
"taskMetrics" : {
- "executorDeserializeTime" : 1,
+ "executorDeserializeTime" : 2,
"executorDeserializeCpuTime" : 0,
- "executorRunTime" : 435,
+ "executorRunTime" : 434,
"executorCpuTime" : 0,
"resultSize" : 1902,
"jvmGcTime" : 19,
- "resultSerializationTime" : 2,
+ "resultSerializationTime" : 1,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -66,17 +71,17 @@
},
"shuffleWriteMetrics" : {
"bytesWritten" : 1648,
- "writeTime" : 94000,
+ "writeTime" : 76000,
"recordsWritten" : 0
}
}
},
- "9" : {
- "taskId" : 9,
- "index" : 1,
+ "14" : {
+ "taskId" : 14,
+ "index" : 6,
"attempt" : 0,
- "launchTime" : "2015-02-03T16:43:05.830GMT",
- "duration" : 436,
+ "launchTime" : "2015-02-03T16:43:05.832GMT",
+ "duration" : 450,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -84,15 +89,16 @@
"speculative" : false,
"accumulatorUpdates" : [ ],
"taskMetrics" : {
- "executorDeserializeTime" : 1,
+ "executorDeserializeTime" : 2,
"executorDeserializeCpuTime" : 0,
- "executorRunTime" : 436,
+ "executorRunTime" : 434,
"executorCpuTime" : 0,
"resultSize" : 1902,
"jvmGcTime" : 19,
- "resultSerializationTime" : 0,
+ "resultSerializationTime" : 1,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -112,17 +118,17 @@
},
"shuffleWriteMetrics" : {
"bytesWritten" : 1648,
- "writeTime" : 98000,
+ "writeTime" : 88000,
"recordsWritten" : 0
}
}
},
- "10" : {
- "taskId" : 10,
- "index" : 2,
+ "9" : {
+ "taskId" : 9,
+ "index" : 1,
"attempt" : 0,
"launchTime" : "2015-02-03T16:43:05.830GMT",
- "duration" : 434,
+ "duration" : 454,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -130,15 +136,16 @@
"speculative" : false,
"accumulatorUpdates" : [ ],
"taskMetrics" : {
- "executorDeserializeTime" : 2,
+ "executorDeserializeTime" : 1,
"executorDeserializeCpuTime" : 0,
- "executorRunTime" : 434,
+ "executorRunTime" : 436,
"executorCpuTime" : 0,
"resultSize" : 1902,
"jvmGcTime" : 19,
- "resultSerializationTime" : 1,
+ "resultSerializationTime" : 0,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -158,17 +165,17 @@
},
"shuffleWriteMetrics" : {
"bytesWritten" : 1648,
- "writeTime" : 76000,
+ "writeTime" : 98000,
"recordsWritten" : 0
}
}
},
- "11" : {
- "taskId" : 11,
- "index" : 3,
+ "13" : {
+ "taskId" : 13,
+ "index" : 5,
"attempt" : 0,
- "launchTime" : "2015-02-03T16:43:05.830GMT",
- "duration" : 434,
+ "launchTime" : "2015-02-03T16:43:05.831GMT",
+ "duration" : 452,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -182,9 +189,10 @@
"executorCpuTime" : 0,
"resultSize" : 1902,
"jvmGcTime" : 19,
- "resultSerializationTime" : 1,
+ "resultSerializationTime" : 2,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -203,8 +211,8 @@
"recordsRead" : 0
},
"shuffleWriteMetrics" : {
- "bytesWritten" : 1647,
- "writeTime" : 83000,
+ "bytesWritten" : 1648,
+ "writeTime" : 73000,
"recordsWritten" : 0
}
}
@@ -214,7 +222,7 @@
"index" : 4,
"attempt" : 0,
"launchTime" : "2015-02-03T16:43:05.831GMT",
- "duration" : 434,
+ "duration" : 454,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -231,6 +239,7 @@
"resultSerializationTime" : 1,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -255,12 +264,12 @@
}
}
},
- "13" : {
- "taskId" : 13,
- "index" : 5,
+ "11" : {
+ "taskId" : 11,
+ "index" : 3,
"attempt" : 0,
- "launchTime" : "2015-02-03T16:43:05.831GMT",
- "duration" : 434,
+ "launchTime" : "2015-02-03T16:43:05.830GMT",
+ "duration" : 454,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -274,9 +283,10 @@
"executorCpuTime" : 0,
"resultSize" : 1902,
"jvmGcTime" : 19,
- "resultSerializationTime" : 2,
+ "resultSerializationTime" : 1,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -295,18 +305,18 @@
"recordsRead" : 0
},
"shuffleWriteMetrics" : {
- "bytesWritten" : 1648,
- "writeTime" : 73000,
+ "bytesWritten" : 1647,
+ "writeTime" : 83000,
"recordsWritten" : 0
}
}
},
- "14" : {
- "taskId" : 14,
- "index" : 6,
+ "8" : {
+ "taskId" : 8,
+ "index" : 0,
"attempt" : 0,
- "launchTime" : "2015-02-03T16:43:05.832GMT",
- "duration" : 434,
+ "launchTime" : "2015-02-03T16:43:05.829GMT",
+ "duration" : 454,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -314,15 +324,16 @@
"speculative" : false,
"accumulatorUpdates" : [ ],
"taskMetrics" : {
- "executorDeserializeTime" : 2,
+ "executorDeserializeTime" : 1,
"executorDeserializeCpuTime" : 0,
- "executorRunTime" : 434,
+ "executorRunTime" : 435,
"executorCpuTime" : 0,
"resultSize" : 1902,
"jvmGcTime" : 19,
- "resultSerializationTime" : 1,
+ "resultSerializationTime" : 2,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -342,7 +353,7 @@
},
"shuffleWriteMetrics" : {
"bytesWritten" : 1648,
- "writeTime" : 88000,
+ "writeTime" : 94000,
"recordsWritten" : 0
}
}
@@ -352,7 +363,7 @@
"index" : 7,
"attempt" : 0,
"launchTime" : "2015-02-03T16:43:05.833GMT",
- "duration" : 435,
+ "duration" : 450,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -369,6 +380,7 @@
"resultSerializationTime" : 1,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -399,12 +411,18 @@
"taskTime" : 3624,
"failedTasks" : 0,
"succeededTasks" : 8,
+ "killedTasks" : 0,
"inputBytes" : 28000128,
+ "inputRecords" : 0,
"outputBytes" : 0,
+ "outputRecords" : 0,
"shuffleRead" : 0,
+ "shuffleReadRecords" : 0,
"shuffleWrite" : 13180,
+ "shuffleWriteRecords" : 0,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0
}
- }
+ },
+ "killedTasksSummary" : { }
}
diff --git a/core/src/test/resources/HistoryServerExpectations/one_stage_json_expectation.json b/core/src/test/resources/HistoryServerExpectations/one_stage_json_expectation.json
index f5a89a210764..601d70695b17 100644
--- a/core/src/test/resources/HistoryServerExpectations/one_stage_json_expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/one_stage_json_expectation.json
@@ -2,9 +2,12 @@
"status" : "COMPLETE",
"stageId" : 1,
"attemptId" : 0,
+ "numTasks" : 8,
"numActiveTasks" : 0,
"numCompleteTasks" : 8,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"executorRunTime" : 3476,
"executorCpuTime" : 0,
"submissionTime" : "2015-02-03T16:43:05.829GMT",
@@ -23,14 +26,15 @@
"name" : "map at :14",
"details" : "org.apache.spark.rdd.RDD.map(RDD.scala:271)\n$line10.$read$$iwC$$iwC$$iwC$$iwC.(:14)\n$line10.$read$$iwC$$iwC$$iwC.(:19)\n$line10.$read$$iwC$$iwC.(:21)\n$line10.$read$$iwC.(:23)\n$line10.$read.(:25)\n$line10.$read$.(:29)\n$line10.$read$.()\n$line10.$eval$.(:7)\n$line10.$eval$.()\n$line10.$eval.$print()\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:606)\norg.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:852)\norg.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1125)\norg.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:674)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:705)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:669)",
"schedulingPool" : "default",
+ "rddIds" : [ 1, 0 ],
"accumulatorUpdates" : [ ],
"tasks" : {
- "8" : {
- "taskId" : 8,
- "index" : 0,
+ "10" : {
+ "taskId" : 10,
+ "index" : 2,
"attempt" : 0,
- "launchTime" : "2015-02-03T16:43:05.829GMT",
- "duration" : 435,
+ "launchTime" : "2015-02-03T16:43:05.830GMT",
+ "duration" : 456,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -38,15 +42,16 @@
"speculative" : false,
"accumulatorUpdates" : [ ],
"taskMetrics" : {
- "executorDeserializeTime" : 1,
+ "executorDeserializeTime" : 2,
"executorDeserializeCpuTime" : 0,
- "executorRunTime" : 435,
+ "executorRunTime" : 434,
"executorCpuTime" : 0,
"resultSize" : 1902,
"jvmGcTime" : 19,
- "resultSerializationTime" : 2,
+ "resultSerializationTime" : 1,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -66,17 +71,17 @@
},
"shuffleWriteMetrics" : {
"bytesWritten" : 1648,
- "writeTime" : 94000,
+ "writeTime" : 76000,
"recordsWritten" : 0
}
}
},
- "9" : {
- "taskId" : 9,
- "index" : 1,
+ "14" : {
+ "taskId" : 14,
+ "index" : 6,
"attempt" : 0,
- "launchTime" : "2015-02-03T16:43:05.830GMT",
- "duration" : 436,
+ "launchTime" : "2015-02-03T16:43:05.832GMT",
+ "duration" : 450,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -84,15 +89,16 @@
"speculative" : false,
"accumulatorUpdates" : [ ],
"taskMetrics" : {
- "executorDeserializeTime" : 1,
+ "executorDeserializeTime" : 2,
"executorDeserializeCpuTime" : 0,
- "executorRunTime" : 436,
+ "executorRunTime" : 434,
"executorCpuTime" : 0,
"resultSize" : 1902,
"jvmGcTime" : 19,
- "resultSerializationTime" : 0,
+ "resultSerializationTime" : 1,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -112,17 +118,17 @@
},
"shuffleWriteMetrics" : {
"bytesWritten" : 1648,
- "writeTime" : 98000,
+ "writeTime" : 88000,
"recordsWritten" : 0
}
}
},
- "10" : {
- "taskId" : 10,
- "index" : 2,
+ "9" : {
+ "taskId" : 9,
+ "index" : 1,
"attempt" : 0,
"launchTime" : "2015-02-03T16:43:05.830GMT",
- "duration" : 434,
+ "duration" : 454,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -130,15 +136,16 @@
"speculative" : false,
"accumulatorUpdates" : [ ],
"taskMetrics" : {
- "executorDeserializeTime" : 2,
+ "executorDeserializeTime" : 1,
"executorDeserializeCpuTime" : 0,
- "executorRunTime" : 434,
+ "executorRunTime" : 436,
"executorCpuTime" : 0,
"resultSize" : 1902,
"jvmGcTime" : 19,
- "resultSerializationTime" : 1,
+ "resultSerializationTime" : 0,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -158,17 +165,17 @@
},
"shuffleWriteMetrics" : {
"bytesWritten" : 1648,
- "writeTime" : 76000,
+ "writeTime" : 98000,
"recordsWritten" : 0
}
}
},
- "11" : {
- "taskId" : 11,
- "index" : 3,
+ "13" : {
+ "taskId" : 13,
+ "index" : 5,
"attempt" : 0,
- "launchTime" : "2015-02-03T16:43:05.830GMT",
- "duration" : 434,
+ "launchTime" : "2015-02-03T16:43:05.831GMT",
+ "duration" : 452,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -182,9 +189,10 @@
"executorCpuTime" : 0,
"resultSize" : 1902,
"jvmGcTime" : 19,
- "resultSerializationTime" : 1,
+ "resultSerializationTime" : 2,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -203,8 +211,8 @@
"recordsRead" : 0
},
"shuffleWriteMetrics" : {
- "bytesWritten" : 1647,
- "writeTime" : 83000,
+ "bytesWritten" : 1648,
+ "writeTime" : 73000,
"recordsWritten" : 0
}
}
@@ -214,7 +222,7 @@
"index" : 4,
"attempt" : 0,
"launchTime" : "2015-02-03T16:43:05.831GMT",
- "duration" : 434,
+ "duration" : 454,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -231,6 +239,7 @@
"resultSerializationTime" : 1,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -255,12 +264,12 @@
}
}
},
- "13" : {
- "taskId" : 13,
- "index" : 5,
+ "11" : {
+ "taskId" : 11,
+ "index" : 3,
"attempt" : 0,
- "launchTime" : "2015-02-03T16:43:05.831GMT",
- "duration" : 434,
+ "launchTime" : "2015-02-03T16:43:05.830GMT",
+ "duration" : 454,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -274,9 +283,10 @@
"executorCpuTime" : 0,
"resultSize" : 1902,
"jvmGcTime" : 19,
- "resultSerializationTime" : 2,
+ "resultSerializationTime" : 1,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -295,18 +305,18 @@
"recordsRead" : 0
},
"shuffleWriteMetrics" : {
- "bytesWritten" : 1648,
- "writeTime" : 73000,
+ "bytesWritten" : 1647,
+ "writeTime" : 83000,
"recordsWritten" : 0
}
}
},
- "14" : {
- "taskId" : 14,
- "index" : 6,
+ "8" : {
+ "taskId" : 8,
+ "index" : 0,
"attempt" : 0,
- "launchTime" : "2015-02-03T16:43:05.832GMT",
- "duration" : 434,
+ "launchTime" : "2015-02-03T16:43:05.829GMT",
+ "duration" : 454,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -314,15 +324,16 @@
"speculative" : false,
"accumulatorUpdates" : [ ],
"taskMetrics" : {
- "executorDeserializeTime" : 2,
+ "executorDeserializeTime" : 1,
"executorDeserializeCpuTime" : 0,
- "executorRunTime" : 434,
+ "executorRunTime" : 435,
"executorCpuTime" : 0,
"resultSize" : 1902,
"jvmGcTime" : 19,
- "resultSerializationTime" : 1,
+ "resultSerializationTime" : 2,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -342,7 +353,7 @@
},
"shuffleWriteMetrics" : {
"bytesWritten" : 1648,
- "writeTime" : 88000,
+ "writeTime" : 94000,
"recordsWritten" : 0
}
}
@@ -352,7 +363,7 @@
"index" : 7,
"attempt" : 0,
"launchTime" : "2015-02-03T16:43:05.833GMT",
- "duration" : 435,
+ "duration" : 450,
"executorId" : "",
"host" : "localhost",
"status" : "SUCCESS",
@@ -369,6 +380,7 @@
"resultSerializationTime" : 1,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0,
+ "peakExecutionMemory" : 0,
"inputMetrics" : {
"bytesRead" : 3500016,
"recordsRead" : 0
@@ -399,12 +411,18 @@
"taskTime" : 3624,
"failedTasks" : 0,
"succeededTasks" : 8,
+ "killedTasks" : 0,
"inputBytes" : 28000128,
+ "inputRecords" : 0,
"outputBytes" : 0,
+ "outputRecords" : 0,
"shuffleRead" : 0,
+ "shuffleReadRecords" : 0,
"shuffleWrite" : 13180,
+ "shuffleWriteRecords" : 0,
"memoryBytesSpilled" : 0,
"diskBytesSpilled" : 0
}
- }
+ },
+ "killedTasksSummary" : { }
} ]
diff --git a/core/src/test/resources/HistoryServerExpectations/stage_list_json_expectation.json b/core/src/test/resources/HistoryServerExpectations/stage_list_json_expectation.json
index 6509df1508b3..1e6fb40d6028 100644
--- a/core/src/test/resources/HistoryServerExpectations/stage_list_json_expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/stage_list_json_expectation.json
@@ -2,9 +2,12 @@
"status" : "COMPLETE",
"stageId" : 3,
"attemptId" : 0,
+ "numTasks" : 8,
"numActiveTasks" : 0,
"numCompleteTasks" : 8,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"executorRunTime" : 162,
"executorCpuTime" : 0,
"submissionTime" : "2015-02-03T16:43:07.191GMT",
@@ -23,14 +26,51 @@
"name" : "count at :17",
"details" : "org.apache.spark.rdd.RDD.count(RDD.scala:910)\n$line19.$read$$iwC$$iwC$$iwC$$iwC.(:17)\n$line19.$read$$iwC$$iwC$$iwC.(:22)\n$line19.$read$$iwC$$iwC.(:24)\n$line19.$read$$iwC.(:26)\n$line19.$read.(:28)\n$line19.$read$.(:32)\n$line19.$read$.()\n$line19.$eval$.(:7)\n$line19.$eval$.()\n$line19.$eval.$print()\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:606)\norg.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:852)\norg.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1125)\norg.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:674)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:705)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:669)",
"schedulingPool" : "default",
- "accumulatorUpdates" : [ ]
+ "rddIds" : [ 6, 5 ],
+ "accumulatorUpdates" : [ ],
+ "killedTasksSummary" : { }
+}, {
+ "status" : "FAILED",
+ "stageId" : 2,
+ "attemptId" : 0,
+ "numTasks" : 8,
+ "numActiveTasks" : 0,
+ "numCompleteTasks" : 7,
+ "numFailedTasks" : 1,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 7,
+ "executorRunTime" : 278,
+ "executorCpuTime" : 0,
+ "submissionTime" : "2015-02-03T16:43:06.296GMT",
+ "firstTaskLaunchedTime" : "2015-02-03T16:43:06.296GMT",
+ "completionTime" : "2015-02-03T16:43:06.347GMT",
+ "failureReason" : "Job aborted due to stage failure: Task 3 in stage 2.0 failed 1 times, most recent failure: Lost task 3.0 in stage 2.0 (TID 19, localhost): java.lang.RuntimeException: got a 3, failing\n\tat $line11.$read$$iwC$$iwC$$iwC$$iwC$$anonfun$1.apply(:18)\n\tat $line11.$read$$iwC$$iwC$$iwC$$iwC$$anonfun$1.apply(:17)\n\tat scala.collection.Iterator$$anon$11.next(Iterator.scala:328)\n\tat org.apache.spark.util.Utils$.getIteratorSize(Utils.scala:1311)\n\tat org.apache.spark.rdd.RDD$$anonfun$count$1.apply(RDD.scala:910)\n\tat org.apache.spark.rdd.RDD$$anonfun$count$1.apply(RDD.scala:910)\n\tat org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1314)\n\tat org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1314)\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:56)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n\tat java.lang.Thread.run(Thread.java:745)\n\nDriver stacktrace:",
+ "inputBytes" : 0,
+ "inputRecords" : 0,
+ "outputBytes" : 0,
+ "outputRecords" : 0,
+ "shuffleReadBytes" : 0,
+ "shuffleReadRecords" : 0,
+ "shuffleWriteBytes" : 0,
+ "shuffleWriteRecords" : 0,
+ "memoryBytesSpilled" : 0,
+ "diskBytesSpilled" : 0,
+ "name" : "count at :20",
+ "details" : "org.apache.spark.rdd.RDD.count(RDD.scala:910)\n$line11.$read$$iwC$$iwC$$iwC$$iwC.(:20)\n$line11.$read$$iwC$$iwC$$iwC.(:25)\n$line11.$read$$iwC$$iwC.(:27)\n$line11.$read$$iwC.(:29)\n$line11.$read.(:31)\n$line11.$read$.(:35)\n$line11.$read$.()\n$line11.$eval$.(:7)\n$line11.$eval$.()\n$line11.$eval.$print()\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:606)\norg.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:852)\norg.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1125)\norg.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:674)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:705)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:669)",
+ "schedulingPool" : "default",
+ "rddIds" : [ 3, 2 ],
+ "accumulatorUpdates" : [ ],
+ "killedTasksSummary" : { }
}, {
"status" : "COMPLETE",
"stageId" : 1,
"attemptId" : 0,
+ "numTasks" : 8,
"numActiveTasks" : 0,
"numCompleteTasks" : 8,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"executorRunTime" : 3476,
"executorCpuTime" : 0,
"submissionTime" : "2015-02-03T16:43:05.829GMT",
@@ -49,14 +89,19 @@
"name" : "map at :14",
"details" : "org.apache.spark.rdd.RDD.map(RDD.scala:271)\n$line10.$read$$iwC$$iwC$$iwC$$iwC.(:14)\n$line10.$read$$iwC$$iwC$$iwC.(:19)\n$line10.$read$$iwC$$iwC.(:21)\n$line10.$read$$iwC.(:23)\n$line10.$read.(:25)\n$line10.$read$.(:29)\n$line10.$read$.()\n$line10.$eval$.(:7)\n$line10.$eval$.()\n$line10.$eval.$print()\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:606)\norg.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:852)\norg.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1125)\norg.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:674)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:705)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:669)",
"schedulingPool" : "default",
- "accumulatorUpdates" : [ ]
+ "rddIds" : [ 1, 0 ],
+ "accumulatorUpdates" : [ ],
+ "killedTasksSummary" : { }
}, {
"status" : "COMPLETE",
"stageId" : 0,
"attemptId" : 0,
+ "numTasks" : 8,
"numActiveTasks" : 0,
"numCompleteTasks" : 8,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"executorRunTime" : 4338,
"executorCpuTime" : 0,
"submissionTime" : "2015-02-03T16:43:04.228GMT",
@@ -75,31 +120,7 @@
"name" : "count at :15",
"details" : "org.apache.spark.rdd.RDD.count(RDD.scala:910)\n$line9.$read$$iwC$$iwC$$iwC$$iwC.(:15)\n$line9.$read$$iwC$$iwC$$iwC.(:20)\n$line9.$read$$iwC$$iwC.(:22)\n$line9.$read$$iwC.(:24)\n$line9.$read.(:26)\n$line9.$read$.(:30)\n$line9.$read$.()\n$line9.$eval$.(:7)\n$line9.$eval$.()\n$line9.$eval.$print()\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:606)\norg.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:852)\norg.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1125)\norg.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:674)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:705)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:669)",
"schedulingPool" : "default",
- "accumulatorUpdates" : [ ]
-}, {
- "status" : "FAILED",
- "stageId" : 2,
- "attemptId" : 0,
- "numActiveTasks" : 0,
- "numCompleteTasks" : 7,
- "numFailedTasks" : 1,
- "executorRunTime" : 278,
- "executorCpuTime" : 0,
- "submissionTime" : "2015-02-03T16:43:06.296GMT",
- "firstTaskLaunchedTime" : "2015-02-03T16:43:06.296GMT",
- "completionTime" : "2015-02-03T16:43:06.347GMT",
- "inputBytes" : 0,
- "inputRecords" : 0,
- "outputBytes" : 0,
- "outputRecords" : 0,
- "shuffleReadBytes" : 0,
- "shuffleReadRecords" : 0,
- "shuffleWriteBytes" : 0,
- "shuffleWriteRecords" : 0,
- "memoryBytesSpilled" : 0,
- "diskBytesSpilled" : 0,
- "name" : "count at :20",
- "details" : "org.apache.spark.rdd.RDD.count(RDD.scala:910)\n$line11.$read$$iwC$$iwC$$iwC$$iwC.(:20)\n$line11.$read$$iwC$$iwC$$iwC.(:25)\n$line11.$read$$iwC$$iwC.(:27)\n$line11.$read$$iwC.(:29)\n$line11.$read.(:31)\n$line11.$read$.(:35)\n$line11.$read$.()\n$line11.$eval$.(:7)\n$line11.$eval$.()\n$line11.$eval.$print()\nsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\nsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\nsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\njava.lang.reflect.Method.invoke(Method.java:606)\norg.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:852)\norg.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1125)\norg.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:674)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:705)\norg.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:669)",
- "schedulingPool" : "default",
- "accumulatorUpdates" : [ ]
+ "rddIds" : [ 0 ],
+ "accumulatorUpdates" : [ ],
+ "killedTasksSummary" : { }
} ]
diff --git a/core/src/test/resources/HistoryServerExpectations/stage_list_with_accumulable_json_expectation.json b/core/src/test/resources/HistoryServerExpectations/stage_list_with_accumulable_json_expectation.json
index 8496863a9346..e6284ccf9b73 100644
--- a/core/src/test/resources/HistoryServerExpectations/stage_list_with_accumulable_json_expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/stage_list_with_accumulable_json_expectation.json
@@ -2,9 +2,12 @@
"status" : "COMPLETE",
"stageId" : 0,
"attemptId" : 0,
+ "numTasks" : 8,
"numActiveTasks" : 0,
"numCompleteTasks" : 8,
"numFailedTasks" : 0,
+ "numKilledTasks" : 0,
+ "numCompletedIndices" : 8,
"executorRunTime" : 120,
"executorCpuTime" : 0,
"submissionTime" : "2015-03-16T19:25:36.103GMT",
@@ -23,9 +26,11 @@
"name" : "foreach at :15",
"details" : "org.apache.spark.rdd.RDD.foreach(RDD.scala:765)\n$line9.$read$$iwC$$iwC$$iwC$$iwC.(:15)\n$line9.$read$$iwC$$iwC$$iwC.(:20)\n$line9.$read$$iwC$$iwC.(:22)\n$line9.$read$$iwC.(:24)\n$line9.$read.