NasaLogSparkStreaming - Details for Query 28

Details for Query 28

Submitted Time: 2024/12/12 00:04:34
Duration: 5 s
Succeeded Jobs: 18 19

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 (7)\n \nduration: total (min, med, max (stageId: taskId))\n4 ms (0 ms, 0 ms, 2 ms (stage 49.0: task 4090))"; 2 [labelType="html" label="Sort sort time total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 49.0: task 4083)) peak memory total (min, med, max (stageId: taskId)) 512.6 MiB (64.0 KiB, 64.1 MiB, 64.1 MiB (stage 49.0: task 4083)) spill size total (min, med, max (stageId: taskId)) 0.0 B (0.0 B, 0.0 B, 0.0 B (stage 49.0: task 4083))"]; } 3 [labelType="html" label="Exchange shuffle records written: 12 shuffle write time total (min, med, max (stageId: taskId)) 4 ms (0 ms, 0 ms, 0 ms (stage 48.0: task 3906)) records read: 12 local bytes read total (min, med, max (stageId: taskId)) 837.0 B (0.0 B, 107.0 B, 115.0 B (stage 49.0: task 4086)) fetch wait time total (min, med, max (stageId: taskId)) 1 ms (0 ms, 0 ms, 1 ms (stage 49.0: task 4090)) remote bytes read total (min, med, max (stageId: taskId)) 380.0 B (0.0 B, 0.0 B, 265.0 B (stage 49.0: task 4090)) local blocks read: 8 remote blocks read: 4 data size total (min, med, max (stageId: taskId)) 832.0 B (0.0 B, 0.0 B, 80.0 B (stage 48.0: task 3922)) shuffle bytes written total (min, med, max (stageId: taskId)) 1217.0 B (0.0 B, 0.0 B, 115.0 B (stage 48.0: task 3922))"]; subgraph cluster4 { isCluster="true"; label="WholeStageCodegen (6)\n \nduration: total (min, med, max (stageId: taskId))\n54 ms (0 ms, 0 ms, 2 ms (stage 46.0: task 3773))"; 5 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 46.0: task 3685)) peak memory total (min, med, max (stageId: taskId)) 1636.0 MiB (256.0 KiB, 256.0 KiB, 64.3 MiB (stage 46.0: task 3706)) number of output rows: 24 avg hash probe bucket list iters (min, med, max (stageId: taskId)): (1, 1, 1 (stage 46.0: task 3706))"]; } 6 [labelType="html" label="StateStoreSave number of total state rows: 24 memory used by state total (min, med, max (stageId: taskId)) 129.1 KiB (232.0 B, 400.0 B, 728.0 B (stage 46.0: task 3727)) count of cache hit on states cache in provider: 14,000 number of output rows: 24 estimated size of state only on current version total (min, med, max (stageId: taskId)) 39.6 KiB (88.0 B, 88.0 B, 320.0 B (stage 46.0: task 3722)) time to commit changes total (min, med, max (stageId: taskId)) 12.3 s (22 ms, 31 ms, 42 ms (stage 46.0: task 3828)) time to remove total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 46.0: task 3685)) number of updated state rows: 6 time to update total (min, med, max (stageId: taskId)) 73 ms (0 ms, 0 ms, 8 ms (stage 48.0: task 4012))"]; subgraph cluster7 { isCluster="true"; label="WholeStageCodegen (5)\n \nduration: total (min, med, max (stageId: taskId))\n73 ms (0 ms, 0 ms, 8 ms (stage 48.0: task 4012))"; 8 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 46.0: task 3685)) peak memory total (min, med, max (stageId: taskId)) 484.0 MiB (256.0 KiB, 256.0 KiB, 64.3 MiB (stage 46.0: task 3727)) number of output rows: 6 avg hash probe bucket list iters (min, med, max (stageId: taskId)): (1, 1, 1 (stage 46.0: task 3727))"]; } 9 [labelType="html" label="StateStoreRestore number of output rows: 6"]; subgraph cluster10 { isCluster="true"; label="WholeStageCodegen (4)\n \nduration: total (min, med, max (stageId: taskId))\n437 ms (0 ms, 1 ms, 4 ms (stage 46.0: task 3837))"; 11 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 1 ms (0 ms, 0 ms, 1 ms (stage 46.0: task 3812)) peak memory total (min, med, max (stageId: taskId)) 484.0 MiB (256.0 KiB, 256.0 KiB, 64.3 MiB (stage 46.0: task 3727)) number of output rows: 6 avg hash probe bucket list iters (min, med, max (stageId: taskId)): (1, 1, 1 (stage 46.0: task 3727))"]; } 12 [labelType="html" label="Exchange shuffle records written: 9 shuffle write time total (min, med, max (stageId: taskId)) 11 ms (2 ms, 3 ms, 3 ms (stage 45.0: task 3682)) records read: 18 local bytes read total (min, med, max (stageId: taskId)) 1064.0 B (0.0 B, 0.0 B, 213.0 B (stage 46.0: task 3812)) fetch wait time total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 46.0: task 3727)) remote bytes read total (min, med, max (stageId: taskId)) 850.0 B (0.0 B, 0.0 B, 213.0 B (stage 46.0: task 3867)) local blocks read: 10 remote blocks read: 8 data size total (min, med, max (stageId: taskId)) 648.0 B (144.0 B, 144.0 B, 216.0 B (stage 45.0: task 3679)) shuffle bytes written total (min, med, max (stageId: taskId)) 957.0 B (212.0 B, 214.0 B, 319.0 B (stage 45.0: task 3679))"]; subgraph cluster13 { isCluster="true"; label="WholeStageCodegen (3)\n \nduration: total (min, med, max (stageId: taskId))\n17 ms (4 ms, 4 ms, 5 ms (stage 45.0: task 3682))"; 14 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 0 ms (0 ms, 0 ms, 0 ms (stage 45.0: task 3681)) peak memory total (min, med, max (stageId: taskId)) 257.0 MiB (64.3 MiB, 64.3 MiB, 64.3 MiB (stage 45.0: task 3681)) number of output rows: 9 avg hash probe bucket list iters (min, med, max (stageId: taskId)): (1, 1, 1 (stage 45.0: task 3681))"]; 15 [labelType="html" label=" Project "]; 16 [labelType="html" label="Filter number of output rows: 10"]; 17 [labelType="html" label="Expand number of output rows: 10"]; 18 [labelType="html" label=" Project "]; } 19 [labelType="html" label=" EventTimeWatermark "]; subgraph cluster20 { isCluster="true"; label="WholeStageCodegen (2)\n \nduration: total (min, med, max (stageId: taskId))\n8 ms (2 ms, 2 ms, 2 ms (stage 45.0: task 3681))"; 21 [labelType="html" label=" Project "]; } 22 [labelType="html" label="Generate number of output rows: 5"]; subgraph cluster23 { isCluster="true"; label="WholeStageCodegen (1)\n \nduration: total (min, med, max (stageId: taskId))\n9 ms (2 ms, 2 ms, 3 ms (stage 45.0: task 3679))"; 24 [labelType="html" label=" Project "]; } 25 [labelType="html" label="MicroBatchScan number of output rows: 5"]; 2->0; 3->2; 5->3; 6->5; 8->6; 9->8; 11->9; 12->11; 14->12; 15->14; 16->15; 17->16; 18->17; 19->18; 21->19; 22->21; 24->22; 25->24; }

WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@6d4b1803

Sort [total_bytes#68 DESC NULLS LAST], true, 0

WholeStageCodegen (7)

Exchange rangepartitioning(total_bytes#68 DESC NULLS LAST, 200), true, [id=#1102]

HashAggregate(keys=[window#69-T3000ms, hostname#12], functions=[sum(cast(bytes#47 as double))])

WholeStageCodegen (6)

StateStoreSave [window#69-T3000ms, hostname#12], state info [ checkpoint = file:/tmp/spark-checkpoints/state, runId = 24a21f8f-7b8d-48ec-9359-b31ba78c32c6, opId = 0, ver = 9, numPartitions = 200], Complete, 1733961865129, 2

HashAggregate(keys=[window#69-T3000ms, hostname#12], functions=[merge_sum(cast(bytes#47 as double))])

WholeStageCodegen (5)

StateStoreRestore [window#69-T3000ms, hostname#12], state info [ checkpoint = file:/tmp/spark-checkpoints/state, runId = 24a21f8f-7b8d-48ec-9359-b31ba78c32c6, opId = 0, ver = 9, numPartitions = 200], 2

HashAggregate(keys=[window#69-T3000ms, hostname#12], functions=[merge_sum(cast(bytes#47 as double))])

WholeStageCodegen (4)

Exchange hashpartitioning(window#69-T3000ms, hostname#12, 200), true, [id=#1090]

HashAggregate(keys=[window#69-T3000ms, hostname#12], functions=[partial_sum(cast(bytes#47 as double))])

Project [window#69-T3000ms, hostname#12, bytes#47]

Filter (((isnotnull(timestamp#1-T3000ms) AND isnotnull(window#69-T3000ms)) AND (timestamp#1-T3000ms >= window#69-T3000ms.start)) AND (timestamp#1-T3000ms < window#69-T3000ms.end))

Expand [ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 0) - 2) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 0) - 2) * 30000000) + 60000000), LongType, TimestampType)), timestamp#1-T3000ms, hostname#12, bytes#47), ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 1) - 2) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 1) - 2) * 30000000) + 60000000), LongType, TimestampType)), timestamp#1-T3000ms, hostname#12, bytes#47)], [window#69-T3000ms, timestamp#1-T3000ms, hostname#12, bytes#47]

Project [timestamp#1-T3000ms, split(logs#5, ,, -1)[1] AS hostname#12, split(logs#5, ,, -1)[6] AS bytes#47]

WholeStageCodegen (3)

EventTimeWatermark timestamp#1: timestamp, 3 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@6d4b1803
+- Sort [total_bytes#68 DESC NULLS LAST], true
   +- Aggregate [window#69-T3000ms, hostname#12], [window#69-T3000ms AS window#57-T3000ms, hostname#12, sum(cast(bytes#47 as double)) AS total_bytes#68]
      +- Filter ((timestamp#1-T3000ms >= window#69-T3000ms.start) AND (timestamp#1-T3000ms < window#69-T3000ms.end))
         +- Expand [ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(0 as bigint)) - cast(2 as bigint)) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion((((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(0 as bigint)) - cast(2 as bigint)) * 30000000) + 0) + 60000000), LongType, TimestampType)), logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, method#23, url#30, responsecode#38, bytes#47), ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(1 as bigint)) - cast(2 as bigint)) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion((((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(1 as bigint)) - cast(2 as bigint)) * 30000000) + 0) + 60000000), LongType, TimestampType)), logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, method#23, url#30, responsecode#38, bytes#47)], [window#69-T3000ms, logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, method#23, url#30, responsecode#38, bytes#47]
            +- 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@1416817a, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], 48, 53

== Analyzed Logical Plan ==

WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@6d4b1803
+- Sort [total_bytes#68 DESC NULLS LAST], true
   +- Aggregate [window#69-T3000ms, hostname#12], [window#69-T3000ms AS window#57-T3000ms, hostname#12, sum(cast(bytes#47 as double)) AS total_bytes#68]
      +- Filter ((timestamp#1-T3000ms >= window#69-T3000ms.start) AND (timestamp#1-T3000ms < window#69-T3000ms.end))
         +- Expand [ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(0 as bigint)) - cast(2 as bigint)) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion((((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(0 as bigint)) - cast(2 as bigint)) * 30000000) + 0) + 60000000), LongType, TimestampType)), logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, method#23, url#30, responsecode#38, bytes#47), ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(1 as bigint)) - cast(2 as bigint)) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion((((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(1 as bigint)) - cast(2 as bigint)) * 30000000) + 0) + 60000000), LongType, TimestampType)), logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, method#23, url#30, responsecode#38, bytes#47)], [window#69-T3000ms, logs#5, timestamp#1-T3000ms, idx#8, hostname#12, time#17, method#23, url#30, responsecode#38, bytes#47]
            +- 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@1416817a, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], 48, 53

== Optimized Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@6d4b1803
+- Sort [total_bytes#68 DESC NULLS LAST], true
   +- Aggregate [window#69-T3000ms, hostname#12], [window#69-T3000ms AS window#57-T3000ms, hostname#12, sum(cast(bytes#47 as double)) AS total_bytes#68]
      +- Project [window#69-T3000ms, hostname#12, bytes#47]
         +- Filter (((isnotnull(timestamp#1-T3000ms) AND isnotnull(window#69-T3000ms)) AND (timestamp#1-T3000ms >= window#69-T3000ms.start)) AND (timestamp#1-T3000ms < window#69-T3000ms.end))
            +- Expand [ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 0) - 2) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 0) - 2) * 30000000) + 60000000), LongType, TimestampType)), timestamp#1-T3000ms, hostname#12, bytes#47), ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 1) - 2) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 1) - 2) * 30000000) + 60000000), LongType, TimestampType)), timestamp#1-T3000ms, hostname#12, bytes#47)], [window#69-T3000ms, timestamp#1-T3000ms, hostname#12, bytes#47]
               +- Project [timestamp#1-T3000ms, split(logs#5, ,, -1)[1] AS hostname#12, split(logs#5, ,, -1)[6] AS bytes#47]
                  +- EventTimeWatermark timestamp#1: timestamp, 3 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@1416817a, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], 48, 53

== Physical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@6d4b1803
+- *(7) Sort [total_bytes#68 DESC NULLS LAST], true, 0
   +- Exchange rangepartitioning(total_bytes#68 DESC NULLS LAST, 200), true, [id=#1102]
      +- *(6) HashAggregate(keys=[window#69-T3000ms, hostname#12], functions=[sum(cast(bytes#47 as double))], output=[window#57-T3000ms, hostname#12, total_bytes#68])
         +- StateStoreSave [window#69-T3000ms, hostname#12], state info [ checkpoint = file:/tmp/spark-checkpoints/state, runId = 24a21f8f-7b8d-48ec-9359-b31ba78c32c6, opId = 0, ver = 9, numPartitions = 200], Complete, 1733961865129, 2
            +- *(5) HashAggregate(keys=[window#69-T3000ms, hostname#12], functions=[merge_sum(cast(bytes#47 as double))], output=[window#69-T3000ms, hostname#12, sum#77])
               +- StateStoreRestore [window#69-T3000ms, hostname#12], state info [ checkpoint = file:/tmp/spark-checkpoints/state, runId = 24a21f8f-7b8d-48ec-9359-b31ba78c32c6, opId = 0, ver = 9, numPartitions = 200], 2
                  +- *(4) HashAggregate(keys=[window#69-T3000ms, hostname#12], functions=[merge_sum(cast(bytes#47 as double))], output=[window#69-T3000ms, hostname#12, sum#77])
                     +- Exchange hashpartitioning(window#69-T3000ms, hostname#12, 200), true, [id=#1090]
                        +- *(3) HashAggregate(keys=[window#69-T3000ms, hostname#12], functions=[partial_sum(cast(bytes#47 as double))], output=[window#69-T3000ms, hostname#12, sum#77])
                           +- *(3) Project [window#69-T3000ms, hostname#12, bytes#47]
                              +- *(3) Filter (((isnotnull(timestamp#1-T3000ms) AND isnotnull(window#69-T3000ms)) AND (timestamp#1-T3000ms >= window#69-T3000ms.start)) AND (timestamp#1-T3000ms < window#69-T3000ms.end))
                                 +- *(3) Expand [ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 0) - 2) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 0) - 2) * 30000000) + 60000000), LongType, TimestampType)), timestamp#1-T3000ms, hostname#12, bytes#47), ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 1) - 2) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 1) - 2) * 30000000) + 60000000), LongType, TimestampType)), timestamp#1-T3000ms, hostname#12, bytes#47)], [window#69-T3000ms, timestamp#1-T3000ms, hostname#12, bytes#47]
                                    +- *(3) Project [timestamp#1-T3000ms, split(logs#5, ,, -1)[1] AS hostname#12, split(logs#5, ,, -1)[6] AS bytes#47]
                                       +- EventTimeWatermark timestamp#1: timestamp, 3 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