• Submitted Time: 2024/12/27 14:32:43
• Duration: 47 s

== Parsed Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@3d025bbd
+- Sort [total_bytes#99L DESC NULLS LAST], true
   +- Aggregate [window#100-T3000ms, hostname#14], [window#100-T3000ms AS window#88-T3000ms, hostname#14, sum(cast(bytes#49 as bigint)) AS total_bytes#99L]
      +- Filter ((timestamp#1-T3000ms >= window#100-T3000ms.start) AND (timestamp#1-T3000ms < window#100-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#10, hostname#14, time#19, method#25, url#32, responsecode#40, bytes#49), 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#10, hostname#14, time#19, method#25, url#32, responsecode#40, bytes#49)], [window#100-T3000ms, logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, responsecode#40, bytes#49]
            +- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, responsecode#40, split(logs#5, ,, -1)[6] AS bytes#49]
               +- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, split(logs#5, ,, -1)[5] AS responsecode#40]
                  +- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, split(logs#5, ,, -1)[4] AS url#32]
                     +- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, split(logs#5, ,, -1)[3] AS method#25]
                        +- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, split(logs#5, ,, -1)[2] AS time#19]
                           +- Project [logs#5, timestamp#1-T3000ms, idx#10, split(logs#5, ,, -1)[1] AS hostname#14]
                              +- Project [logs#5, timestamp#1-T3000ms, split(logs#5, ,, -1)[0] AS idx#10]
                                 +- 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@60063c8e, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], -1, -1

== Analyzed Logical Plan ==

WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@3d025bbd
+- Sort [total_bytes#99L DESC NULLS LAST], true
   +- Aggregate [window#100-T3000ms, hostname#14], [window#100-T3000ms AS window#88-T3000ms, hostname#14, sum(cast(bytes#49 as bigint)) AS total_bytes#99L]
      +- Filter ((timestamp#1-T3000ms >= window#100-T3000ms.start) AND (timestamp#1-T3000ms < window#100-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#10, hostname#14, time#19, method#25, url#32, responsecode#40, bytes#49), 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#10, hostname#14, time#19, method#25, url#32, responsecode#40, bytes#49)], [window#100-T3000ms, logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, responsecode#40, bytes#49]
            +- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, responsecode#40, split(logs#5, ,, -1)[6] AS bytes#49]
               +- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, split(logs#5, ,, -1)[5] AS responsecode#40]
                  +- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, split(logs#5, ,, -1)[4] AS url#32]
                     +- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, split(logs#5, ,, -1)[3] AS method#25]
                        +- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, split(logs#5, ,, -1)[2] AS time#19]
                           +- Project [logs#5, timestamp#1-T3000ms, idx#10, split(logs#5, ,, -1)[1] AS hostname#14]
                              +- Project [logs#5, timestamp#1-T3000ms, split(logs#5, ,, -1)[0] AS idx#10]
                                 +- 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@60063c8e, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], -1, -1

== Optimized Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@3d025bbd
+- Sort [total_bytes#99L DESC NULLS LAST], true
   +- Aggregate [window#100-T3000ms, hostname#14], [window#100-T3000ms AS window#88-T3000ms, hostname#14, sum(cast(bytes#49 as bigint)) AS total_bytes#99L]
      +- Project [window#100-T3000ms, hostname#14, bytes#49]
         +- Filter (((isnotnull(timestamp#1-T3000ms) AND isnotnull(window#100-T3000ms)) AND (timestamp#1-T3000ms >= window#100-T3000ms.start)) AND (timestamp#1-T3000ms < window#100-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#14, bytes#49), 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#14, bytes#49)], [window#100-T3000ms, timestamp#1-T3000ms, hostname#14, bytes#49]
               +- Project [timestamp#1-T3000ms, split(logs#5, ,, -1)[1] AS hostname#14, split(logs#5, ,, -1)[6] AS bytes#49]
                  +- 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@60063c8e, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], -1, -1

== Physical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@3d025bbd
+- *(7) Sort [total_bytes#99L DESC NULLS LAST], true, 0
   +- Exchange rangepartitioning(total_bytes#99L DESC NULLS LAST, 200), true, [id=#261]
      +- *(6) HashAggregate(keys=[window#100-T3000ms, hostname#14], functions=[sum(cast(bytes#49 as bigint))], output=[window#88-T3000ms, hostname#14, total_bytes#99L])
         +- StateStoreSave [window#100-T3000ms, hostname#14], state info [ checkpoint = file:/tmp/temporary-aedcc499-293d-46eb-a3bd-fc1a6948006f/state, runId = 7e18f44d-be9b-4945-b3a4-eccf8dfc2985, opId = 0, ver = 0, numPartitions = 200], Complete, 0, 2
            +- *(5) HashAggregate(keys=[window#100-T3000ms, hostname#14], functions=[merge_sum(cast(bytes#49 as bigint))], output=[window#100-T3000ms, hostname#14, sum#130L])
               +- StateStoreRestore [window#100-T3000ms, hostname#14], state info [ checkpoint = file:/tmp/temporary-aedcc499-293d-46eb-a3bd-fc1a6948006f/state, runId = 7e18f44d-be9b-4945-b3a4-eccf8dfc2985, opId = 0, ver = 0, numPartitions = 200], 2
                  +- *(4) HashAggregate(keys=[window#100-T3000ms, hostname#14], functions=[merge_sum(cast(bytes#49 as bigint))], output=[window#100-T3000ms, hostname#14, sum#130L])
                     +- Exchange hashpartitioning(window#100-T3000ms, hostname#14, 200), true, [id=#249]
                        +- *(3) HashAggregate(keys=[window#100-T3000ms, hostname#14], functions=[partial_sum(cast(bytes#49 as bigint))], output=[window#100-T3000ms, hostname#14, sum#130L])
                           +- *(3) Project [window#100-T3000ms, hostname#14, bytes#49]
                              +- *(3) Filter (((isnotnull(timestamp#1-T3000ms) AND isnotnull(window#100-T3000ms)) AND (timestamp#1-T3000ms >= window#100-T3000ms.start)) AND (timestamp#1-T3000ms < window#100-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#14, bytes#49), 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#14, bytes#49)], [window#100-T3000ms, timestamp#1-T3000ms, hostname#14, bytes#49]
                                    +- *(3) Project [timestamp#1-T3000ms, split(logs#5, ,, -1)[1] AS hostname#14, split(logs#5, ,, -1)[6] AS bytes#49]
                                       +- 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