Improve implementation of finding median in StatisticsMetrics

[amahussein]

Fixes #1461

Adds an InPlace median finding to improve the performance of the metric aggregates.
We used to sort a sequence to create StatisticsMetrics which turned out to be very expensive in large eventlogs.

It was found that there is a bottleneck in generateStageLevelAccums and generateSQLAccums due to the cost of sorting and using sequences of Long on metrics aggregates.

Impact on performance:

The PR improves the Qualification runtime. methods like

• generateSQLAccums is down from 21,550 to 3,480 ms (~80% improvement)
• generateStageLevelAccums is down from 115580 to 61,611 ms (~45% improvement)

Main optimization

Median finding: we used to convert a map to sequence. Then get it sorted. Then we pick the median, max, and min, and then we call seq.sum

• This turned to be expensive for large eventlogs
• This PR implements adds finding-median in linear time. On average, this is a linear complexity compared to sorting
• This pull request includes several changes to the AppSQLPlanAnalyzer and AppSparkMetricsAnalyzer classes to improve performance and simplify the codebase by using the breakOut method for collection transformations and introducing new methods in StatisticsMetrics. Additionally, it removes an unused object and refactors the AccumInfo class.

Impact on output

By doing a diff on the output folder, it was found that the sql_plan_metrics_for_application.csv is differemt. The new output sounds more correct

In output generated by the dev branch:

appIndex,sqlID,nodeID,nodeName,accumulatorId,name,min,median,max,total,metricType,stageIds
1,236,0,"Execute InsertIntoHadoopFsRelationCommand",0,"number of written files",0,0,0,2,"sum","2"

The above seems incorrect because the total is non-zero while min, max and median are zeros.

Vs the new output

appIndex,sqlID,nodeID,nodeName,accumulatorId,name,min,median,max,total,metricType,stageIds
1,236,0,"Execute InsertIntoHadoopFsRelationCommand",0,"number of written files",2,2,2,2,"sum","2"

Improvements to Collection Transformations:

• core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSQLPlanAnalyzer.scala: Replaced intermediate collection variables with direct transformations using breakOut to improve performance. [1] [2] [3] [4] [5]
• core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSparkMetricsAnalyzer.scala: Applied breakOut to streamline collection processing and reduce memory overhead. [1] [2] [3] [4] [5] [6] [7] [8] [9] [10]

Introduction of New Methods in StatisticsMetrics:

• core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/StatisticsMetrics.scala: Added createFromArr and createOptionalFromArr methods to compute statistics directly from arrays, improving code reuse and readability.

Refactoring and Cleanup:

• core/src/main/scala/org/apache/spark/sql/rapids/tool/store/AccumInfo.scala: Refactored the calculateAccStats method to use the new createFromArr method from StatisticsMetrics, simplifying the code.
• core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSparkMetricsAnalyzer.scala: Removed the unused AppSparkMetricsAnalyzer object and its getStatistics method.

[parthosa]

Thanks @amahussein for these improvements. Minor question.

[parthosa]

Q: Should this be bounded by until - from?

[parthosa]

This is an interesting usage of implicit mechanism.

[cindyyuanjiang]

Thanks @amahussein for this refactor!