[SPARK-3984] [SPARK-3983] Improve UI task metrics.

This commit fixes the scheduler delay in the UI (which previously
included things that are not scheduler delay, like time to
deserialize the task and serialize the result), and also
adds finer-grained information to the summary table for each
stage about task launch overhead (which is useful for debugging
performance of short jobs, where the overhead is not-insignificant).

Author: kayousterhout

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -110,10 +110,15 @@ private[spark] class Executor(
   // Maintains the list of running tasks.
   private val runningTasks = new ConcurrentHashMap[Long, TaskRunner]
 
+  // Time when the task arrived on the executor. Used to track the overhead of getting a thread for
+  // the task to run in.
+  private val taskStartTimes = new ConcurrentHashMap[Long, Long]
+
   startDriverHeartbeater()
 
   def launchTask(
       context: ExecutorBackend, taskId: Long, taskName: String, serializedTask: ByteBuffer) {
+    taskStartTimes.put(taskId, System.currentTimeMillis)
     val tr = new TaskRunner(context, taskId, taskName, serializedTask)
     runningTasks.put(taskId, tr)
     threadPool.execute(tr)
@@ -152,7 +157,7 @@ private[spark] class Executor(
     }
 
     override def run() {
-      val startTime = System.currentTimeMillis()
+      val deserializeStartTime = System.currentTimeMillis()
       Thread.currentThread.setContextClassLoader(replClassLoader)
       val ser = SparkEnv.get.closureSerializer.newInstance()
       logInfo(s"Running $taskName (TID $taskId)")
@@ -197,7 +202,8 @@ private[spark] class Executor(
         val afterSerialization = System.currentTimeMillis()
 
         for (m <- task.metrics) {
-          m.executorDeserializeTime = taskStart - startTime
+          m.executorLaunchTime = deserializeStartTime - taskStartTimes.get(taskId)
+          m.executorDeserializeTime = taskStart - deserializeStartTime
           m.executorRunTime = taskFinish - taskStart
           m.jvmGCTime = gcTime - startGCTime
           m.resultSerializationTime = afterSerialization - beforeSerialization
@@ -267,6 +273,7 @@ private[spark] class Executor(
         // Release memory used by this thread for unrolling blocks
         env.blockManager.memoryStore.releaseUnrollMemoryForThisThread()
         runningTasks.remove(taskId)
+        taskStartTimes.remove(taskId)
       }
     }
   }

core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala

@@ -41,6 +41,11 @@ class TaskMetrics extends Serializable {
    */
   var hostname: String = _
 
+  /**
+   * Time taken on the executor to launch the task in its own thread.
+   */
+  var executorLaunchTime: Long = _
+
   /**
    * Time taken on the executor to deserialize this task
    */

core/src/main/scala/org/apache/spark/ui/ToolTips.scala

@@ -18,6 +18,9 @@
 package org.apache.spark.ui
 
 private[spark] object ToolTips {
+  val EXECUTOR_LAUNCH_TIME =
+    """Overhead associated with launching the task in its own thread on the executor."""
+
   val SCHEDULER_DELAY =
     """Scheduler delay includes time to ship the task from the scheduler to
        the executor, and time to send the task result from the executor to the scheduler. If

core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala

@@ -179,6 +179,20 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
             }
           }
 
+          val executorLaunchTimes = validTasks.map { case TaskUIData(_, metrics, _) =>
+            metrics.get.executorLaunchTime.toDouble
+          }
+          val executorLaunchTitle = <td><span data-toggle="tooltip"
+            title={ToolTips.EXECUTOR_LAUNCH_TIME} data-placement="right">Launch time</span></td>
+          val executorLaunchQuantiles =
+            executorLaunchTitle +: getFormattedTimeQuantiles(executorLaunchTimes)
+
+          val deserializationTimes = validTasks.map { case TaskUIData(_, metrics, _) =>
+            metrics.get.executorDeserializeTime.toDouble
+          }
+          val deserializationQuantiles =
+            <td>Task deserialization time</td> +: getFormattedTimeQuantiles(deserializationTimes)
+
           val serviceTimes = validTasks.map { case TaskUIData(_, metrics, _) =>
             metrics.get.executorRunTime.toDouble
           }
@@ -248,6 +262,7 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
           val shuffleWriteSizes = validTasks.map { case TaskUIData(_, metrics, _) =>
             metrics.get.shuffleWriteMetrics.map(_.shuffleBytesWritten).getOrElse(0L).toDouble
           }
+
           val shuffleWriteQuantiles = <td>Shuffle Write</td> +:
             getFormattedSizeQuantiles(shuffleWriteSizes)
 
@@ -266,6 +281,10 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
           val listings: Seq[Seq[Node]] = Seq(
             <tr>{serviceQuantiles}</tr>,
             <tr class={TaskDetailsClassNames.SCHEDULER_DELAY}>{schedulerDelayQuantiles}</tr>,
+            <tr class={TaskDetailsClassNames.EXECUTOR_LAUNCH_TIME}>{executorLaunchQuantiles}</tr>
+            <tr class={TaskDetailsClassNames.TASK_DESERIALIZATION_TIME}>
+              {deserializationQuantiles}
+            </tr>
             <tr class={TaskDetailsClassNames.GC_TIME}>{gcQuantiles}</tr>,
             <tr class={TaskDetailsClassNames.RESULT_SERIALIZATION_TIME}>
               {serializationQuantiles}
@@ -424,6 +443,9 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
         (info.finishTime - info.launchTime)
       }
     }
-    totalExecutionTime - metrics.executorRunTime
+    val executorOverhead = (metrics.executorLaunchTime +
+      metrics.executorDeserializeTime +
+      metrics.resultSerializationTime)
+    totalExecutionTime - metrics.executorRunTime - executorOverhead
   }
 }

core/src/main/scala/org/apache/spark/ui/jobs/TaskDetailsClassNames.scala

@@ -24,6 +24,8 @@ package org.apache.spark.ui.jobs
 private object TaskDetailsClassNames {
   val SCHEDULER_DELAY = "scheduler_delay"
   val GC_TIME = "gc_time"
+  val EXECUTOR_LAUNCH_TIME = "launch_time"
+  val TASK_DESERIALIZATION_TIME = "deserialization_time"
   val RESULT_SERIALIZATION_TIME = "serialization_time"
   val GETTING_RESULT_TIME = "getting_result_time"
 }

core/src/main/scala/org/apache/spark/util/JsonProtocol.scala

@@ -240,6 +240,7 @@ private[spark] object JsonProtocol {
         })
       }.getOrElse(JNothing)
     ("Host Name" -> taskMetrics.hostname) ~
+    ("Executor Launch Time" -> taskMetrics.executorLaunchTime) ~
     ("Executor Deserialize Time" -> taskMetrics.executorDeserializeTime) ~
     ("Executor Run Time" -> taskMetrics.executorRunTime) ~
     ("Result Size" -> taskMetrics.resultSize) ~
@@ -562,6 +563,7 @@ private[spark] object JsonProtocol {
     }
     val metrics = new TaskMetrics
     metrics.hostname = (json \ "Host Name").extract[String]
+    metrics.executorLaunchTime = (json \ "Executor Launch Time").extractOpt[Long].getOrElse(0)
     metrics.executorDeserializeTime = (json \ "Executor Deserialize Time").extract[Long]
     metrics.executorRunTime = (json \ "Executor Run Time").extract[Long]
     metrics.resultSize = (json \ "Result Size").extract[Long]

core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala

@@ -154,6 +154,15 @@ class JsonProtocolSuite extends FunSuite {
     assert(newMetrics.inputMetrics.isEmpty)
   }
 
+  test("TaskMetrics.executorLaunchTime backward compatibility") {
+    // executorLaunchTime was added after 1.1.
+    val metrics = makeTaskMetrics(1L, 2L, 3L, 4L, 5, 6, hasHadoopInput = true)
+    val newJson = JsonProtocol.taskMetricsToJson(metrics)
+    val oldJson = newJson.removeField { case (field, _) => field == "Executor Launch Time" }
+    val newMetrics = JsonProtocol.taskMetricsFromJson(oldJson)
+    assert(newMetrics.executorLaunchTime === 0L)
+  }
+
   test("BlockManager events backward compatibility") {
     // SparkListenerBlockManagerAdded/Removed in Spark 1.0.0 do not have a "time" property.
     val blockManagerAdded = SparkListenerBlockManagerAdded(1L,
@@ -554,6 +563,7 @@ class JsonProtocolSuite extends FunSuite {
     val t = new TaskMetrics
     val sw = new ShuffleWriteMetrics
     t.hostname = "localhost"
+    t.executorLaunchTime = c + d
     t.executorDeserializeTime = a
     t.executorRunTime = b
     t.resultSize = c
@@ -796,6 +806,7 @@ class JsonProtocolSuite extends FunSuite {
       |  },
       |  "Task Metrics": {
       |    "Host Name": "localhost",
+      |    "Executor Launch Time": 1100,
       |    "Executor Deserialize Time": 300,
       |    "Executor Run Time": 400,
       |    "Result Size": 500,
@@ -879,6 +890,7 @@ class JsonProtocolSuite extends FunSuite {
       |  },
       |  "Task Metrics": {
       |    "Host Name": "localhost",
+      |    "Executor Launch Time": 1100,
       |    "Executor Deserialize Time": 300,
       |    "Executor Run Time": 400,
       |    "Result Size": 500,