[FEA] Add filtered diagnostic output for GPU slowness in Profiler tool

[cindyyuanjiang]

Contributes to #1374

Changes

• Add a filter diagnostic (operator and SQL metrics) view in Profiler output: filtered_diagnostic_metrics.csv
  • Introduced function generateFilteredDiagnosticAccums to compute filtered diagnostic metrics in core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSQLPlanAnalyzer.scala
  • Introduced FilteredAccumDiagnosticMetrics to store selected filtered related metric names and methods
  • Added FilteredDiagnosticResult to represent filtered diagnostic metrics result

Testing

• Add unit test test filtered diagnostic metrics in core/src/test/scala/com/nvidia/spark/rapids/tool/profiling/AnalysisSuite.scala

Example Output

appIndex,appName,appId,sqlId,stageId,stageDurationMs,nodeId,nodeName,numFilesReadMin,numFilesReadMedian,numFilesReadMax,numFilesReadTotal,numPartitionsMin,numPartitionsMedian,numPartitionsMax,numPartitionsTotal,metadataTimeMin,metadataTimeMedian,metadataTimeMax,metadataTimeTotal,outputBatchesMin,outputBatchesMedian,outputBatchesMax,outputBatchesTotal,inputBatchesMin,inputBatchesMedian,inputBatchesMax,inputBatchesTotal,outputRowsMin,outputRowsMedian,outputRowsMax,outputRowsTotal,sortTimeMin,sortTimeMedian,sortTimeMax,sortTimeTotal,peakMemoryMin,peakMemoryMedian,peakMemoryMax,peakMemoryTotal,shuffleBytesWrittenMin,shuffleBytesWrittenMedian,shuffleBytesWrittenMax,shuffleBytesWrittenTotal,shuffleWriteTimeMin,shuffleWriteTimeMedian,shuffleWriteTimeMax,shuffleWriteTimeTotal
1,Spark shell,local-1622814619968,0,0,1743,16,"GpuColumnarExchange",0,0,0,0,0,0,0,0,0,0,0,0,200,200,200,1200,0,0,0,0,1666666,1666666,1666667,10000000,0,0,0,0,0,0,0,0,6688608,6688707,6688825,40132250,41434653,66714083,100858775,400284505
1,Spark shell,local-1622814619968,0,0,1743,21,"Scan",0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1666666,1666666,1666667,10000000,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0
1,Spark shell,local-1622814619968,0,1,1631,8,"GpuColumnarExchange",0,0,0,0,0,0,0,0,0,0,0,0,200,200,200,1200,0,0,0,0,1666666,1666666,1666667,10000000,0,0,0,0,0,0,0,0,6688602,6688708,6688833,40132258,37444140,84791745,108992798,508750471
1,Spark shell,local-1622814619968,0,1,1631,13,"Scan",0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1666666,1666666,1666667,10000000,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0
1,Spark shell,local-1622814619968,0,2,688,3,"GpuColumnarExchange",0,0,0,0,0,0,0,0,0,0,0,0,1,1,1,200,0,0,0,0,1,1,1,200,0,0,0,0,0,0,0,0,77,77,77,15400,139875,465911,9747416,93193331
1,Spark shell,local-1622814619968,0,2,688,4,"GpuHashAggregate",0,0,0,0,0,0,0,0,0,0,0,0,1,1,1,200,0,0,0,0,1,1,1,200,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0

[sayedbilalbari]

Just a few doubts

[sayedbilalbari]

Why has this been removed ?

[sayedbilalbari]

I am assuming this is named as FilteredAccumDiagnostic metrics because it represents a subset of metrics that are more relevant to profiler output. The candidates for this being the names mentioned. Is there a doc somewhere highlighting why these operators ?

[sayedbilalbari]

Just a thought, so we iterate over all the SQLMetricInfoCase objects, and using that try to maintain a mapping of accumId -> (sqlID, nodeID, nodeName)
Can there be a case where an entry belonging to the same accumID is overridden in the next iteration. So basically can one accumID belong to multiple combinations of sqlID, nodeID and nodeName combos ?
In my minimal understanding, accumID is agnostic of sql and multiple sqls and nodes should be able to update the same accumulable

[sayedbilalbari]

So the logic here to be clear ->

1. We first get the filteredStageAccumList from the accumStage level aggregated metrics by doing a name match
2. Then iterate over that list ( granularity is accumMeta +StageID)
3. For the accumID, we find the sqlID,nodeID, nodeName
4. Maintain a cache of sqlID, nodeID, nodeName, stageID -> accumName, statistic
5. In this, can there be multiple accumulable updates coming for the same key as above ? Here the value distinction is correct ( name + statistic) but on the key level there can be multiple accum updates
6. Also, contextual doubt - all the metrics being filtered, is generating these from a SQL Metric View relevant ? Because as part of some previous discussion, the SQL Metrics sometimes may not be that accurate
7. So a better way could be getting them from the internal.metrics.* accums. But that is only relevant in case SQL View is not needed by downstream ( otherwise can ignore)

[sayedbilalbari]

Is there a way to eliminate this new map creation ? So we have accumID, we have the stageID, we can get accum + stage stats from accumManager. and I see a map sqlPlanNodeIdToStageIds above. This implementation is cleaner, but just verifying is this can be avoided and kept clean at the same time would be good