Removed executor launch time

Author: kayousterhout

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

@@ -110,15 +110,10 @@ 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)
@@ -202,7 +197,6 @@ private[spark] class Executor(
         val afterSerialization = System.currentTimeMillis()
 
         for (m <- task.metrics) {
-          m.executorLaunchTime = deserializeStartTime - taskStartTimes.get(taskId)
           m.executorDeserializeTime = taskStart - deserializeStartTime
           m.executorRunTime = taskFinish - taskStart
           m.jvmGCTime = gcTime - startGCTime
@@ -273,7 +267,6 @@ 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,11 +41,6 @@ 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

@@ -24,9 +24,6 @@ private[spark] object ToolTips {
        scheduler delay is large, consider decreasing the size of tasks or decreasing the size
        of task results."""
 
-  val EXECUTOR_LAUNCH_TIME =
-    """Overhead associated with launching the task in its own thread on the executor."""
-
   val TASK_DESERIALIZATION_TIME =
     """Time spent deserializating the task closure on the executor."""
 

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

@@ -119,13 +119,6 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
                   <span class="additional-metric-title">GC Time</span>
                 </span>
               </li>
-              <li>
-                <span data-toggle="tooltip"
-                      title={ToolTips.EXECUTOR_LAUNCH_TIME} data-placement="right">
-                  <input type="checkbox" name={TaskDetailsClassNames.EXECUTOR_LAUNCH_TIME}/>
-                  <span class="additional-metric-title">Executor Launch Time</span>
-                </span>
-              </li>
               <li>
                 <span data-toggle="tooltip"
                       title={ToolTips.TASK_DESERIALIZATION_TIME} data-placement="right">
@@ -162,7 +155,6 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
           ("Executor ID / Host", ""), ("Launch Time", ""), ("Duration", ""),
           ("Scheduler Delay", TaskDetailsClassNames.SCHEDULER_DELAY),
           ("GC Time", TaskDetailsClassNames.GC_TIME),
-          ("Executor Launch Time", TaskDetailsClassNames.EXECUTOR_LAUNCH_TIME),
           ("Task Deserialization Time", TaskDetailsClassNames.TASK_DESERIALIZATION_TIME),
           ("Result Serialization Time", TaskDetailsClassNames.RESULT_SERIALIZATION_TIME),
           ("Getting Result Time", TaskDetailsClassNames.GETTING_RESULT_TIME)) ++
@@ -195,19 +187,6 @@ 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">
-                Executor Launch Time
-              </span>
-            </td>
-          val executorLaunchQuantiles =
-            executorLaunchTitle +: getFormattedTimeQuantiles(executorLaunchTimes)
-
           val deserializationTimes = validTasks.map { case TaskUIData(_, metrics, _) =>
             metrics.get.executorDeserializeTime.toDouble
           }
@@ -306,7 +285,6 @@ 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>
@@ -358,7 +336,6 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
         else metrics.map(m => UIUtils.formatDuration(m.executorRunTime)).getOrElse("")
       val schedulerDelay = getSchedulerDelay(info, metrics.get)
       val gcTime = metrics.map(_.jvmGCTime).getOrElse(0L)
-      val executorLaunchTime = metrics.map(_.executorLaunchTime).getOrElse(0L)
       val taskDeserializationTime = metrics.map(_.executorDeserializeTime).getOrElse(0L)
       val serializationTime = metrics.map(_.resultSerializationTime).getOrElse(0L)
       val gettingResultTime = info.gettingResultTime
@@ -416,10 +393,6 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
         <td sorttable_customkey={gcTime.toString} class={TaskDetailsClassNames.GC_TIME}>
           {if (gcTime > 0) UIUtils.formatDuration(gcTime) else ""}
         </td>
-        <td sorttable_customkey={executorLaunchTime.toString}
-            class={TaskDetailsClassNames.EXECUTOR_LAUNCH_TIME}>
-          {UIUtils.formatDuration(executorLaunchTime.toLong)}
-        </td>
         <td sorttable_customkey={taskDeserializationTime.toString}
             class={TaskDetailsClassNames.TASK_DESERIALIZATION_TIME}>
           {UIUtils.formatDuration(taskDeserializationTime.toLong)}
@@ -478,8 +451,7 @@ private[ui] class StagePage(parent: JobProgressTab) extends WebUIPage("stage") {
         (info.finishTime - info.launchTime)
       }
     }
-    val executorOverhead = (metrics.executorLaunchTime +
-      metrics.executorDeserializeTime +
+    val executorOverhead = (metrics.executorDeserializeTime +
       metrics.resultSerializationTime)
     totalExecutionTime - metrics.executorRunTime - executorOverhead
   }

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

@@ -24,7 +24,6 @@ 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,7 +240,6 @@ 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) ~
@@ -563,7 +562,6 @@ 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,15 +154,6 @@ 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,
@@ -563,7 +554,6 @@ 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
@@ -806,7 +796,6 @@ class JsonProtocolSuite extends FunSuite {
       |  },
       |  "Task Metrics": {
       |    "Host Name": "localhost",
-      |    "Executor Launch Time": 1100,
       |    "Executor Deserialize Time": 300,
       |    "Executor Run Time": 400,
       |    "Result Size": 500,
@@ -890,7 +879,6 @@ class JsonProtocolSuite extends FunSuite {
       |  },
       |  "Task Metrics": {
       |    "Host Name": "localhost",
-      |    "Executor Launch Time": 1100,
       |    "Executor Deserialize Time": 300,
       |    "Executor Run Time": 400,
       |    "Result Size": 500,