You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ir...@apache.org on 2017/11/08 05:14:39 UTC
spark git commit: [SPARK-20646][CORE] Port executors page to new UI
backend.
Repository: spark
Updated Branches:
refs/heads/master 2ca5aae47 -> 11eea1a4c
[SPARK-20646][CORE] Port executors page to new UI backend.
The executors page is built on top of the REST API, so the page itself
was easy to hook up to the new code.
Some other pages depend on the `ExecutorListener` class that is being
removed, though, so they needed to be modified to use data from the
new store. Fortunately, all they seemed to need is the map of executor
logs, so that was somewhat easy too.
The executor timeline graph required adding some properties to the
ExecutorSummary API type. Instead of following the previous code,
which stored all the listener events in memory, the timeline is
now created based on the data available from the API.
I had to change some of the test golden files because the old code would
return executors in "random" order (since it used a mutable Map instead
of something that returns a sorted list), and the new code returns executors
in id order.
Tested with existing unit tests.
Author: Marcelo Vanzin <va...@cloudera.com>
Closes #19678 from vanzin/SPARK-20646.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/11eea1a4
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/11eea1a4
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/11eea1a4
Branch: refs/heads/master
Commit: 11eea1a4ce32c9018218d4dfc9f46b744eb82991
Parents: 2ca5aae
Author: Marcelo Vanzin <va...@cloudera.com>
Authored: Tue Nov 7 23:14:29 2017 -0600
Committer: Imran Rashid <ir...@cloudera.com>
Committed: Tue Nov 7 23:14:29 2017 -0600
----------------------------------------------------------------------
.../apache/spark/status/AppStatusListener.scala | 36 ++--
.../apache/spark/status/AppStatusStore.scala | 18 +-
.../org/apache/spark/status/LiveEntity.scala | 9 +-
.../status/api/v1/AllExecutorListResource.scala | 15 +-
.../status/api/v1/ExecutorListResource.scala | 14 +-
.../org/apache/spark/status/api/v1/api.scala | 3 +
.../scala/org/apache/spark/ui/SparkUI.scala | 13 +-
.../spark/ui/exec/ExecutorThreadDumpPage.scala | 9 +-
.../apache/spark/ui/exec/ExecutorsPage.scala | 154 --------------
.../org/apache/spark/ui/exec/ExecutorsTab.scala | 206 +++----------------
.../org/apache/spark/ui/jobs/AllJobsPage.scala | 60 +++---
.../apache/spark/ui/jobs/ExecutorTable.scala | 11 +-
.../org/apache/spark/ui/jobs/JobPage.scala | 59 +++---
.../org/apache/spark/ui/jobs/JobsTab.scala | 3 +-
.../org/apache/spark/ui/jobs/StagePage.scala | 18 +-
.../org/apache/spark/ui/jobs/StagesTab.scala | 8 +-
.../executor_list_json_expectation.json | 1 +
.../executor_memory_usage_expectation.json | 127 ++++++------
.../executor_node_blacklisting_expectation.json | 117 ++++++-----
...blacklisting_unblacklisting_expectation.json | 89 ++++----
.../org/apache/spark/ui/StagePageSuite.scala | 11 +-
project/MimaExcludes.scala | 1 +
22 files changed, 354 insertions(+), 628 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala b/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
index 424a115..0469c87 100644
--- a/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
+++ b/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
@@ -103,6 +103,9 @@ private[spark] class AppStatusListener(
details.getOrElse("System Properties", Nil),
details.getOrElse("Classpath Entries", Nil))
+ coresPerTask = envInfo.sparkProperties.toMap.get("spark.task.cpus").map(_.toInt)
+ .getOrElse(coresPerTask)
+
kvstore.write(new ApplicationEnvironmentInfoWrapper(envInfo))
}
@@ -132,7 +135,7 @@ private[spark] class AppStatusListener(
override def onExecutorAdded(event: SparkListenerExecutorAdded): Unit = {
// This needs to be an update in case an executor re-registers after the driver has
// marked it as "dead".
- val exec = getOrCreateExecutor(event.executorId)
+ val exec = getOrCreateExecutor(event.executorId, event.time)
exec.host = event.executorInfo.executorHost
exec.isActive = true
exec.totalCores = event.executorInfo.totalCores
@@ -144,6 +147,8 @@ private[spark] class AppStatusListener(
override def onExecutorRemoved(event: SparkListenerExecutorRemoved): Unit = {
liveExecutors.remove(event.executorId).foreach { exec =>
exec.isActive = false
+ exec.removeTime = new Date(event.time)
+ exec.removeReason = event.reason
update(exec, System.nanoTime())
}
}
@@ -357,18 +362,25 @@ private[spark] class AppStatusListener(
}
liveExecutors.get(event.taskInfo.executorId).foreach { exec =>
- if (event.taskMetrics != null) {
- val readMetrics = event.taskMetrics.shuffleReadMetrics
- exec.totalGcTime += event.taskMetrics.jvmGCTime
- exec.totalInputBytes += event.taskMetrics.inputMetrics.bytesRead
- exec.totalShuffleRead += readMetrics.localBytesRead + readMetrics.remoteBytesRead
- exec.totalShuffleWrite += event.taskMetrics.shuffleWriteMetrics.bytesWritten
- }
-
exec.activeTasks -= 1
exec.completedTasks += completedDelta
exec.failedTasks += failedDelta
exec.totalDuration += event.taskInfo.duration
+
+ // Note: For resubmitted tasks, we continue to use the metrics that belong to the
+ // first attempt of this task. This may not be 100% accurate because the first attempt
+ // could have failed half-way through. The correct fix would be to keep track of the
+ // metrics added by each attempt, but this is much more complicated.
+ if (event.reason != Resubmitted) {
+ if (event.taskMetrics != null) {
+ val readMetrics = event.taskMetrics.shuffleReadMetrics
+ exec.totalGcTime += event.taskMetrics.jvmGCTime
+ exec.totalInputBytes += event.taskMetrics.inputMetrics.bytesRead
+ exec.totalShuffleRead += readMetrics.localBytesRead + readMetrics.remoteBytesRead
+ exec.totalShuffleWrite += event.taskMetrics.shuffleWriteMetrics.bytesWritten
+ }
+ }
+
maybeUpdate(exec, now)
}
}
@@ -409,7 +421,7 @@ private[spark] class AppStatusListener(
override def onBlockManagerAdded(event: SparkListenerBlockManagerAdded): Unit = {
// This needs to set fields that are already set by onExecutorAdded because the driver is
// considered an "executor" in the UI, but does not have a SparkListenerExecutorAdded event.
- val exec = getOrCreateExecutor(event.blockManagerId.executorId)
+ val exec = getOrCreateExecutor(event.blockManagerId.executorId, event.time)
exec.hostPort = event.blockManagerId.hostPort
event.maxOnHeapMem.foreach { _ =>
exec.totalOnHeap = event.maxOnHeapMem.get
@@ -561,8 +573,8 @@ private[spark] class AppStatusListener(
}
}
- private def getOrCreateExecutor(executorId: String): LiveExecutor = {
- liveExecutors.getOrElseUpdate(executorId, new LiveExecutor(executorId))
+ private def getOrCreateExecutor(executorId: String, addTime: Long): LiveExecutor = {
+ liveExecutors.getOrElseUpdate(executorId, new LiveExecutor(executorId, addTime))
}
private def getOrCreateStage(info: StageInfo): LiveStage = {
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala
index d6b5d26..3344078 100644
--- a/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala
+++ b/core/src/main/scala/org/apache/spark/status/AppStatusStore.scala
@@ -56,8 +56,22 @@ private[spark] class AppStatusStore(store: KVStore) {
}
def executorList(activeOnly: Boolean): Seq[v1.ExecutorSummary] = {
- store.view(classOf[ExecutorSummaryWrapper]).index("active").reverse().first(true)
- .last(true).asScala.map(_.info).toSeq
+ val base = store.view(classOf[ExecutorSummaryWrapper])
+ val filtered = if (activeOnly) {
+ base.index("active").reverse().first(true).last(true)
+ } else {
+ base
+ }
+ filtered.asScala.map(_.info).toSeq
+ }
+
+ def executorSummary(executorId: String): Option[v1.ExecutorSummary] = {
+ try {
+ Some(store.read(classOf[ExecutorSummaryWrapper], executorId).info)
+ } catch {
+ case _: NoSuchElementException =>
+ None
+ }
}
def stageList(statuses: JList[v1.StageStatus]): Seq[v1.StageData] = {
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/status/LiveEntity.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/status/LiveEntity.scala b/core/src/main/scala/org/apache/spark/status/LiveEntity.scala
index 041dfe1..8c48020 100644
--- a/core/src/main/scala/org/apache/spark/status/LiveEntity.scala
+++ b/core/src/main/scala/org/apache/spark/status/LiveEntity.scala
@@ -212,13 +212,17 @@ private class LiveTask(
}
-private class LiveExecutor(val executorId: String) extends LiveEntity {
+private class LiveExecutor(val executorId: String, _addTime: Long) extends LiveEntity {
var hostPort: String = null
var host: String = null
var isActive = true
var totalCores = 0
+ val addTime = new Date(_addTime)
+ var removeTime: Date = null
+ var removeReason: String = null
+
var rddBlocks = 0
var memoryUsed = 0L
var diskUsed = 0L
@@ -276,6 +280,9 @@ private class LiveExecutor(val executorId: String) extends LiveEntity {
totalShuffleWrite,
isBlacklisted,
maxMemory,
+ addTime,
+ Option(removeTime),
+ Option(removeReason),
executorLogs,
memoryMetrics)
new ExecutorSummaryWrapper(info)
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/status/api/v1/AllExecutorListResource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/AllExecutorListResource.scala b/core/src/main/scala/org/apache/spark/status/api/v1/AllExecutorListResource.scala
index eb5cc1b..5522f4c 100644
--- a/core/src/main/scala/org/apache/spark/status/api/v1/AllExecutorListResource.scala
+++ b/core/src/main/scala/org/apache/spark/status/api/v1/AllExecutorListResource.scala
@@ -20,22 +20,11 @@ import javax.ws.rs.{GET, Produces}
import javax.ws.rs.core.MediaType
import org.apache.spark.ui.SparkUI
-import org.apache.spark.ui.exec.ExecutorsPage
@Produces(Array(MediaType.APPLICATION_JSON))
private[v1] class AllExecutorListResource(ui: SparkUI) {
@GET
- def executorList(): Seq[ExecutorSummary] = {
- val listener = ui.executorsListener
- listener.synchronized {
- // The follow codes should be protected by `listener` to make sure no executors will be
- // removed before we query their status. See SPARK-12784.
- (0 until listener.activeStorageStatusList.size).map { statusId =>
- ExecutorsPage.getExecInfo(listener, statusId, isActive = true)
- } ++ (0 until listener.deadStorageStatusList.size).map { statusId =>
- ExecutorsPage.getExecInfo(listener, statusId, isActive = false)
- }
- }
- }
+ def executorList(): Seq[ExecutorSummary] = ui.store.executorList(false)
+
}
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/status/api/v1/ExecutorListResource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/ExecutorListResource.scala b/core/src/main/scala/org/apache/spark/status/api/v1/ExecutorListResource.scala
index 2f3b5e9..975101c 100644
--- a/core/src/main/scala/org/apache/spark/status/api/v1/ExecutorListResource.scala
+++ b/core/src/main/scala/org/apache/spark/status/api/v1/ExecutorListResource.scala
@@ -20,21 +20,11 @@ import javax.ws.rs.{GET, Produces}
import javax.ws.rs.core.MediaType
import org.apache.spark.ui.SparkUI
-import org.apache.spark.ui.exec.ExecutorsPage
@Produces(Array(MediaType.APPLICATION_JSON))
private[v1] class ExecutorListResource(ui: SparkUI) {
@GET
- def executorList(): Seq[ExecutorSummary] = {
- val listener = ui.executorsListener
- listener.synchronized {
- // The follow codes should be protected by `listener` to make sure no executors will be
- // removed before we query their status. See SPARK-12784.
- val storageStatusList = listener.activeStorageStatusList
- (0 until storageStatusList.size).map { statusId =>
- ExecutorsPage.getExecInfo(listener, statusId, isActive = true)
- }
- }
- }
+ def executorList(): Seq[ExecutorSummary] = ui.store.executorList(true)
+
}
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/status/api/v1/api.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/api.scala b/core/src/main/scala/org/apache/spark/status/api/v1/api.scala
index bff6f90..b338b1f 100644
--- a/core/src/main/scala/org/apache/spark/status/api/v1/api.scala
+++ b/core/src/main/scala/org/apache/spark/status/api/v1/api.scala
@@ -85,6 +85,9 @@ class ExecutorSummary private[spark](
val totalShuffleWrite: Long,
val isBlacklisted: Boolean,
val maxMemory: Long,
+ val addTime: Date,
+ val removeTime: Option[Date],
+ val removeReason: Option[String],
val executorLogs: Map[String, String],
val memoryMetrics: Option[MemoryMetrics])
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/SparkUI.scala b/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
index 43b57a1..79d40b6 100644
--- a/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
+++ b/core/src/main/scala/org/apache/spark/ui/SparkUI.scala
@@ -29,7 +29,7 @@ import org.apache.spark.status.api.v1._
import org.apache.spark.storage.StorageStatusListener
import org.apache.spark.ui.JettyUtils._
import org.apache.spark.ui.env.EnvironmentTab
-import org.apache.spark.ui.exec.{ExecutorsListener, ExecutorsTab}
+import org.apache.spark.ui.exec.ExecutorsTab
import org.apache.spark.ui.jobs.{JobProgressListener, JobsTab, StagesTab}
import org.apache.spark.ui.scope.RDDOperationGraphListener
import org.apache.spark.ui.storage.{StorageListener, StorageTab}
@@ -44,7 +44,6 @@ private[spark] class SparkUI private (
val conf: SparkConf,
securityManager: SecurityManager,
val storageStatusListener: StorageStatusListener,
- val executorsListener: ExecutorsListener,
val jobProgressListener: JobProgressListener,
val storageListener: StorageListener,
val operationGraphListener: RDDOperationGraphListener,
@@ -68,7 +67,7 @@ private[spark] class SparkUI private (
def initialize() {
val jobsTab = new JobsTab(this)
attachTab(jobsTab)
- val stagesTab = new StagesTab(this)
+ val stagesTab = new StagesTab(this, store)
attachTab(stagesTab)
attachTab(new StorageTab(this))
attachTab(new EnvironmentTab(this, store))
@@ -189,18 +188,16 @@ private[spark] object SparkUI {
}
val storageStatusListener = new StorageStatusListener(conf)
- val executorsListener = new ExecutorsListener(storageStatusListener, conf)
val storageListener = new StorageListener(storageStatusListener)
val operationGraphListener = new RDDOperationGraphListener(conf)
addListenerFn(storageStatusListener)
- addListenerFn(executorsListener)
addListenerFn(storageListener)
addListenerFn(operationGraphListener)
- new SparkUI(store, sc, conf, securityManager, storageStatusListener, executorsListener,
- jobProgressListener, storageListener, operationGraphListener, appName, basePath,
- lastUpdateTime, startTime, appSparkVersion)
+ new SparkUI(store, sc, conf, securityManager, storageStatusListener, jobProgressListener,
+ storageListener, operationGraphListener, appName, basePath, lastUpdateTime, startTime,
+ appSparkVersion)
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/ui/exec/ExecutorThreadDumpPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorThreadDumpPage.scala b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorThreadDumpPage.scala
index 7b211ea..f4686ea 100644
--- a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorThreadDumpPage.scala
+++ b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorThreadDumpPage.scala
@@ -22,11 +22,12 @@ import javax.servlet.http.HttpServletRequest
import scala.xml.{Node, Text}
-import org.apache.spark.ui.{UIUtils, WebUIPage}
+import org.apache.spark.SparkContext
+import org.apache.spark.ui.{SparkUITab, UIUtils, WebUIPage}
-private[ui] class ExecutorThreadDumpPage(parent: ExecutorsTab) extends WebUIPage("threadDump") {
-
- private val sc = parent.sc
+private[ui] class ExecutorThreadDumpPage(
+ parent: SparkUITab,
+ sc: Option[SparkContext]) extends WebUIPage("threadDump") {
// stripXSS is called first to remove suspicious characters used in XSS attacks
def render(request: HttpServletRequest): Seq[Node] = {
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala
deleted file mode 100644
index 7b2767f..0000000
--- a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala
+++ /dev/null
@@ -1,154 +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.exec
-
-import javax.servlet.http.HttpServletRequest
-
-import scala.xml.Node
-
-import org.apache.spark.status.api.v1.{ExecutorSummary, MemoryMetrics}
-import org.apache.spark.ui.{UIUtils, WebUIPage}
-
-// This isn't even used anymore -- but we need to keep it b/c of a MiMa false positive
-private[ui] case class ExecutorSummaryInfo(
- id: String,
- hostPort: String,
- rddBlocks: Int,
- memoryUsed: Long,
- diskUsed: Long,
- activeTasks: Int,
- failedTasks: Int,
- completedTasks: Int,
- totalTasks: Int,
- totalDuration: Long,
- totalInputBytes: Long,
- totalShuffleRead: Long,
- totalShuffleWrite: Long,
- isBlacklisted: Int,
- maxOnHeapMem: Long,
- maxOffHeapMem: Long,
- executorLogs: Map[String, String])
-
-
-private[ui] class ExecutorsPage(
- parent: ExecutorsTab,
- threadDumpEnabled: Boolean)
- extends WebUIPage("") {
-
- def render(request: HttpServletRequest): Seq[Node] = {
- val content =
- <div>
- {
- <div>
- <span class="expand-additional-metrics">
- <span class="expand-additional-metrics-arrow arrow-closed"></span>
- <a>Show Additional Metrics</a>
- </span>
- <div class="additional-metrics collapsed">
- <ul>
- <li>
- <input type="checkbox" id="select-all-metrics"/>
- <span class="additional-metric-title"><em>(De)select All</em></span>
- </li>
- <li>
- <span data-toggle="tooltip"
- title={ExecutorsPage.ON_HEAP_MEMORY_TOOLTIP} data-placement="right">
- <input type="checkbox" name="on_heap_memory"/>
- <span class="additional-metric-title">On Heap Storage Memory</span>
- </span>
- </li>
- <li>
- <span data-toggle="tooltip"
- title={ExecutorsPage.OFF_HEAP_MEMORY_TOOLTIP} data-placement="right">
- <input type="checkbox" name="off_heap_memory"/>
- <span class="additional-metric-title">Off Heap Storage Memory</span>
- </span>
- </li>
- </ul>
- </div>
- </div> ++
- <div id="active-executors" class="row-fluid"></div> ++
- <script src={UIUtils.prependBaseUri("/static/utils.js")}></script> ++
- <script src={UIUtils.prependBaseUri("/static/executorspage.js")}></script> ++
- <script>setThreadDumpEnabled({threadDumpEnabled})</script>
- }
- </div>
-
- UIUtils.headerSparkPage("Executors", content, parent, useDataTables = true)
- }
-}
-
-private[spark] object ExecutorsPage {
- private val ON_HEAP_MEMORY_TOOLTIP = "Memory used / total available memory for on heap " +
- "storage of data like RDD partitions cached in memory."
- private val OFF_HEAP_MEMORY_TOOLTIP = "Memory used / total available memory for off heap " +
- "storage of data like RDD partitions cached in memory."
-
- /** Represent an executor's info as a map given a storage status index */
- def getExecInfo(
- listener: ExecutorsListener,
- statusId: Int,
- isActive: Boolean): ExecutorSummary = {
- val status = if (isActive) {
- listener.activeStorageStatusList(statusId)
- } else {
- listener.deadStorageStatusList(statusId)
- }
- val execId = status.blockManagerId.executorId
- val hostPort = status.blockManagerId.hostPort
- val rddBlocks = status.numBlocks
- val memUsed = status.memUsed
- val maxMem = status.maxMem
- val memoryMetrics = for {
- onHeapUsed <- status.onHeapMemUsed
- offHeapUsed <- status.offHeapMemUsed
- maxOnHeap <- status.maxOnHeapMem
- maxOffHeap <- status.maxOffHeapMem
- } yield {
- new MemoryMetrics(onHeapUsed, offHeapUsed, maxOnHeap, maxOffHeap)
- }
-
-
- val diskUsed = status.diskUsed
- val taskSummary = listener.executorToTaskSummary.getOrElse(execId, ExecutorTaskSummary(execId))
-
- new ExecutorSummary(
- execId,
- hostPort,
- isActive,
- rddBlocks,
- memUsed,
- diskUsed,
- taskSummary.totalCores,
- taskSummary.tasksMax,
- taskSummary.tasksActive,
- taskSummary.tasksFailed,
- taskSummary.tasksComplete,
- taskSummary.tasksActive + taskSummary.tasksFailed + taskSummary.tasksComplete,
- taskSummary.duration,
- taskSummary.jvmGCTime,
- taskSummary.inputBytes,
- taskSummary.shuffleRead,
- taskSummary.shuffleWrite,
- taskSummary.isBlacklisted,
- maxMem,
- taskSummary.executorLogs,
- memoryMetrics
- )
- }
-}
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala
index 64a1a29..843486f 100644
--- a/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala
+++ b/core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala
@@ -17,192 +17,44 @@
package org.apache.spark.ui.exec
-import scala.collection.mutable.{LinkedHashMap, ListBuffer}
+import javax.servlet.http.HttpServletRequest
-import org.apache.spark.{Resubmitted, SparkConf, SparkContext}
-import org.apache.spark.annotation.DeveloperApi
-import org.apache.spark.scheduler._
-import org.apache.spark.storage.{StorageStatus, StorageStatusListener}
-import org.apache.spark.ui.{SparkUI, SparkUITab}
+import scala.xml.Node
-private[ui] class ExecutorsTab(parent: SparkUI) extends SparkUITab(parent, "executors") {
- val listener = parent.executorsListener
- val sc = parent.sc
- val threadDumpEnabled =
- sc.isDefined && parent.conf.getBoolean("spark.ui.threadDumpsEnabled", true)
-
- attachPage(new ExecutorsPage(this, threadDumpEnabled))
- if (threadDumpEnabled) {
- attachPage(new ExecutorThreadDumpPage(this))
- }
-}
-
-private[ui] case class ExecutorTaskSummary(
- var executorId: String,
- var totalCores: Int = 0,
- var tasksMax: Int = 0,
- var tasksActive: Int = 0,
- var tasksFailed: Int = 0,
- var tasksComplete: Int = 0,
- var duration: Long = 0L,
- var jvmGCTime: Long = 0L,
- var inputBytes: Long = 0L,
- var inputRecords: Long = 0L,
- var outputBytes: Long = 0L,
- var outputRecords: Long = 0L,
- var shuffleRead: Long = 0L,
- var shuffleWrite: Long = 0L,
- var executorLogs: Map[String, String] = Map.empty,
- var isAlive: Boolean = true,
- var isBlacklisted: Boolean = false
-)
-
-/**
- * :: DeveloperApi ::
- * A SparkListener that prepares information to be displayed on the ExecutorsTab
- */
-@DeveloperApi
-@deprecated("This class will be removed in a future release.", "2.2.0")
-class ExecutorsListener(storageStatusListener: StorageStatusListener, conf: SparkConf)
- extends SparkListener {
- val executorToTaskSummary = LinkedHashMap[String, ExecutorTaskSummary]()
- var executorEvents = new ListBuffer[SparkListenerEvent]()
-
- private val maxTimelineExecutors = conf.getInt("spark.ui.timeline.executors.maximum", 1000)
- private val retainedDeadExecutors = conf.getInt("spark.ui.retainedDeadExecutors", 100)
-
- def activeStorageStatusList: Seq[StorageStatus] = storageStatusListener.storageStatusList
+import org.apache.spark.ui.{SparkUI, SparkUITab, UIUtils, WebUIPage}
- def deadStorageStatusList: Seq[StorageStatus] = storageStatusListener.deadStorageStatusList
-
- override def onExecutorAdded(
- executorAdded: SparkListenerExecutorAdded): Unit = synchronized {
- val eid = executorAdded.executorId
- val taskSummary = executorToTaskSummary.getOrElseUpdate(eid, ExecutorTaskSummary(eid))
- taskSummary.executorLogs = executorAdded.executorInfo.logUrlMap
- taskSummary.totalCores = executorAdded.executorInfo.totalCores
- taskSummary.tasksMax = taskSummary.totalCores / conf.getInt("spark.task.cpus", 1)
- executorEvents += executorAdded
- if (executorEvents.size > maxTimelineExecutors) {
- executorEvents.remove(0)
- }
-
- val deadExecutors = executorToTaskSummary.filter(e => !e._2.isAlive)
- if (deadExecutors.size > retainedDeadExecutors) {
- val head = deadExecutors.head
- executorToTaskSummary.remove(head._1)
- }
- }
-
- override def onExecutorRemoved(
- executorRemoved: SparkListenerExecutorRemoved): Unit = synchronized {
- executorEvents += executorRemoved
- if (executorEvents.size > maxTimelineExecutors) {
- executorEvents.remove(0)
- }
- executorToTaskSummary.get(executorRemoved.executorId).foreach(e => e.isAlive = false)
- }
-
- override def onApplicationStart(
- applicationStart: SparkListenerApplicationStart): Unit = {
- applicationStart.driverLogs.foreach { logs =>
- val storageStatus = activeStorageStatusList.find { s =>
- s.blockManagerId.executorId == SparkContext.LEGACY_DRIVER_IDENTIFIER ||
- s.blockManagerId.executorId == SparkContext.DRIVER_IDENTIFIER
- }
- storageStatus.foreach { s =>
- val eid = s.blockManagerId.executorId
- val taskSummary = executorToTaskSummary.getOrElseUpdate(eid, ExecutorTaskSummary(eid))
- taskSummary.executorLogs = logs.toMap
- }
- }
- }
-
- override def onTaskStart(
- taskStart: SparkListenerTaskStart): Unit = synchronized {
- val eid = taskStart.taskInfo.executorId
- val taskSummary = executorToTaskSummary.getOrElseUpdate(eid, ExecutorTaskSummary(eid))
- taskSummary.tasksActive += 1
- }
+private[ui] class ExecutorsTab(parent: SparkUI) extends SparkUITab(parent, "executors") {
- override def onTaskEnd(
- taskEnd: SparkListenerTaskEnd): Unit = synchronized {
- val info = taskEnd.taskInfo
- if (info != null) {
- val eid = info.executorId
- val taskSummary = executorToTaskSummary.getOrElseUpdate(eid, ExecutorTaskSummary(eid))
- // Note: For resubmitted tasks, we continue to use the metrics that belong to the
- // first attempt of this task. This may not be 100% accurate because the first attempt
- // could have failed half-way through. The correct fix would be to keep track of the
- // metrics added by each attempt, but this is much more complicated.
- if (taskEnd.reason == Resubmitted) {
- return
- }
- if (info.successful) {
- taskSummary.tasksComplete += 1
- } else {
- taskSummary.tasksFailed += 1
- }
- if (taskSummary.tasksActive >= 1) {
- taskSummary.tasksActive -= 1
- }
- taskSummary.duration += info.duration
+ init()
- // Update shuffle read/write
- val metrics = taskEnd.taskMetrics
- if (metrics != null) {
- taskSummary.inputBytes += metrics.inputMetrics.bytesRead
- taskSummary.inputRecords += metrics.inputMetrics.recordsRead
- taskSummary.outputBytes += metrics.outputMetrics.bytesWritten
- taskSummary.outputRecords += metrics.outputMetrics.recordsWritten
+ private def init(): Unit = {
+ val threadDumpEnabled =
+ parent.sc.isDefined && parent.conf.getBoolean("spark.ui.threadDumpsEnabled", true)
- taskSummary.shuffleRead += metrics.shuffleReadMetrics.remoteBytesRead
- taskSummary.shuffleWrite += metrics.shuffleWriteMetrics.bytesWritten
- taskSummary.jvmGCTime += metrics.jvmGCTime
- }
+ attachPage(new ExecutorsPage(this, threadDumpEnabled))
+ if (threadDumpEnabled) {
+ attachPage(new ExecutorThreadDumpPage(this, parent.sc))
}
}
- private def updateExecutorBlacklist(
- eid: String,
- isBlacklisted: Boolean): Unit = {
- val execTaskSummary = executorToTaskSummary.getOrElseUpdate(eid, ExecutorTaskSummary(eid))
- execTaskSummary.isBlacklisted = isBlacklisted
- }
-
- override def onExecutorBlacklisted(
- executorBlacklisted: SparkListenerExecutorBlacklisted)
- : Unit = synchronized {
- updateExecutorBlacklist(executorBlacklisted.executorId, true)
- }
-
- override def onExecutorUnblacklisted(
- executorUnblacklisted: SparkListenerExecutorUnblacklisted)
- : Unit = synchronized {
- updateExecutorBlacklist(executorUnblacklisted.executorId, false)
- }
-
- override def onNodeBlacklisted(
- nodeBlacklisted: SparkListenerNodeBlacklisted)
- : Unit = synchronized {
- // Implicitly blacklist every executor associated with this node, and show this in the UI.
- activeStorageStatusList.foreach { status =>
- if (status.blockManagerId.host == nodeBlacklisted.hostId) {
- updateExecutorBlacklist(status.blockManagerId.executorId, true)
- }
- }
- }
+}
- override def onNodeUnblacklisted(
- nodeUnblacklisted: SparkListenerNodeUnblacklisted)
- : Unit = synchronized {
- // Implicitly unblacklist every executor associated with this node, regardless of how
- // they may have been blacklisted initially (either explicitly through executor blacklisting
- // or implicitly through node blacklisting). Show this in the UI.
- activeStorageStatusList.foreach { status =>
- if (status.blockManagerId.host == nodeUnblacklisted.hostId) {
- updateExecutorBlacklist(status.blockManagerId.executorId, false)
- }
- }
+private[ui] class ExecutorsPage(
+ parent: SparkUITab,
+ threadDumpEnabled: Boolean)
+ extends WebUIPage("") {
+
+ def render(request: HttpServletRequest): Seq[Node] = {
+ val content =
+ <div>
+ {
+ <div id="active-executors" class="row-fluid"></div> ++
+ <script src={UIUtils.prependBaseUri("/static/utils.js")}></script> ++
+ <script src={UIUtils.prependBaseUri("/static/executorspage.js")}></script> ++
+ <script>setThreadDumpEnabled({threadDumpEnabled})</script>
+ }
+ </div>
+
+ UIUtils.headerSparkPage("Executors", content, parent, useDataTables = true)
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
index a7f2caa..a647a11 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
@@ -29,6 +29,7 @@ import org.apache.commons.lang3.StringEscapeUtils
import org.apache.spark.JobExecutionStatus
import org.apache.spark.scheduler._
+import org.apache.spark.status.api.v1.ExecutorSummary
import org.apache.spark.ui._
import org.apache.spark.ui.jobs.UIData.{JobUIData, StageUIData}
import org.apache.spark.util.Utils
@@ -123,55 +124,53 @@ private[ui] class AllJobsPage(parent: JobsTab) extends WebUIPage("") {
}
}
- private def makeExecutorEvent(executorUIDatas: Seq[SparkListenerEvent]):
+ private def makeExecutorEvent(executors: Seq[ExecutorSummary]):
Seq[String] = {
val events = ListBuffer[String]()
- executorUIDatas.foreach {
- case a: SparkListenerExecutorAdded =>
- val addedEvent =
- s"""
- |{
- | 'className': 'executor added',
- | 'group': 'executors',
- | 'start': new Date(${a.time}),
- | 'content': '<div class="executor-event-content"' +
- | 'data-toggle="tooltip" data-placement="bottom"' +
- | 'data-title="Executor ${a.executorId}<br>' +
- | 'Added at ${UIUtils.formatDate(new Date(a.time))}"' +
- | 'data-html="true">Executor ${a.executorId} added</div>'
- |}
- """.stripMargin
- events += addedEvent
- case e: SparkListenerExecutorRemoved =>
+ executors.foreach { e =>
+ val addedEvent =
+ s"""
+ |{
+ | 'className': 'executor added',
+ | 'group': 'executors',
+ | 'start': new Date(${e.addTime.getTime()}),
+ | 'content': '<div class="executor-event-content"' +
+ | 'data-toggle="tooltip" data-placement="bottom"' +
+ | 'data-title="Executor ${e.id}<br>' +
+ | 'Added at ${UIUtils.formatDate(e.addTime)}"' +
+ | 'data-html="true">Executor ${e.id} added</div>'
+ |}
+ """.stripMargin
+ events += addedEvent
+
+ e.removeTime.foreach { removeTime =>
val removedEvent =
s"""
|{
| 'className': 'executor removed',
| 'group': 'executors',
- | 'start': new Date(${e.time}),
+ | 'start': new Date(${removeTime.getTime()}),
| 'content': '<div class="executor-event-content"' +
| 'data-toggle="tooltip" data-placement="bottom"' +
- | 'data-title="Executor ${e.executorId}<br>' +
- | 'Removed at ${UIUtils.formatDate(new Date(e.time))}' +
+ | 'data-title="Executor ${e.id}<br>' +
+ | 'Removed at ${UIUtils.formatDate(removeTime)}' +
| '${
- if (e.reason != null) {
- s"""<br>Reason: ${e.reason.replace("\n", " ")}"""
- } else {
- ""
- }
+ e.removeReason.map { reason =>
+ s"""<br>Reason: ${reason.replace("\n", " ")}"""
+ }.getOrElse("")
}"' +
- | 'data-html="true">Executor ${e.executorId} removed</div>'
+ | 'data-html="true">Executor ${e.id} removed</div>'
|}
""".stripMargin
events += removedEvent
-
+ }
}
events.toSeq
}
private def makeTimeline(
jobs: Seq[JobUIData],
- executors: Seq[SparkListenerEvent],
+ executors: Seq[ExecutorSummary],
startTime: Long): Seq[Node] = {
val jobEventJsonAsStrSeq = makeJobEvent(jobs)
@@ -360,9 +359,8 @@ private[ui] class AllJobsPage(parent: JobsTab) extends WebUIPage("") {
</div>
var content = summary
- val executorListener = parent.executorListener
content ++= makeTimeline(activeJobs ++ completedJobs ++ failedJobs,
- executorListener.executorEvents, startTime)
+ parent.parent.store.executorList(false), startTime)
if (shouldShowActiveJobs) {
content ++= <h4 id="active">Active Jobs ({activeJobs.size})</h4> ++
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/ui/jobs/ExecutorTable.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/ExecutorTable.scala b/core/src/main/scala/org/apache/spark/ui/jobs/ExecutorTable.scala
index 382a6f9..07a41d1 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/ExecutorTable.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/ExecutorTable.scala
@@ -20,12 +20,17 @@ package org.apache.spark.ui.jobs
import scala.collection.mutable
import scala.xml.{Node, Unparsed}
+import org.apache.spark.status.AppStatusStore
import org.apache.spark.ui.{ToolTips, UIUtils}
import org.apache.spark.ui.jobs.UIData.StageUIData
import org.apache.spark.util.Utils
/** Stage summary grouped by executors. */
-private[ui] class ExecutorTable(stageId: Int, stageAttemptId: Int, parent: StagesTab) {
+private[ui] class ExecutorTable(
+ stageId: Int,
+ stageAttemptId: Int,
+ parent: StagesTab,
+ store: AppStatusStore) {
private val listener = parent.progressListener
def toNodeSeq: Seq[Node] = {
@@ -123,9 +128,7 @@ private[ui] class ExecutorTable(stageId: Int, stageAttemptId: Int, parent: Stage
<div style="float: left">{k}</div>
<div style="float: right">
{
- val logs = parent.executorsListener.executorToTaskSummary.get(k)
- .map(_.executorLogs).getOrElse(Map.empty)
- logs.map {
+ store.executorSummary(k).map(_.executorLogs).getOrElse(Map.empty).map {
case (logName, logUrl) => <div><a href={logUrl}>{logName}</a></div>
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala
index 9fb011a..7ed0164 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/JobPage.scala
@@ -27,6 +27,7 @@ import org.apache.commons.lang3.StringEscapeUtils
import org.apache.spark.JobExecutionStatus
import org.apache.spark.scheduler._
+import org.apache.spark.status.api.v1.ExecutorSummary
import org.apache.spark.ui.{ToolTips, UIUtils, WebUIPage}
/** Page showing statistics and stage list for a given job */
@@ -92,55 +93,52 @@ private[ui] class JobPage(parent: JobsTab) extends WebUIPage("job") {
}
}
- def makeExecutorEvent(executorUIDatas: Seq[SparkListenerEvent]): Seq[String] = {
+ def makeExecutorEvent(executors: Seq[ExecutorSummary]): Seq[String] = {
val events = ListBuffer[String]()
- executorUIDatas.foreach {
- case a: SparkListenerExecutorAdded =>
- val addedEvent =
- s"""
- |{
- | 'className': 'executor added',
- | 'group': 'executors',
- | 'start': new Date(${a.time}),
- | 'content': '<div class="executor-event-content"' +
- | 'data-toggle="tooltip" data-placement="bottom"' +
- | 'data-title="Executor ${a.executorId}<br>' +
- | 'Added at ${UIUtils.formatDate(new Date(a.time))}"' +
- | 'data-html="true">Executor ${a.executorId} added</div>'
- |}
- """.stripMargin
- events += addedEvent
+ executors.foreach { e =>
+ val addedEvent =
+ s"""
+ |{
+ | 'className': 'executor added',
+ | 'group': 'executors',
+ | 'start': new Date(${e.addTime.getTime()}),
+ | 'content': '<div class="executor-event-content"' +
+ | 'data-toggle="tooltip" data-placement="bottom"' +
+ | 'data-title="Executor ${e.id}<br>' +
+ | 'Added at ${UIUtils.formatDate(e.addTime)}"' +
+ | 'data-html="true">Executor ${e.id} added</div>'
+ |}
+ """.stripMargin
+ events += addedEvent
- case e: SparkListenerExecutorRemoved =>
+ e.removeTime.foreach { removeTime =>
val removedEvent =
s"""
|{
| 'className': 'executor removed',
| 'group': 'executors',
- | 'start': new Date(${e.time}),
+ | 'start': new Date(${removeTime.getTime()}),
| 'content': '<div class="executor-event-content"' +
| 'data-toggle="tooltip" data-placement="bottom"' +
- | 'data-title="Executor ${e.executorId}<br>' +
- | 'Removed at ${UIUtils.formatDate(new Date(e.time))}' +
+ | 'data-title="Executor ${e.id}<br>' +
+ | 'Removed at ${UIUtils.formatDate(removeTime)}' +
| '${
- if (e.reason != null) {
- s"""<br>Reason: ${e.reason.replace("\n", " ")}"""
- } else {
- ""
- }
+ e.removeReason.map { reason =>
+ s"""<br>Reason: ${reason.replace("\n", " ")}"""
+ }.getOrElse("")
}"' +
- | 'data-html="true">Executor ${e.executorId} removed</div>'
+ | 'data-html="true">Executor ${e.id} removed</div>'
|}
""".stripMargin
events += removedEvent
-
+ }
}
events.toSeq
}
private def makeTimeline(
stages: Seq[StageInfo],
- executors: Seq[SparkListenerEvent],
+ executors: Seq[ExecutorSummary],
appStartTime: Long): Seq[Node] = {
val stageEventJsonAsStrSeq = makeStageEvent(stages)
@@ -322,11 +320,10 @@ private[ui] class JobPage(parent: JobsTab) extends WebUIPage("job") {
var content = summary
val appStartTime = listener.startTime
- val executorListener = parent.executorListener
val operationGraphListener = parent.operationGraphListener
content ++= makeTimeline(activeStages ++ completedStages ++ failedStages,
- executorListener.executorEvents, appStartTime)
+ parent.parent.store.executorList(false), appStartTime)
content ++= UIUtils.showDagVizForJob(
jobId, operationGraphListener.getOperationGraphForJob(jobId))
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala b/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala
index cc17338..81ffe04 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/JobsTab.scala
@@ -23,11 +23,10 @@ import org.apache.spark.scheduler.SchedulingMode
import org.apache.spark.ui.{SparkUI, SparkUITab, UIUtils}
/** Web UI showing progress status of all jobs in the given SparkContext. */
-private[ui] class JobsTab(parent: SparkUI) extends SparkUITab(parent, "jobs") {
+private[ui] class JobsTab(val parent: SparkUI) extends SparkUITab(parent, "jobs") {
val sc = parent.sc
val killEnabled = parent.killEnabled
val jobProgresslistener = parent.jobProgressListener
- val executorListener = parent.executorsListener
val operationGraphListener = parent.operationGraphListener
def isFairScheduler: Boolean =
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala
index 4d80308..3151b8d 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala
@@ -29,18 +29,17 @@ import org.apache.commons.lang3.StringEscapeUtils
import org.apache.spark.SparkConf
import org.apache.spark.executor.TaskMetrics
import org.apache.spark.scheduler.{AccumulableInfo, TaskInfo, TaskLocality}
+import org.apache.spark.status.AppStatusStore
import org.apache.spark.ui._
-import org.apache.spark.ui.exec.ExecutorsListener
import org.apache.spark.ui.jobs.UIData._
import org.apache.spark.util.{Distribution, Utils}
/** Page showing statistics and task list for a given stage */
-private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
+private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends WebUIPage("stage") {
import StagePage._
private val progressListener = parent.progressListener
private val operationGraphListener = parent.operationGraphListener
- private val executorsListener = parent.executorsListener
private val TIMELINE_LEGEND = {
<div class="legend-area">
@@ -304,7 +303,7 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
pageSize = taskPageSize,
sortColumn = taskSortColumn,
desc = taskSortDesc,
- executorsListener = executorsListener
+ store = store
)
(_taskTable, _taskTable.table(page))
} catch {
@@ -563,7 +562,7 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
stripeRowsWithCss = false))
}
- val executorTable = new ExecutorTable(stageId, stageAttemptId, parent)
+ val executorTable = new ExecutorTable(stageId, stageAttemptId, parent, store)
val maybeAccumulableTable: Seq[Node] =
if (hasAccumulators) { <h4>Accumulators</h4> ++ accumulableTable } else Seq.empty
@@ -869,7 +868,7 @@ private[ui] class TaskDataSource(
pageSize: Int,
sortColumn: String,
desc: Boolean,
- executorsListener: ExecutorsListener) extends PagedDataSource[TaskTableRowData](pageSize) {
+ store: AppStatusStore) extends PagedDataSource[TaskTableRowData](pageSize) {
import StagePage._
// Convert TaskUIData to TaskTableRowData which contains the final contents to show in the table
@@ -1012,8 +1011,7 @@ private[ui] class TaskDataSource(
None
}
- val logs = executorsListener.executorToTaskSummary.get(info.executorId)
- .map(_.executorLogs).getOrElse(Map.empty)
+ val logs = store.executorSummary(info.executorId).map(_.executorLogs).getOrElse(Map.empty)
new TaskTableRowData(
info.index,
info.taskId,
@@ -1162,7 +1160,7 @@ private[ui] class TaskPagedTable(
pageSize: Int,
sortColumn: String,
desc: Boolean,
- executorsListener: ExecutorsListener) extends PagedTable[TaskTableRowData] {
+ store: AppStatusStore) extends PagedTable[TaskTableRowData] {
override def tableId: String = "task-table"
@@ -1188,7 +1186,7 @@ private[ui] class TaskPagedTable(
pageSize,
sortColumn,
desc,
- executorsListener)
+ store)
override def pageLink(page: Int): String = {
val encodedSortColumn = URLEncoder.encode(sortColumn, "UTF-8")
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/main/scala/org/apache/spark/ui/jobs/StagesTab.scala
----------------------------------------------------------------------
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 0787ea6..65446f9 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
@@ -20,20 +20,22 @@ package org.apache.spark.ui.jobs
import javax.servlet.http.HttpServletRequest
import org.apache.spark.scheduler.SchedulingMode
+import org.apache.spark.status.AppStatusStore
import org.apache.spark.ui.{SparkUI, SparkUITab, UIUtils}
/** Web UI showing progress status of all stages in the given SparkContext. */
-private[ui] class StagesTab(parent: SparkUI) extends SparkUITab(parent, "stages") {
+private[ui] class StagesTab(val parent: SparkUI, 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 executorsListener = parent.executorsListener
val lastUpdateTime = parent.lastUpdateTime
attachPage(new AllStagesPage(this))
- attachPage(new StagePage(this))
+ attachPage(new StagePage(this, store))
attachPage(new PoolPage(this))
def isFairScheduler: Boolean = progressListener.schedulingMode == Some(SchedulingMode.FAIR)
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/test/resources/HistoryServerExpectations/executor_list_json_expectation.json
----------------------------------------------------------------------
diff --git a/core/src/test/resources/HistoryServerExpectations/executor_list_json_expectation.json b/core/src/test/resources/HistoryServerExpectations/executor_list_json_expectation.json
index 6b9f29e..942e6d8 100644
--- a/core/src/test/resources/HistoryServerExpectations/executor_list_json_expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/executor_list_json_expectation.json
@@ -18,5 +18,6 @@
"totalShuffleWrite" : 13180,
"isBlacklisted" : false,
"maxMemory" : 278302556,
+ "addTime" : "2015-02-03T16:43:00.906GMT",
"executorLogs" : { }
} ]
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/test/resources/HistoryServerExpectations/executor_memory_usage_expectation.json
----------------------------------------------------------------------
diff --git a/core/src/test/resources/HistoryServerExpectations/executor_memory_usage_expectation.json b/core/src/test/resources/HistoryServerExpectations/executor_memory_usage_expectation.json
index 0f94e3b..ed33c90 100644
--- a/core/src/test/resources/HistoryServerExpectations/executor_memory_usage_expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/executor_memory_usage_expectation.json
@@ -1,6 +1,34 @@
[ {
- "id" : "2",
- "hostPort" : "172.22.0.167:51487",
+ "id" : "driver",
+ "hostPort" : "172.22.0.167:51475",
+ "isActive" : true,
+ "rddBlocks" : 0,
+ "memoryUsed" : 0,
+ "diskUsed" : 0,
+ "totalCores" : 0,
+ "maxTasks" : 0,
+ "activeTasks" : 0,
+ "failedTasks" : 0,
+ "completedTasks" : 0,
+ "totalTasks" : 0,
+ "totalDuration" : 0,
+ "totalGCTime" : 0,
+ "totalInputBytes" : 0,
+ "totalShuffleRead" : 0,
+ "totalShuffleWrite" : 0,
+ "isBlacklisted" : true,
+ "maxMemory" : 908381388,
+ "addTime" : "2016-11-16T22:33:31.477GMT",
+ "executorLogs" : { },
+ "memoryMetrics" : {
+ "usedOnHeapStorageMemory" : 0,
+ "usedOffHeapStorageMemory" : 0,
+ "totalOnHeapStorageMemory" : 384093388,
+ "totalOffHeapStorageMemory" : 524288000
+ }
+}, {
+ "id" : "3",
+ "hostPort" : "172.22.0.167:51485",
"isActive" : true,
"rddBlocks" : 0,
"memoryUsed" : 0,
@@ -8,52 +36,57 @@
"totalCores" : 4,
"maxTasks" : 4,
"activeTasks" : 0,
- "failedTasks" : 4,
- "completedTasks" : 0,
- "totalTasks" : 4,
- "totalDuration" : 2537,
- "totalGCTime" : 88,
+ "failedTasks" : 0,
+ "completedTasks" : 12,
+ "totalTasks" : 12,
+ "totalDuration" : 2453,
+ "totalGCTime" : 72,
"totalInputBytes" : 0,
"totalShuffleRead" : 0,
"totalShuffleWrite" : 0,
"isBlacklisted" : true,
"maxMemory" : 908381388,
+ "addTime" : "2016-11-16T22:33:35.320GMT",
"executorLogs" : {
- "stdout" : "http://172.22.0.167:51469/logPage/?appId=app-20161116163331-0000&executorId=2&logType=stdout",
- "stderr" : "http://172.22.0.167:51469/logPage/?appId=app-20161116163331-0000&executorId=2&logType=stderr"
+ "stdout" : "http://172.22.0.167:51466/logPage/?appId=app-20161116163331-0000&executorId=3&logType=stdout",
+ "stderr" : "http://172.22.0.167:51466/logPage/?appId=app-20161116163331-0000&executorId=3&logType=stderr"
},
"memoryMetrics": {
- "usedOnHeapStorageMemory": 0,
- "usedOffHeapStorageMemory": 0,
- "totalOnHeapStorageMemory": 384093388,
- "totalOffHeapStorageMemory": 524288000
+ "usedOnHeapStorageMemory" : 0,
+ "usedOffHeapStorageMemory" : 0,
+ "totalOnHeapStorageMemory" : 384093388,
+ "totalOffHeapStorageMemory" : 524288000
}
-}, {
- "id" : "driver",
- "hostPort" : "172.22.0.167:51475",
+} ,{
+ "id" : "2",
+ "hostPort" : "172.22.0.167:51487",
"isActive" : true,
"rddBlocks" : 0,
"memoryUsed" : 0,
"diskUsed" : 0,
- "totalCores" : 0,
- "maxTasks" : 0,
+ "totalCores" : 4,
+ "maxTasks" : 4,
"activeTasks" : 0,
- "failedTasks" : 0,
+ "failedTasks" : 4,
"completedTasks" : 0,
- "totalTasks" : 0,
- "totalDuration" : 0,
- "totalGCTime" : 0,
+ "totalTasks" : 4,
+ "totalDuration" : 2537,
+ "totalGCTime" : 88,
"totalInputBytes" : 0,
"totalShuffleRead" : 0,
"totalShuffleWrite" : 0,
"isBlacklisted" : true,
"maxMemory" : 908381388,
- "executorLogs" : { },
+ "addTime" : "2016-11-16T22:33:35.393GMT",
+ "executorLogs" : {
+ "stdout" : "http://172.22.0.167:51469/logPage/?appId=app-20161116163331-0000&executorId=2&logType=stdout",
+ "stderr" : "http://172.22.0.167:51469/logPage/?appId=app-20161116163331-0000&executorId=2&logType=stderr"
+ },
"memoryMetrics": {
- "usedOnHeapStorageMemory": 0,
- "usedOffHeapStorageMemory": 0,
- "totalOnHeapStorageMemory": 384093388,
- "totalOffHeapStorageMemory": 524288000
+ "usedOnHeapStorageMemory" : 0,
+ "usedOffHeapStorageMemory" : 0,
+ "totalOnHeapStorageMemory" : 384093388,
+ "totalOffHeapStorageMemory" : 524288000
}
}, {
"id" : "1",
@@ -75,6 +108,7 @@
"totalShuffleWrite" : 0,
"isBlacklisted" : true,
"maxMemory" : 908381388,
+ "addTime" : "2016-11-16T22:33:35.443GMT",
"executorLogs" : {
"stdout" : "http://172.22.0.167:51467/logPage/?appId=app-20161116163331-0000&executorId=1&logType=stdout",
"stderr" : "http://172.22.0.167:51467/logPage/?appId=app-20161116163331-0000&executorId=1&logType=stderr"
@@ -105,44 +139,15 @@
"totalShuffleWrite" : 0,
"isBlacklisted" : true,
"maxMemory" : 908381388,
+ "addTime" : "2016-11-16T22:33:35.462GMT",
"executorLogs" : {
"stdout" : "http://172.22.0.167:51465/logPage/?appId=app-20161116163331-0000&executorId=0&logType=stdout",
"stderr" : "http://172.22.0.167:51465/logPage/?appId=app-20161116163331-0000&executorId=0&logType=stderr"
},
"memoryMetrics": {
- "usedOnHeapStorageMemory": 0,
- "usedOffHeapStorageMemory": 0,
- "totalOnHeapStorageMemory": 384093388,
- "totalOffHeapStorageMemory": 524288000
- }
-}, {
- "id" : "3",
- "hostPort" : "172.22.0.167:51485",
- "isActive" : true,
- "rddBlocks" : 0,
- "memoryUsed" : 0,
- "diskUsed" : 0,
- "totalCores" : 4,
- "maxTasks" : 4,
- "activeTasks" : 0,
- "failedTasks" : 0,
- "completedTasks" : 12,
- "totalTasks" : 12,
- "totalDuration" : 2453,
- "totalGCTime" : 72,
- "totalInputBytes" : 0,
- "totalShuffleRead" : 0,
- "totalShuffleWrite" : 0,
- "isBlacklisted" : true,
- "maxMemory" : 908381388,
- "executorLogs" : {
- "stdout" : "http://172.22.0.167:51466/logPage/?appId=app-20161116163331-0000&executorId=3&logType=stdout",
- "stderr" : "http://172.22.0.167:51466/logPage/?appId=app-20161116163331-0000&executorId=3&logType=stderr"
- },
- "memoryMetrics": {
- "usedOnHeapStorageMemory": 0,
- "usedOffHeapStorageMemory": 0,
- "totalOnHeapStorageMemory": 384093388,
- "totalOffHeapStorageMemory": 524288000
+ "usedOnHeapStorageMemory" : 0,
+ "usedOffHeapStorageMemory" : 0,
+ "totalOnHeapStorageMemory" : 384093388,
+ "totalOffHeapStorageMemory" : 524288000
}
} ]
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/test/resources/HistoryServerExpectations/executor_node_blacklisting_expectation.json
----------------------------------------------------------------------
diff --git a/core/src/test/resources/HistoryServerExpectations/executor_node_blacklisting_expectation.json b/core/src/test/resources/HistoryServerExpectations/executor_node_blacklisting_expectation.json
index 0f94e3b..73519f1 100644
--- a/core/src/test/resources/HistoryServerExpectations/executor_node_blacklisting_expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/executor_node_blacklisting_expectation.json
@@ -1,6 +1,34 @@
[ {
- "id" : "2",
- "hostPort" : "172.22.0.167:51487",
+ "id" : "driver",
+ "hostPort" : "172.22.0.167:51475",
+ "isActive" : true,
+ "rddBlocks" : 0,
+ "memoryUsed" : 0,
+ "diskUsed" : 0,
+ "totalCores" : 0,
+ "maxTasks" : 0,
+ "activeTasks" : 0,
+ "failedTasks" : 0,
+ "completedTasks" : 0,
+ "totalTasks" : 0,
+ "totalDuration" : 0,
+ "totalGCTime" : 0,
+ "totalInputBytes" : 0,
+ "totalShuffleRead" : 0,
+ "totalShuffleWrite" : 0,
+ "isBlacklisted" : true,
+ "maxMemory" : 908381388,
+ "addTime" : "2016-11-16T22:33:31.477GMT",
+ "executorLogs" : { },
+ "memoryMetrics": {
+ "usedOnHeapStorageMemory" : 0,
+ "usedOffHeapStorageMemory" : 0,
+ "totalOnHeapStorageMemory" : 384093388,
+ "totalOffHeapStorageMemory" : 524288000
+ }
+}, {
+ "id" : "3",
+ "hostPort" : "172.22.0.167:51485",
"isActive" : true,
"rddBlocks" : 0,
"memoryUsed" : 0,
@@ -8,52 +36,57 @@
"totalCores" : 4,
"maxTasks" : 4,
"activeTasks" : 0,
- "failedTasks" : 4,
- "completedTasks" : 0,
- "totalTasks" : 4,
- "totalDuration" : 2537,
- "totalGCTime" : 88,
+ "failedTasks" : 0,
+ "completedTasks" : 12,
+ "totalTasks" : 12,
+ "totalDuration" : 2453,
+ "totalGCTime" : 72,
"totalInputBytes" : 0,
"totalShuffleRead" : 0,
"totalShuffleWrite" : 0,
"isBlacklisted" : true,
"maxMemory" : 908381388,
+ "addTime" : "2016-11-16T22:33:35.320GMT",
"executorLogs" : {
- "stdout" : "http://172.22.0.167:51469/logPage/?appId=app-20161116163331-0000&executorId=2&logType=stdout",
- "stderr" : "http://172.22.0.167:51469/logPage/?appId=app-20161116163331-0000&executorId=2&logType=stderr"
+ "stdout" : "http://172.22.0.167:51466/logPage/?appId=app-20161116163331-0000&executorId=3&logType=stdout",
+ "stderr" : "http://172.22.0.167:51466/logPage/?appId=app-20161116163331-0000&executorId=3&logType=stderr"
},
"memoryMetrics": {
- "usedOnHeapStorageMemory": 0,
- "usedOffHeapStorageMemory": 0,
- "totalOnHeapStorageMemory": 384093388,
- "totalOffHeapStorageMemory": 524288000
+ "usedOnHeapStorageMemory" : 0,
+ "usedOffHeapStorageMemory" : 0,
+ "totalOnHeapStorageMemory" : 384093388,
+ "totalOffHeapStorageMemory" : 524288000
}
}, {
- "id" : "driver",
- "hostPort" : "172.22.0.167:51475",
+ "id" : "2",
+ "hostPort" : "172.22.0.167:51487",
"isActive" : true,
"rddBlocks" : 0,
"memoryUsed" : 0,
"diskUsed" : 0,
- "totalCores" : 0,
- "maxTasks" : 0,
+ "totalCores" : 4,
+ "maxTasks" : 4,
"activeTasks" : 0,
- "failedTasks" : 0,
+ "failedTasks" : 4,
"completedTasks" : 0,
- "totalTasks" : 0,
- "totalDuration" : 0,
- "totalGCTime" : 0,
+ "totalTasks" : 4,
+ "totalDuration" : 2537,
+ "totalGCTime" : 88,
"totalInputBytes" : 0,
"totalShuffleRead" : 0,
"totalShuffleWrite" : 0,
"isBlacklisted" : true,
"maxMemory" : 908381388,
- "executorLogs" : { },
+ "addTime" : "2016-11-16T22:33:35.393GMT",
+ "executorLogs" : {
+ "stdout" : "http://172.22.0.167:51469/logPage/?appId=app-20161116163331-0000&executorId=2&logType=stdout",
+ "stderr" : "http://172.22.0.167:51469/logPage/?appId=app-20161116163331-0000&executorId=2&logType=stderr"
+ },
"memoryMetrics": {
- "usedOnHeapStorageMemory": 0,
- "usedOffHeapStorageMemory": 0,
- "totalOnHeapStorageMemory": 384093388,
- "totalOffHeapStorageMemory": 524288000
+ "usedOnHeapStorageMemory" : 0,
+ "usedOffHeapStorageMemory" : 0,
+ "totalOnHeapStorageMemory" : 384093388,
+ "totalOffHeapStorageMemory" : 524288000
}
}, {
"id" : "1",
@@ -75,6 +108,7 @@
"totalShuffleWrite" : 0,
"isBlacklisted" : true,
"maxMemory" : 908381388,
+ "addTime" : "2016-11-16T22:33:35.443GMT",
"executorLogs" : {
"stdout" : "http://172.22.0.167:51467/logPage/?appId=app-20161116163331-0000&executorId=1&logType=stdout",
"stderr" : "http://172.22.0.167:51467/logPage/?appId=app-20161116163331-0000&executorId=1&logType=stderr"
@@ -105,6 +139,7 @@
"totalShuffleWrite" : 0,
"isBlacklisted" : true,
"maxMemory" : 908381388,
+ "addTime" : "2016-11-16T22:33:35.462GMT",
"executorLogs" : {
"stdout" : "http://172.22.0.167:51465/logPage/?appId=app-20161116163331-0000&executorId=0&logType=stdout",
"stderr" : "http://172.22.0.167:51465/logPage/?appId=app-20161116163331-0000&executorId=0&logType=stderr"
@@ -115,34 +150,4 @@
"totalOnHeapStorageMemory": 384093388,
"totalOffHeapStorageMemory": 524288000
}
-}, {
- "id" : "3",
- "hostPort" : "172.22.0.167:51485",
- "isActive" : true,
- "rddBlocks" : 0,
- "memoryUsed" : 0,
- "diskUsed" : 0,
- "totalCores" : 4,
- "maxTasks" : 4,
- "activeTasks" : 0,
- "failedTasks" : 0,
- "completedTasks" : 12,
- "totalTasks" : 12,
- "totalDuration" : 2453,
- "totalGCTime" : 72,
- "totalInputBytes" : 0,
- "totalShuffleRead" : 0,
- "totalShuffleWrite" : 0,
- "isBlacklisted" : true,
- "maxMemory" : 908381388,
- "executorLogs" : {
- "stdout" : "http://172.22.0.167:51466/logPage/?appId=app-20161116163331-0000&executorId=3&logType=stdout",
- "stderr" : "http://172.22.0.167:51466/logPage/?appId=app-20161116163331-0000&executorId=3&logType=stderr"
- },
- "memoryMetrics": {
- "usedOnHeapStorageMemory": 0,
- "usedOffHeapStorageMemory": 0,
- "totalOnHeapStorageMemory": 384093388,
- "totalOffHeapStorageMemory": 524288000
- }
} ]
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/test/resources/HistoryServerExpectations/executor_node_blacklisting_unblacklisting_expectation.json
----------------------------------------------------------------------
diff --git a/core/src/test/resources/HistoryServerExpectations/executor_node_blacklisting_unblacklisting_expectation.json b/core/src/test/resources/HistoryServerExpectations/executor_node_blacklisting_unblacklisting_expectation.json
index 92e249c..6931fea 100644
--- a/core/src/test/resources/HistoryServerExpectations/executor_node_blacklisting_unblacklisting_expectation.json
+++ b/core/src/test/resources/HistoryServerExpectations/executor_node_blacklisting_unblacklisting_expectation.json
@@ -1,6 +1,28 @@
[ {
- "id" : "2",
- "hostPort" : "172.22.0.111:64539",
+ "id" : "driver",
+ "hostPort" : "172.22.0.111:64527",
+ "isActive" : true,
+ "rddBlocks" : 0,
+ "memoryUsed" : 0,
+ "diskUsed" : 0,
+ "totalCores" : 0,
+ "maxTasks" : 0,
+ "activeTasks" : 0,
+ "failedTasks" : 0,
+ "completedTasks" : 0,
+ "totalTasks" : 0,
+ "totalDuration" : 0,
+ "totalGCTime" : 0,
+ "totalInputBytes" : 0,
+ "totalShuffleRead" : 0,
+ "totalShuffleWrite" : 0,
+ "isBlacklisted" : false,
+ "maxMemory" : 384093388,
+ "addTime" : "2016-11-15T23:20:38.836GMT",
+ "executorLogs" : { }
+}, {
+ "id" : "3",
+ "hostPort" : "172.22.0.111:64543",
"isActive" : true,
"rddBlocks" : 0,
"memoryUsed" : 0,
@@ -8,41 +30,46 @@
"totalCores" : 4,
"maxTasks" : 4,
"activeTasks" : 0,
- "failedTasks" : 6,
- "completedTasks" : 0,
- "totalTasks" : 6,
- "totalDuration" : 2792,
- "totalGCTime" : 128,
+ "failedTasks" : 0,
+ "completedTasks" : 4,
+ "totalTasks" : 4,
+ "totalDuration" : 3457,
+ "totalGCTime" : 72,
"totalInputBytes" : 0,
"totalShuffleRead" : 0,
"totalShuffleWrite" : 0,
"isBlacklisted" : false,
"maxMemory" : 384093388,
+ "addTime" : "2016-11-15T23:20:42.711GMT",
"executorLogs" : {
- "stdout" : "http://172.22.0.111:64519/logPage/?appId=app-20161115172038-0000&executorId=2&logType=stdout",
- "stderr" : "http://172.22.0.111:64519/logPage/?appId=app-20161115172038-0000&executorId=2&logType=stderr"
+ "stdout" : "http://172.22.0.111:64521/logPage/?appId=app-20161115172038-0000&executorId=3&logType=stdout",
+ "stderr" : "http://172.22.0.111:64521/logPage/?appId=app-20161115172038-0000&executorId=3&logType=stderr"
}
}, {
- "id" : "driver",
- "hostPort" : "172.22.0.111:64527",
+ "id" : "2",
+ "hostPort" : "172.22.0.111:64539",
"isActive" : true,
"rddBlocks" : 0,
"memoryUsed" : 0,
"diskUsed" : 0,
- "totalCores" : 0,
- "maxTasks" : 0,
+ "totalCores" : 4,
+ "maxTasks" : 4,
"activeTasks" : 0,
- "failedTasks" : 0,
+ "failedTasks" : 6,
"completedTasks" : 0,
- "totalTasks" : 0,
- "totalDuration" : 0,
- "totalGCTime" : 0,
+ "totalTasks" : 6,
+ "totalDuration" : 2792,
+ "totalGCTime" : 128,
"totalInputBytes" : 0,
"totalShuffleRead" : 0,
"totalShuffleWrite" : 0,
"isBlacklisted" : false,
"maxMemory" : 384093388,
- "executorLogs" : { }
+ "addTime" : "2016-11-15T23:20:42.589GMT",
+ "executorLogs" : {
+ "stdout" : "http://172.22.0.111:64519/logPage/?appId=app-20161115172038-0000&executorId=2&logType=stdout",
+ "stderr" : "http://172.22.0.111:64519/logPage/?appId=app-20161115172038-0000&executorId=2&logType=stderr"
+ }
}, {
"id" : "1",
"hostPort" : "172.22.0.111:64541",
@@ -63,6 +90,7 @@
"totalShuffleWrite" : 0,
"isBlacklisted" : false,
"maxMemory" : 384093388,
+ "addTime" : "2016-11-15T23:20:42.629GMT",
"executorLogs" : {
"stdout" : "http://172.22.0.111:64518/logPage/?appId=app-20161115172038-0000&executorId=1&logType=stdout",
"stderr" : "http://172.22.0.111:64518/logPage/?appId=app-20161115172038-0000&executorId=1&logType=stderr"
@@ -87,32 +115,9 @@
"totalShuffleWrite" : 0,
"isBlacklisted" : false,
"maxMemory" : 384093388,
+ "addTime" : "2016-11-15T23:20:42.593GMT",
"executorLogs" : {
"stdout" : "http://172.22.0.111:64517/logPage/?appId=app-20161115172038-0000&executorId=0&logType=stdout",
"stderr" : "http://172.22.0.111:64517/logPage/?appId=app-20161115172038-0000&executorId=0&logType=stderr"
}
-}, {
- "id" : "3",
- "hostPort" : "172.22.0.111:64543",
- "isActive" : true,
- "rddBlocks" : 0,
- "memoryUsed" : 0,
- "diskUsed" : 0,
- "totalCores" : 4,
- "maxTasks" : 4,
- "activeTasks" : 0,
- "failedTasks" : 0,
- "completedTasks" : 4,
- "totalTasks" : 4,
- "totalDuration" : 3457,
- "totalGCTime" : 72,
- "totalInputBytes" : 0,
- "totalShuffleRead" : 0,
- "totalShuffleWrite" : 0,
- "isBlacklisted" : false,
- "maxMemory" : 384093388,
- "executorLogs" : {
- "stdout" : "http://172.22.0.111:64521/logPage/?appId=app-20161115172038-0000&executorId=3&logType=stdout",
- "stderr" : "http://172.22.0.111:64521/logPage/?appId=app-20161115172038-0000&executorId=3&logType=stderr"
- }
} ]
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala b/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala
index 499d47b..1c51c14 100644
--- a/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/StagePageSuite.scala
@@ -22,13 +22,13 @@ import javax.servlet.http.HttpServletRequest
import scala.xml.Node
+import org.mockito.Matchers.anyString
import org.mockito.Mockito.{mock, when, RETURNS_SMART_NULLS}
import org.apache.spark._
import org.apache.spark.executor.TaskMetrics
import org.apache.spark.scheduler._
-import org.apache.spark.storage.StorageStatusListener
-import org.apache.spark.ui.exec.ExecutorsListener
+import org.apache.spark.status.AppStatusStore
import org.apache.spark.ui.jobs.{JobProgressListener, StagePage, StagesTab}
import org.apache.spark.ui.scope.RDDOperationGraphListener
@@ -55,20 +55,21 @@ class StagePageSuite extends SparkFunSuite with LocalSparkContext {
* This also runs a dummy stage to populate the page with useful content.
*/
private def renderStagePage(conf: SparkConf): Seq[Node] = {
+ val store = mock(classOf[AppStatusStore])
+ when(store.executorSummary(anyString())).thenReturn(None)
+
val jobListener = new JobProgressListener(conf)
val graphListener = new RDDOperationGraphListener(conf)
- val executorsListener = new ExecutorsListener(new StorageStatusListener(conf), conf)
val tab = mock(classOf[StagesTab], RETURNS_SMART_NULLS)
val request = mock(classOf[HttpServletRequest])
when(tab.conf).thenReturn(conf)
when(tab.progressListener).thenReturn(jobListener)
when(tab.operationGraphListener).thenReturn(graphListener)
- when(tab.executorsListener).thenReturn(executorsListener)
when(tab.appName).thenReturn("testing")
when(tab.headerTabs).thenReturn(Seq.empty)
when(request.getParameter("id")).thenReturn("0")
when(request.getParameter("attempt")).thenReturn("0")
- val page = new StagePage(tab)
+ val page = new StagePage(tab, store)
// Simulate a stage in job progress listener
val stageInfo = new StageInfo(0, 0, "dummy", 1, Seq.empty, Seq.empty, "details")
http://git-wip-us.apache.org/repos/asf/spark/blob/11eea1a4/project/MimaExcludes.scala
----------------------------------------------------------------------
diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala
index 62930e2..0c31b2b 100644
--- a/project/MimaExcludes.scala
+++ b/project/MimaExcludes.scala
@@ -40,6 +40,7 @@ object MimaExcludes {
ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.spark.status.api.v1.ExecutorSummary.executorLogs"),
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.deploy.history.HistoryServer.getSparkUI"),
ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.ui.env.EnvironmentListener"),
+ ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.ui.exec.ExecutorsListener"),
// [SPARK-20495][SQL] Add StorageLevel to cacheTable API
ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.catalog.Catalog.cacheTable"),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org