NasaLogSparkStreaming - Details for Query 4

Details for Query 4

Submitted Time: 2024/12/04 20:23:18
Duration: 11 s
Succeeded Jobs: 1

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))\n421 ms (1 ms, 2 ms, 24 ms (stage 3.0: task 286))"; 2 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 22 ms (0 ms, 0 ms, 22 ms (stage 3.0: task 286)) peak memory total (min, med, max (stageId: taskId)) 114.0 MiB (256.0 KiB, 256.0 KiB, 64.3 MiB (stage 3.0: task 286)) number of output rows: 1 avg hash probe bucket list iters: 1"]; } 3 [labelType="html" label="StateStoreSave number of total state rows: 1 memory used by state total (min, med, max (stageId: taskId)) 73.6 KiB (376.0 B, 376.0 B, 568.0 B (stage 3.0: task 286)) count of cache hit on states cache in provider: 400 number of output rows: 1 estimated size of state only on current version total (min, med, max (stageId: taskId)) 17.4 KiB (88.0 B, 88.0 B, 280.0 B (stage 3.0: task 286)) time to commit changes total (min, med, max (stageId: taskId)) 12.3 s (38 ms, 59 ms, 191 ms (stage 3.0: task 319)) time to remove total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 3.0: task 204)) number of updated state rows: 1 time to update total (min, med, max (stageId: taskId)) 507 ms (1 ms, 2 ms, 73 ms (stage 3.0: task 286))"]; subgraph cluster4 { isCluster="true"; label="WholeStageCodegen (5)\n \nduration: total (min, med, max (stageId: taskId))\n513 ms (1 ms, 2 ms, 73 ms (stage 3.0: task 286))"; 5 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 1 ms (0 ms, 0 ms, 1 ms (stage 3.0: task 286)) peak memory total (min, med, max (stageId: taskId)) 50.0 MiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 3.0: task 204)) number of output rows: 1"]; } 6 [labelType="html" label="StateStoreRestore number of output rows: 1"]; subgraph cluster7 { isCluster="true"; label="WholeStageCodegen (4)\n \nduration: total (min, med, max (stageId: taskId))\n2.1 s (4 ms, 8 ms, 373 ms (stage 3.0: task 382))"; 8 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 31 ms (0 ms, 0 ms, 31 ms (stage 3.0: task 286)) peak memory total (min, med, max (stageId: taskId)) 50.0 MiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 3.0: task 204)) number of output rows: 1"]; } 9 [labelType="html" label="Exchange shuffle records written: 2 shuffle write time total (min, med, max (stageId: taskId)) 40 ms (19 ms, 21 ms, 21 ms (stage 2.0: task 203)) records read: 2 local bytes read total (min, med, max (stageId: taskId)) 80.0 B (0.0 B, 0.0 B, 80.0 B (stage 3.0: task 286)) fetch wait time total (min, med, max (stageId: taskId)) 20 ms (0 ms, 0 ms, 20 ms (stage 3.0: task 286)) remote bytes read total (min, med, max (stageId: taskId)) 80.0 B (0.0 B, 0.0 B, 80.0 B (stage 3.0: task 286)) local blocks read: 1 remote blocks read: 1 data size total (min, med, max (stageId: taskId)) 80.0 B (40.0 B, 40.0 B, 40.0 B (stage 2.0: task 202)) shuffle bytes written total (min, med, max (stageId: taskId)) 160.0 B (80.0 B, 80.0 B, 80.0 B (stage 2.0: task 202))"]; subgraph cluster10 { isCluster="true"; label="WholeStageCodegen (3)\n \nduration: total (min, med, max (stageId: taskId))\n67 ms (30 ms, 37 ms, 37 ms (stage 2.0: task 203))"; 11 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 25 ms (10 ms, 15 ms, 15 ms (stage 2.0: task 203)) peak memory total (min, med, max (stageId: taskId)) 512.0 KiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 2.0: task 202)) number of output rows: 2"]; 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))\n37 ms (16 ms, 21 ms, 21 ms (stage 2.0: task 203))"; 15 [labelType="html" label=" Project "]; 16 [labelType="html" label="Filter number of output rows: 15"]; } 17 [labelType="html" label="Generate number of output rows: 18"]; subgraph cluster18 { isCluster="true"; label="WholeStageCodegen (1)\n \nduration: total (min, med, max (stageId: taskId))\n42 ms (18 ms, 24 ms, 24 ms (stage 2.0: task 203))"; 19 [labelType="html" label=" Project "]; } 20 [labelType="html" label="MicroBatchScan number of output rows: 16"]; 2->0; 3->2; 5->3; 6->5; 8->6; 9->8; 11->9; 12->11; 13->12; 15->13; 16->15; 17->16; 19->17; 20->19; }

WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@74576b15

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

WholeStageCodegen (6)

