Adds filter for failed and non completed stages

[sayedbilalbari]

Fixes #1552

Currently we store the stageInfo using the stageModelManager class where we map incoming stage information during the following events -

1. doSparkListenerStageCompleted
   

   spark-rapids-tools/core/src/main/scala/org/apache/spark/sql/rapids/tool/EventProcessorBase.scala

   Line 475 in 1f037fa

   app.getOrCreateStage(event.stageInfo)

2. doSparkListenerStageSubmitted. -
   

   spark-rapids-tools/core/src/main/scala/org/apache/spark/sql/rapids/tool/EventProcessorBase.scala

   Line 464 in 1f037fa

   app.getOrCreateStage(event.stageInfo)

So a stage information is updated once when a stage is submitted and once during completion. A stageCompleted event comes for an attempt for a failed stage as well ( eg - there will be two stage Submitted and StageCompleted events for stage that fails on first attempt and succeeds on attempt 2)
Both those stageInfo objects are updated in the map
This PR adds extra filters to calculate aggregate metrics only for stages that have not failed and have completed successfully.

Changes -

1. Add a new method - getAllCompletedStages for filtering StageInfo objects on being successful and completed
2. Changes previous usages of getAllStages to getAllCompletedStages

Effects -

This change now ensures that all stage level aggregate metrics generated only take into account the successful ( not failed and completed ) ones

[amahussein]

Did you see any change in the output as an impact of that change? If you a diff between the outputs of before/after is there any change in the metrics?

[amahussein]

Do we still need the Todo? If not we can replace it with explanation about why we are using getAllCompleteStages

[sayedbilalbari]

Have removed the TODO

[amahussein]

TODO still there

[sayedbilalbari]

Updated

[eordentlich]

@leewyang can you take a look at this wrt qualx?

[amahussein]

let's make that a proper jdoc on top of the method with more details as much as we can.
I am not sure, the name of the method is easy to understand what this method does.

[sayedbilalbari]

A killed status for a stage is not there. It can only exist for a Task.
Updated the method with the correct jdoc and the naming is now better.
Thanks for pointing it out.

[amahussein]

TODO still there

[amahussein]

Thanks @sayedbilalbari

It is interesting that this change did not trigger any change in the expected results of the unit tests. If that's the case, it will be great if we can add a test that shows this version is actually generating something different compared to the previous version.
Otherwise, it might be possible that the code had no impact at all.

[leewyang]

For qualx purposes, would there be a way to get the metrics associated with the failed stages/tasks, or will these be dropped entirely? For customers who care about the total time/resources used, it would be useful to get the metrics associated with successful stages/tasks and teh failed stages/tasks, so we could tell that a job spent 90% of it's time/resources on failed stages/stasks. Otherwise, the job would appear to only take 10% of the actual time/resources used in reality.

[sayedbilalbari]

Thanks @amahussein for reviewing.
For the killed/failed tasks, as mentioned before, the way accumulable come in is different from how we are currently parsing it. The logic is as follows-

1. For killed tasks, the accumUpdates show up in the object event.reason.accumUpdates but not all of them are relevant. Ob further analysis, majority of them are zeros. They just denote the accumulables this task was supposed to deal with.
2. But the accumulables where it actually made updates show up in the taskInfo.accumulables ( as mentioned above like peakExecutoinMemory, jvmGCTime etc. )
3. So no extra logic is needed to parse the accumulableUpdates that come in with killedTasks, since we already parse the TaskInfo and use that to store relevant information.

As for no failure scenarios in the test, this change tackles two major scenarios -

1. Multiple successful attempts for a stage - will update the test case for this ( have already tested that this gives correct and different results for event logs )
2. Failed or non completed stages - test event logs did not have scenarios where a stage does not complete at all. There are scenarios where a primary attempt will fail but the secondary attempt will override the primary failure entry and hence no change in aggregated files is seen.

We can discuss this offline about what would be the best way to generate event logs like this for the test case.

[leewyang]

For qualx purposes, would there be a way to get the metrics associated with the failed stages/tasks, or will these be dropped entirely? For customers who care about the total time/resources used, it would be useful to get the metrics associated with successful stages/tasks and teh failed stages/tasks, so we could tell that a job spent 90% of it's time/resources on failed stages/stasks. Otherwise, the job would appear to only take 10% of the actual time/resources used in reality.

FYI, we have existing code that is trying to look at metrics for failed stages.

[winningsix]

For sm.duration, can we output both all stages and successFulStageAttempts? It's quite common to have failed stages for an in-product runs. For duration, we want to see the cost there.