Show the Stage ID and Task ID that corresponds to the max metric
digraph G {
subgraph cluster0 {
isCluster="true";
label="WholeStageCodegen (1)\n \nduration: total (min, med, max (stageId: taskId))\n1.6 s (515 ms, 516 ms, 520 ms (stage 1661.0: task 5083))";
1 [labelType="html" label="<br><b>Project</b><br><br>"];
2 [labelType="html" label="<br><b>Project</b><br><br>"];
}
3 [labelType="html" label="<br><b>Project</b><br><br>"];
4 [labelType="html" label="<b>MicroBatchScan</b><br><br>number of output rows: 511<br>estimated number of fetched offsets out of range: 0<br>number of data loss error: 0"];
2->1;
3->2;
4->3;
}
5
Project [topic#9, partition#10, offset#11L, kafka_timestamp#21, kafka_timestamp_type#22, kafka_key#23, raw_value#24, event_type#34, ticker#45, synced_at#57, ingested_at#70, cast(kafka_timestamp#21 as date) AS event_date#83, date_format(kafka_timestamp#21, HH, Some(UTC)) AS event_hour#97]
Project [topic#9, partition#10, offset#11L, kafka_timestamp#21, kafka_timestamp_type#22, kafka_key#23, raw_value#24, event#25.type AS event_type#34, event#25.ticker AS ticker#45, event#25.syncedAt AS synced_at#57, 2026-06-19 02:38:30.007 AS ingested_at#70]
WholeStageCodegen (1)
Project [topic#9, partition#10, offset#11L, timestamp#12 AS kafka_timestamp#21, timestampType#13 AS kafka_timestamp_type#22, cast(key#7 as string) AS kafka_key#23, cast(value#8 as string) AS raw_value#24, from_json(StructField(type,StringType,true), StructField(ticker,StringType,true), StructField(interval,StringType,true), StructField(fromTime,LongType,true), StructField(toTime,LongType,true), StructField(nextCheckpoint,LongType,true), StructField(syncedAt,StringType,true), cast(value#8 as string), Some(UTC)) AS event#25]
MicroBatchScan[key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13] class org.apache.spark.sql.kafka010.KafkaSourceProvider$KafkaScan