edwinalu commented on a change in pull request #23340: [SPARK-23431][CORE]
Expose the new executor memory metrics at the stage level
URL: https://github.com/apache/spark/pull/23340#discussion_r243088549
##########
File path: core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
##########
@@ -802,15 +817,24 @@ private[spark] class AppStatusListener(
}
}
- override def onStageExecutorMetrics(executorMetrics:
SparkListenerStageExecutorMetrics): Unit = {
+ override def onStageExecutorMetrics(event:
SparkListenerStageExecutorMetrics): Unit = {
val now = System.nanoTime()
-
+ Option(liveStages.get((event.stageId, event.stageAttemptId)))
+ .foreach { stage =>
+ if
(stage.peakExecutorMetrics.compareAndUpdatePeakValues(event.executorMetrics)) {
+ update(stage, now)
+ }
+ val esummary = stage.executorSummary(event.execId)
+ if
(esummary.peakExecutorMetrics.compareAndUpdatePeakValues(event.executorMetrics))
{
+ update(esummary, now)
+ }
+ }
// check if there is a new peak value for any of the executor level memory
metrics,
// while reading from the log. SparkListenerStageExecutorMetrics are only
processed
// when reading logs.
Review comment:
Yes, comments are still applicable, although for stage level. I'll move the
comments around, and and add a pointer to onExecutorMetricsUpdate for the live
app case.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
