Github user cenyuhai commented on a diff in the pull request:
https://github.com/apache/spark/pull/14969#discussion_r78351884
--- Diff: core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala
---
@@ -38,47 +37,68 @@ private[ui] class ExecutorsTab(parent: SparkUI) extends
SparkUITab(parent, "exec
}
}
+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
+)
+
/**
* :: DeveloperApi ::
* A SparkListener that prepares information to be displayed on the
ExecutorsTab
*/
@DeveloperApi
class ExecutorsListener(storageStatusListener: StorageStatusListener,
conf: SparkConf)
extends SparkListener {
- val executorToTotalCores = HashMap[String, Int]()
- val executorToTasksMax = HashMap[String, Int]()
- val executorToTasksActive = HashMap[String, Int]()
- val executorToTasksComplete = HashMap[String, Int]()
- val executorToTasksFailed = HashMap[String, Int]()
- val executorToDuration = HashMap[String, Long]()
- val executorToJvmGCTime = HashMap[String, Long]()
- val executorToInputBytes = HashMap[String, Long]()
- val executorToInputRecords = HashMap[String, Long]()
- val executorToOutputBytes = HashMap[String, Long]()
- val executorToOutputRecords = HashMap[String, Long]()
- val executorToShuffleRead = HashMap[String, Long]()
- val executorToShuffleWrite = HashMap[String, Long]()
- val executorToLogUrls = HashMap[String, Map[String, String]]()
- val executorIdToData = HashMap[String, ExecutorUIData]()
+ var 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)
+ private var deadExecutorCount = 0
def activeStorageStatusList: Seq[StorageStatus] =
storageStatusListener.storageStatusList
def deadStorageStatusList: Seq[StorageStatus] =
storageStatusListener.deadStorageStatusList
override def onExecutorAdded(executorAdded: SparkListenerExecutorAdded):
Unit = synchronized {
val eid = executorAdded.executorId
- executorToLogUrls(eid) = executorAdded.executorInfo.logUrlMap
- executorToTotalCores(eid) = executorAdded.executorInfo.totalCores
- executorToTasksMax(eid) = executorToTotalCores(eid) /
conf.getInt("spark.task.cpus", 1)
- executorIdToData(eid) = new ExecutorUIData(executorAdded.time)
+ 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)
+ }
+ if (deadExecutorCount > retainedDeadExecutors) {
+ val head = executorToTaskSummary.filter(e => !e._2.isAlive).head
+ executorToTaskSummary.remove(head._1)
+ deadExecutorCount -= 1
+ }
}
override def onExecutorRemoved(
executorRemoved: SparkListenerExecutorRemoved): Unit = synchronized {
- val eid = executorRemoved.executorId
- val uiData = executorIdToData(eid)
- uiData.finishTime = Some(executorRemoved.time)
- uiData.finishReason = Some(executorRemoved.reason)
+ executorEvents += executorRemoved
+ if (executorEvents.size > maxTimelineExecutors) {
+ executorEvents.remove(0)
+ }
+ deadExecutorCount += 1
+ executorToTaskSummary.get(executorRemoved.executorId).map(e =>
e.isAlive = false)
--- End diff --
OK
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