StateStoreSave [hostname#12], state info [ checkpoint = file:/tmp/temporary-7dde7be9-c7f2-4a1a-96c6-fb19f120126d/state, runId = 10bd72d4-04b3-48e3-97a6-62b1f13a8094, opId = 0, ver = 1, numPartitions = 200], Complete, 0, 2

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

WholeStageCodegen (5)

StateStoreRestore [hostname#12], state info [ checkpoint = file:/tmp/temporary-7dde7be9-c7f2-4a1a-96c6-fb19f120126d/state, runId = 10bd72d4-04b3-48e3-97a6-62b1f13a8094, opId = 0, ver = 1, numPartitions = 200], 2

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

WholeStageCodegen (4)

Exchange hashpartitioning(hostname#12, 200), true, [id=#162]

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

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

WholeStageCodegen (3)

EventTimeWatermark timestamp#1: timestamp, 3 seconds

Project [logs#5, timestamp#1]

Filter ((split(logs#5, ,, -1)[3] = GET) AND (split(logs#5, ,, -1)[5] = 200))

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@74576b15
+- Project [hostname#12, count#67L AS correct_count#70L]
   +- Aggregate [hostname#12], [hostname#12, count(1) AS count#67L]
      +- Filter ((method#23 = GET) AND (responsecode#38 = 200))
         +- Project [logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, method#23, url#30, responsecode#38, split(logs#5, ,, -1)[6] AS bytes#47]
            +- Project [logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, method#23, url#30, split(logs#5, ,, -1)[5] AS responsecode#38]
               +- Project [logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, method#23, split(logs#5, ,, -1)[4] AS url#30]
                  +- Project [logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, split(logs#5, ,, -1)[3] AS method#23]
                     +- Project [logs#5, timestamp#1-T3000ms, idx#8, hostname#12, split(logs#5, ,, -1)[2] AS time#17]
                        +- Project [logs#5, timestamp#1-T3000ms, idx#8, split(logs#5, ,, -1)[1] AS hostname#12]
                           +- Project [logs#5, timestamp#1-T3000ms, split(logs#5, ,, -1)[0] AS idx#8]
                              +- EventTimeWatermark timestamp#1: timestamp, 3 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@484fdb6, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], -1, 15

== Analyzed Logical Plan ==

WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@74576b15
+- Project [hostname#12, count#67L AS correct_count#70L]
   +- Aggregate [hostname#12], [hostname#12, count(1) AS count#67L]
      +- Filter ((method#23 = GET) AND (responsecode#38 = 200))
         +- Project [logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, method#23, url#30, responsecode#38, split(logs#5, ,, -1)[6] AS bytes#47]
            +- Project [logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, method#23, url#30, split(logs#5, ,, -1)[5] AS responsecode#38]
               +- Project [logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, method#23, split(logs#5, ,, -1)[4] AS url#30]
                  +- Project [logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, split(logs#5, ,, -1)[3] AS method#23]
                     +- Project [logs#5, timestamp#1-T3000ms, idx#8, hostname#12, split(logs#5, ,, -1)[2] AS time#17]
                        +- Project [logs#5, timestamp#1-T3000ms, idx#8, split(logs#5, ,, -1)[1] AS hostname#12]
                           +- Project [logs#5, timestamp#1-T3000ms, split(logs#5, ,, -1)[0] AS idx#8]
                              +- EventTimeWatermark timestamp#1: timestamp, 3 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@484fdb6, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], -1, 15

== Optimized Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@74576b15
+- Aggregate [hostname#12], [hostname#12, count(1) AS correct_count#70L]
   +- Project [split(logs#5, ,, -1)[1] AS hostname#12]
      +- EventTimeWatermark timestamp#1: timestamp, 3 seconds
         +- Project [logs#5, timestamp#1]
            +- Filter ((split(logs#5, ,, -1)[3] = GET) AND (split(logs#5, ,, -1)[5] = 200))
               +- 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@484fdb6, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], -1, 15

== Physical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@74576b15
+- *(6) HashAggregate(keys=[hostname#12], functions=[count(1)], output=[hostname#12, correct_count#70L])
   +- StateStoreSave [hostname#12], state info [ checkpoint = file:/tmp/temporary-7dde7be9-c7f2-4a1a-96c6-fb19f120126d/state, runId = 10bd72d4-04b3-48e3-97a6-62b1f13a8094, opId = 0, ver = 1, numPartitions = 200], Complete, 0, 2
      +- *(5) HashAggregate(keys=[hostname#12], functions=[merge_count(1)], output=[hostname#12, count#76L])
         +- StateStoreRestore [hostname#12], state info [ checkpoint = file:/tmp/temporary-7dde7be9-c7f2-4a1a-96c6-fb19f120126d/state, runId = 10bd72d4-04b3-48e3-97a6-62b1f13a8094, opId = 0, ver = 1, numPartitions = 200], 2
            +- *(4) HashAggregate(keys=[hostname#12], functions=[merge_count(1)], output=[hostname#12, count#76L])
               +- Exchange hashpartitioning(hostname#12, 200), true, [id=#162]
                  +- *(3) HashAggregate(keys=[hostname#12], functions=[partial_count(1)], output=[hostname#12, count#76L])
                     +- *(3) Project [split(logs#5, ,, -1)[1] AS hostname#12]
                        +- EventTimeWatermark timestamp#1: timestamp, 3 seconds
                           +- *(2) Project [logs#5, timestamp#1]
                              +- *(2) Filter ((split(logs#5, ,, -1)[3] = GET) AND (split(logs#5, ,, -1)[5] = 200))
                                 +- 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