[SPARK-34898] Send ExecutorMetricsUpdate EventLog appropriately

• Type: Sub-task
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 3.2.0
• Fix Version/s: 3.2.0
• Component/s: Spark Core
• Assignee: Apache Spark
• Reporter: angerszhu
• Created: 29/Mar/21 08:55
• Updated: 17/Jun/21 17:10
• Resolved: 17/Jun/21 17:10

Description

In current EventLoggingListener, we won't write SparkListenerExecutorMetricsUpdate message at all

override def onExecutorMetricsUpdate(event: SparkListenerExecutorMetricsUpdate): Unit = {
  if (shouldLogStageExecutorMetrics) {
    event.executorUpdates.foreach { case (stageKey1, newPeaks) =>
      liveStageExecutorMetrics.foreach { case (stageKey2, metricsPerExecutor) =>
        // If the update came from the driver, stageKey1 will be the dummy key (-1, -1),
        // so record those peaks for all active stages.
        // Otherwise, record the peaks for the matching stage.
        if (stageKey1 == DRIVER_STAGE_KEY || stageKey1 == stageKey2) {
          val metrics = metricsPerExecutor.getOrElseUpdate(
            event.execId, new ExecutorMetrics())
          metrics.compareAndUpdatePeakValues(newPeaks)
        }
      }
    }
  }
}

It causes this effect that we can't get driver peakMemoryMetrics in SHS.  We can get executor's since it will update with TaskEnd events.

Issue Links

• links to: [Github] Pull Request #31992 (AngersZhuuuu)