NasaLogSparkStreaming - Details for Query 19

Details for Query 19

Submitted Time: 2025/03/20 14:04:54
Duration: 12 s
Succeeded Jobs: 6

Show the Stage ID and Task ID that corresponds to the max metric

digraph G { 0 [labelType="html" label=" WriteToDataSourceV2 "]; subgraph cluster1 { isCluster="true"; label="WholeStageCodegen (6)\n \nduration: total (min, med, max (stageId: taskId))\n251 ms (0 ms, 1 ms, 4 ms (stage 13.0: task 1253))"; 2 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 3 ms (0 ms, 0 ms, 1 ms (stage 13.0: task 1220)) peak memory total (min, med, max (stageId: taskId)) 434.0 MiB (256.0 KiB, 256.0 KiB, 64.3 MiB (stage 13.0: task 1220)) number of output rows: 6 avg hash probe bucket list iters (min, med, max (stageId: taskId)): (1, 1, 1 (stage 13.0: task 1220))"]; } 3 [labelType="html" label="StateStoreSave number of total state rows: 6 memory used by state total (min, med, max (stageId: taskId)) 79.6 KiB (400.0 B, 400.0 B, 712.0 B (stage 13.0: task 1238)) count of cache hit on states cache in provider: 2,400 number of output rows: 6 estimated size of state only on current version total (min, med, max (stageId: taskId)) 18.4 KiB (88.0 B, 88.0 B, 312.0 B (stage 13.0: task 1398)) time to commit changes total (min, med, max (stageId: taskId)) 15.4 s (46 ms, 77 ms, 108 ms (stage 13.0: task 1294)) time to remove total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 13.0: task 1214)) number of updated state rows: 2 time to update total (min, med, max (stageId: taskId)) 344 ms (0 ms, 1 ms, 16 ms (stage 13.0: task 1238))"]; subgraph cluster4 { isCluster="true"; label="WholeStageCodegen (5)\n \nduration: total (min, med, max (stageId: taskId))\n345 ms (0 ms, 1 ms, 16 ms (stage 13.0: task 1238))"; 5 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 13.0: task 1214)) peak memory total (min, med, max (stageId: taskId)) 50.0 MiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 13.0: task 1214)) number of output rows: 2"]; } 6 [labelType="html" label="StateStoreRestore number of output rows: 2"]; subgraph cluster7 { isCluster="true"; label="WholeStageCodegen (4)\n \nduration: total (min, med, max (stageId: taskId))\n1.3 s (3 ms, 6 ms, 12 ms (stage 13.0: task 1338))"; 8 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 7 ms (0 ms, 0 ms, 7 ms (stage 13.0: task 1238)) peak memory total (min, med, max (stageId: taskId)) 50.0 MiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 13.0: task 1214)) number of output rows: 2"]; } 9 [labelType="html" label="Exchange shuffle records written: 4 shuffle write time total (min, med, max (stageId: taskId)) 8 ms (3 ms, 5 ms, 5 ms (stage 12.0: task 1212)) records read: 4 local bytes read total (min, med, max (stageId: taskId)) 210.0 B (0.0 B, 0.0 B, 113.0 B (stage 13.0: task 1398)) fetch wait time total (min, med, max (stageId: taskId)) 6 ms (0 ms, 0 ms, 6 ms (stage 13.0: task 1238)) remote bytes read total (min, med, max (stageId: taskId)) 210.0 B (0.0 B, 0.0 B, 113.0 B (stage 13.0: task 1398)) local blocks read: 2 remote blocks read: 2 data size total (min, med, max (stageId: taskId)) 256.0 B (128.0 B, 128.0 B, 128.0 B (stage 12.0: task 1213)) shuffle bytes written total (min, med, max (stageId: taskId)) 420.0 B (210.0 B, 210.0 B, 210.0 B (stage 12.0: task 1213))"]; subgraph cluster10 { isCluster="true"; label="WholeStageCodegen (3)\n \nduration: total (min, med, max (stageId: taskId))\n17 ms (8 ms, 9 ms, 9 ms (stage 12.0: task 1213))"; 11 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 2 ms (1 ms, 1 ms, 1 ms (stage 12.0: task 1213)) peak memory total (min, med, max (stageId: taskId)) 512.0 KiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 12.0: task 1213)) number of output rows: 4"]; 12 [labelType="html" label=" Project "]; } 13 [labelType="html" label=" EventTimeWatermark "]; subgraph cluster14 { isCluster="true"; label="WholeStageCodegen (2)\n \nduration: total (min, med, max (stageId: taskId))\n12 ms (5 ms, 7 ms, 7 ms (stage 12.0: task 1213))"; 15 [labelType="html" label=" Project "]; } 16 [labelType="html" label="Generate number of output rows: 12"]; subgraph cluster17 { isCluster="true"; label="WholeStageCodegen (1)\n \nduration: total (min, med, max (stageId: taskId))\n16 ms (7 ms, 9 ms, 9 ms (stage 12.0: task 1213))"; 18 [labelType="html" label=" Project "]; } 19 [labelType="html" label="MicroBatchScan number of output rows: 12"]; 2->0; 3->2; 5->3; 6->5; 8->6; 9->8; 11->9; 12->11; 13->12; 15->13; 16->15; 18->16; 19->18; }

WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@5435a6e5

HashAggregate(keys=[hostname#12, responsecode#38], functions=[count(1)])

WholeStageCodegen (6)

StateStoreSave [hostname#12, responsecode#38], state info [ checkpoint = file:/tmp/temporary-1de1ee23-414b-4a88-8dc3-b55201323fb3/state, runId = 05d0ae24-6241-4634-a3bd-732191bb756e, opId = 0, ver = 6, numPartitions = 200], Complete, 1742479477566, 2

HashAggregate(keys=[hostname#12, responsecode#38], functions=[merge_count(1)])

WholeStageCodegen (5)

StateStoreRestore [hostname#12, responsecode#38], state info [ checkpoint = file:/tmp/temporary-1de1ee23-414b-4a88-8dc3-b55201323fb3/state, runId = 05d0ae24-6241-4634-a3bd-732191bb756e, opId = 0, ver = 6, numPartitions = 200], 2

HashAggregate(keys=[hostname#12, responsecode#38], functions=[merge_count(1)])

WholeStageCodegen (4)

Exchange hashpartitioning(hostname#12, responsecode#38, 200), true, [id=#572]

HashAggregate(keys=[hostname#12, responsecode#38], functions=[partial_count(1)])

Project [split(logs#5, ,, -1)[1] AS hostname#12, split(logs#5, ,, -1)[5] AS responsecode#38]

WholeStageCodegen (3)

EventTimeWatermark timestamp#1: timestamp, 5 seconds

Project [logs#5, timestamp#1]

WholeStageCodegen (2)

Generate explode(split(value#0, , -1)), [timestamp#1], false, [logs#5]

Project [value#0, timestamp#1]

WholeStageCodegen (1)

MicroBatchScan[value#0, timestamp#1] class org.apache.spark.sql.execution.streaming.sources.TextSocketTable$$anon$1

Details

== Parsed Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@5435a6e5
+- Aggregate [hostname#12, responsecode#38], [hostname#12, responsecode#38, count(1) AS count#67L]
   +- Project [logs#5, timestamp#1-T5000ms, idx#8, hostname#12, time#17, method#23, resource#30, responsecode#38, split(logs#5, ,, -1)[6] AS bytes#47]
      +- Project [logs#5, timestamp#1-T5000ms, idx#8, hostname#12, time#17, method#23, resource#30, split(logs#5, ,, -1)[5] AS responsecode#38]
         +- Project [logs#5, timestamp#1-T5000ms, idx#8, hostname#12, time#17, method#23, split(logs#5, ,, -1)[4] AS resource#30]
            +- Project [logs#5, timestamp#1-T5000ms, idx#8, hostname#12, time#17, split(logs#5, ,, -1)[3] AS method#23]
               +- Project [logs#5, timestamp#1-T5000ms, idx#8, hostname#12, split(logs#5, ,, -1)[2] AS time#17]
                  +- Project [logs#5, timestamp#1-T5000ms, idx#8, split(logs#5, ,, -1)[1] AS hostname#12]
                     +- Project [logs#5, timestamp#1-T5000ms, split(logs#5, ,, -1)[0] AS idx#8]
                        +- EventTimeWatermark timestamp#1: timestamp, 5 seconds
                           +- Project [logs#5, timestamp#1]
                              +- Generate explode(split(value#0,  , -1)), false, [logs#5]
                                 +- StreamingDataSourceV2Relation [value#0, timestamp#1], org.apache.spark.sql.execution.streaming.sources.TextSocketTable$$anon$1@3bb35f79, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], 68, 80

== Analyzed Logical Plan ==

WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@5435a6e5
+- Aggregate [hostname#12, responsecode#38], [hostname#12, responsecode#38, count(1) AS count#67L]
   +- Project [logs#5, timestamp#1-T5000ms, idx#8, hostname#12, time#17, method#23, resource#30, responsecode#38, split(logs#5, ,, -1)[6] AS bytes#47]
      +- Project [logs#5, timestamp#1-T5000ms, idx#8, hostname#12, time#17, method#23, resource#30, split(logs#5, ,, -1)[5] AS responsecode#38]
         +- Project [logs#5, timestamp#1-T5000ms, idx#8, hostname#12, time#17, method#23, split(logs#5, ,, -1)[4] AS resource#30]
            +- Project [logs#5, timestamp#1-T5000ms, idx#8, hostname#12, time#17, split(logs#5, ,, -1)[3] AS method#23]
               +- Project [logs#5, timestamp#1-T5000ms, idx#8, hostname#12, split(logs#5, ,, -1)[2] AS time#17]
                  +- Project [logs#5, timestamp#1-T5000ms, idx#8, split(logs#5, ,, -1)[1] AS hostname#12]
                     +- Project [logs#5, timestamp#1-T5000ms, split(logs#5, ,, -1)[0] AS idx#8]
                        +- EventTimeWatermark timestamp#1: timestamp, 5 seconds
                           +- Project [logs#5, timestamp#1]
                              +- Generate explode(split(value#0,  , -1)), false, [logs#5]
                                 +- StreamingDataSourceV2Relation [value#0, timestamp#1], org.apache.spark.sql.execution.streaming.sources.TextSocketTable$$anon$1@3bb35f79, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], 68, 80

== Optimized Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@5435a6e5
+- Aggregate [hostname#12, responsecode#38], [hostname#12, responsecode#38, count(1) AS count#67L]
   +- Project [split(logs#5, ,, -1)[1] AS hostname#12, split(logs#5, ,, -1)[5] AS responsecode#38]
      +- EventTimeWatermark timestamp#1: timestamp, 5 seconds
         +- Project [logs#5, timestamp#1]
            +- Generate explode(split(value#0,  , -1)), [0], false, [logs#5]
               +- StreamingDataSourceV2Relation [value#0, timestamp#1], org.apache.spark.sql.execution.streaming.sources.TextSocketTable$$anon$1@3bb35f79, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], 68, 80

== Physical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@5435a6e5
+- *(6) HashAggregate(keys=[hostname#12, responsecode#38], functions=[count(1)], output=[hostname#12, responsecode#38, count#67L])
   +- StateStoreSave [hostname#12, responsecode#38], state info [ checkpoint = file:/tmp/temporary-1de1ee23-414b-4a88-8dc3-b55201323fb3/state, runId = 05d0ae24-6241-4634-a3bd-732191bb756e, opId = 0, ver = 6, numPartitions = 200], Complete, 1742479477566, 2
      +- *(5) HashAggregate(keys=[hostname#12, responsecode#38], functions=[merge_count(1)], output=[hostname#12, responsecode#38, count#75L])
         +- StateStoreRestore [hostname#12, responsecode#38], state info [ checkpoint = file:/tmp/temporary-1de1ee23-414b-4a88-8dc3-b55201323fb3/state, runId = 05d0ae24-6241-4634-a3bd-732191bb756e, opId = 0, ver = 6, numPartitions = 200], 2
            +- *(4) HashAggregate(keys=[hostname#12, responsecode#38], functions=[merge_count(1)], output=[hostname#12, responsecode#38, count#75L])
               +- Exchange hashpartitioning(hostname#12, responsecode#38, 200), true, [id=#572]
                  +- *(3) HashAggregate(keys=[hostname#12, responsecode#38], functions=[partial_count(1)], output=[hostname#12, responsecode#38, count#75L])
                     +- *(3) Project [split(logs#5, ,, -1)[1] AS hostname#12, split(logs#5, ,, -1)[5] AS responsecode#38]
                        +- EventTimeWatermark timestamp#1: timestamp, 5 seconds
                           +- *(2) Project [logs#5, timestamp#1]
                              +- Generate explode(split(value#0,  , -1)), [timestamp#1], false, [logs#5]
                                 +- *(1) Project [value#0, timestamp#1]
                                    +- MicroBatchScan[value#0, timestamp#1] class org.apache.spark.sql.execution.streaming.sources.TextSocketTable$$anon$1