[BUG] Aggregate metric per stage is missing filter for stage attempts

[sayedbilalbari]

Currently aggregateDiagnosticMetricByStage and aggregateSparkMetricsByStageInternal use the getAllStages method of the stageModelManager which returns all the stages ( failed, successful, incomplete etc.).

spark-rapids-tools/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSparkMetricsAnalyzer.scala

Line 395 in 78cab00

app.stageManager.getAllStages.foreach { sm =>

and

spark-rapids-tools/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSparkMetricsAnalyzer.scala

Line 324 in 78cab00

app.stageManager.getAllStages.map { sm =>

This could lead to incorrect aggregation since different stageAttempts can override each other. The behavior here is non-deterministic as we are not sure which attempt will end up being last.

Solution

We want to enforce behaviour that just counts for the successful attempts of a stage as we are already dumping the failed stages in a separate report

• When aggregating metrics, we should make sure that we do not mix-and-match-between different attempts.
• This can be done by only picking the attempts that have not failed. Same applies for incomplete attempts since those ones can override each other as well.
• Another alternative is to aggregate per stage attempt but this might not be ideal because failed stages do not have associated accumulables in the eventlog.