[GitHub] spark pull request #14969: [SPARK-17406][WEB UI] limit timeline executor eve...

[zhaorongsheng <gi...@git.apache.org>]

Github user zhaorongsheng commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14969#discussion_r97465466
  
    --- Diff: core/src/main/scala/org/apache/spark/ui/exec/ExecutorsTab.scala ---
    @@ -38,47 +37,67 @@ 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)
     
       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)
    +    }
    +
    +    val deadExecutors = executorToTaskSummary.filter(e => !e._2.isAlive)
    +    if (deadExecutors.size > retainedDeadExecutors) {
    +      val head = deadExecutors.head
    +      executorToTaskSummary.remove(head._1)
    --- End diff --
    
    Here we remove only one elements in each time. So we would remove one element when each new executor is added.
    Could we remove more elements at once time?


---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org