squito 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_r243024717
##########
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:
I need a little refresher here -- does this comment also apply to the new
code you're adding above?
and why again does this only apply for when we're reading from the event
logs (maybe this comment should be updated to point to whatever is happening
for a live app)
----------------------------------------------------------------
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]
