NasaLogSparkStreaming - Details for Query 61

Details for Query 61

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

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))\n11.5 s (37 ms, 45 ms, 170 ms (stage 21.0: task 42))"; 2 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 11.1 s (36 ms, 43 ms, 162 ms (stage 21.0: task 42)) peak memory total (min, med, max (stageId: taskId)) 50.0 MiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 21.0: task 43))"]; } 3 [labelType="html" label="StateStoreSave memory used by state total (min, med, max (stageId: taskId)) 43.8 KiB (224.0 B, 224.0 B, 224.0 B (stage 21.0: task 43)) estimated size of state only on current version total (min, med, max (stageId: taskId)) 15.6 KiB (80.0 B, 80.0 B, 80.0 B (stage 21.0: task 43)) time to commit changes total (min, med, max (stageId: taskId)) 10.4 s (34 ms, 40 ms, 141 ms (stage 21.0: task 42)) time to remove total (min, med, max (stageId: taskId)) 63 ms (0 ms, 0 ms, 6 ms (stage 21.0: task 42)) time to update total (min, med, max (stageId: taskId)) 710 ms (1 ms, 3 ms, 42 ms (stage 21.0: task 42))"]; subgraph cluster4 { isCluster="true"; label="WholeStageCodegen (5)\n \nduration: total (min, med, max (stageId: taskId))\n938 ms (2 ms, 4 ms, 73 ms (stage 21.0: task 42))"; 5 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 5 ms (0 ms, 0 ms, 4 ms (stage 21.0: task 106)) peak memory total (min, med, max (stageId: taskId)) 50.0 MiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 21.0: task 43))"]; } 6 [labelType="html" label=" StateStoreRestore "]; subgraph cluster7 { isCluster="true"; label="WholeStageCodegen (4)\n \nduration: total (min, med, max (stageId: taskId))\n1.9 s (4 ms, 8 ms, 132 ms (stage 21.0: task 42))"; 8 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 4 ms (0 ms, 0 ms, 3 ms (stage 21.0: task 42)) peak memory total (min, med, max (stageId: taskId)) 50.0 MiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 21.0: task 43))"]; } 9 [labelType="html" label=" Exchange "]; subgraph cluster10 { isCluster="true"; label="WholeStageCodegen (3)\n \nduration: total (min, med, max (stageId: taskId))\n223 ms (79 ms, 144 ms, 144 ms (stage 20.0: task 40))"; 11 [labelType="html" label="HashAggregate time in aggregation build total (min, med, max (stageId: taskId)) 10 ms (4 ms, 6 ms, 6 ms (stage 20.0: task 40)) peak memory total (min, med, max (stageId: taskId)) 512.0 KiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 20.0: task 41))"]; 12 [labelType="html" label=" Project "]; 13 [labelType="html" label=" Filter "]; 14 [labelType="html" label=" Expand "]; } 15 [labelType="html" label=" EventTimeWatermark "]; subgraph cluster16 { isCluster="true"; label="WholeStageCodegen (2)\n \nduration: total (min, med, max (stageId: taskId))\n424 ms (163 ms, 261 ms, 261 ms (stage 20.0: task 40))"; 17 [labelType="html" label=" Project "]; 18 [labelType="html" label=" Filter "]; } 19 [labelType="html" label=" Generate "]; subgraph cluster20 { isCluster="true"; label="WholeStageCodegen (1)\n \nduration: total (min, med, max (stageId: taskId))\n469 ms (180 ms, 289 ms, 289 ms (stage 20.0: task 40))"; 21 [labelType="html" label=" Project "]; } 22 [labelType="html" label=" MicroBatchScan "]; 2->0; 3->2; 5->3; 6->5; 8->6; 9->8; 11->9; 12->11; 13->12; 14->13; 15->14; 17->15; 18->17; 19->18; 21->19; 22->21; }

WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@3e8c9c4c

HashAggregate(keys=[window#822-T3000ms], functions=[count(1)])

WholeStageCodegen (6)

StateStoreSave [window#822-T3000ms], state info [ checkpoint = file:/tmp/temporary-a4a518b2-a3ff-4163-8a29-c5a18d39c201/state, runId = 22115180-3cbd-43f4-9c89-7ad00fbe0238, opId = 0, ver = 0, numPartitions = 200], Update, 0, 2

HashAggregate(keys=[window#822-T3000ms], functions=[merge_count(1)])

WholeStageCodegen (5)

StateStoreRestore [window#822-T3000ms], state info [ checkpoint = file:/tmp/temporary-a4a518b2-a3ff-4163-8a29-c5a18d39c201/state, runId = 22115180-3cbd-43f4-9c89-7ad00fbe0238, opId = 0, ver = 0, numPartitions = 200], 2

HashAggregate(keys=[window#822-T3000ms], functions=[merge_count(1)])

WholeStageCodegen (4)

Exchange hashpartitioning(window#822-T3000ms, 200), true, [id=#775]

HashAggregate(keys=[window#822-T3000ms], functions=[partial_count(1)])

Project [window#822-T3000ms]

Filter (((isnotnull(timestamp#1-T3000ms) AND isnotnull(window#822-T3000ms)) AND (timestamp#1-T3000ms >= window#822-T3000ms.start)) AND (timestamp#1-T3000ms < window#822-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), 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)], [window#822-T3000ms, timestamp#1-T3000ms]

WholeStageCodegen (3)

EventTimeWatermark timestamp#1: timestamp, 3 seconds

Project [timestamp#1]

Filter Contains(split(logs#5, ,, -1)[4], gif)

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@3e8c9c4c
+- Project [window#810-T3000ms, count#821L AS gif_count#825L]
   +- Aggregate [window#822-T3000ms], [window#822-T3000ms AS window#810-T3000ms, count(1) AS count#821L]
      +- Filter ((timestamp#1-T3000ms >= window#822-T3000ms.start) AND (timestamp#1-T3000ms < window#822-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#136, hostname#140, time#145, method#151, url#158, responsecode#166, bytes#175), 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#136, hostname#140, time#145, method#151, url#158, responsecode#166, bytes#175)], [window#822-T3000ms, logs#5, timestamp#1-T3000ms, idx#136, hostname#140, time#145, method#151, url#158, responsecode#166, bytes#175]
            +- Filter url#158 LIKE %gif%
               +- Project [logs#5, timestamp#1-T3000ms, idx#136, hostname#140, time#145, method#151, url#158, responsecode#166, split(logs#5, ,, -1)[6] AS bytes#175]
                  +- Project [logs#5, timestamp#1-T3000ms, idx#136, hostname#140, time#145, method#151, url#158, split(logs#5, ,, -1)[5] AS responsecode#166]
                     +- Project [logs#5, timestamp#1-T3000ms, idx#136, hostname#140, time#145, method#151, split(logs#5, ,, -1)[4] AS url#158]
                        +- Project [logs#5, timestamp#1-T3000ms, idx#136, hostname#140, time#145, split(logs#5, ,, -1)[3] AS method#151]
                           +- Project [logs#5, timestamp#1-T3000ms, idx#136, hostname#140, split(logs#5, ,, -1)[2] AS time#145]
                              +- Project [logs#5, timestamp#1-T3000ms, idx#136, split(logs#5, ,, -1)[1] AS hostname#140]
                                 +- Project [logs#5, timestamp#1-T3000ms, split(logs#5, ,, -1)[0] AS idx#136]
                                    +- 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@7eafcdef, 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@3e8c9c4c
+- Project [window#810-T3000ms, count#821L AS gif_count#825L]
   +- Aggregate [window#822-T3000ms], [window#822-T3000ms AS window#810-T3000ms, count(1) AS count#821L]
      +- Filter ((timestamp#1-T3000ms >= window#822-T3000ms.start) AND (timestamp#1-T3000ms < window#822-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#136, hostname#140, time#145, method#151, url#158, responsecode#166, bytes#175), 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#136, hostname#140, time#145, method#151, url#158, responsecode#166, bytes#175)], [window#822-T3000ms, logs#5, timestamp#1-T3000ms, idx#136, hostname#140, time#145, method#151, url#158, responsecode#166, bytes#175]
            +- Filter url#158 LIKE %gif%
               +- Project [logs#5, timestamp#1-T3000ms, idx#136, hostname#140, time#145, method#151, url#158, responsecode#166, split(logs#5, ,, -1)[6] AS bytes#175]
                  +- Project [logs#5, timestamp#1-T3000ms, idx#136, hostname#140, time#145, method#151, url#158, split(logs#5, ,, -1)[5] AS responsecode#166]
                     +- Project [logs#5, timestamp#1-T3000ms, idx#136, hostname#140, time#145, method#151, split(logs#5, ,, -1)[4] AS url#158]
                        +- Project [logs#5, timestamp#1-T3000ms, idx#136, hostname#140, time#145, split(logs#5, ,, -1)[3] AS method#151]
                           +- Project [logs#5, timestamp#1-T3000ms, idx#136, hostname#140, split(logs#5, ,, -1)[2] AS time#145]
                              +- Project [logs#5, timestamp#1-T3000ms, idx#136, split(logs#5, ,, -1)[1] AS hostname#140]
                                 +- Project [logs#5, timestamp#1-T3000ms, split(logs#5, ,, -1)[0] AS idx#136]
                                    +- 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@7eafcdef, 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@3e8c9c4c
+- Aggregate [window#822-T3000ms], [window#822-T3000ms AS window#810-T3000ms, count(1) AS gif_count#825L]
   +- Project [window#822-T3000ms]
      +- Filter (((isnotnull(timestamp#1-T3000ms) AND isnotnull(window#822-T3000ms)) AND (timestamp#1-T3000ms >= window#822-T3000ms.start)) AND (timestamp#1-T3000ms < window#822-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), 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)], [window#822-T3000ms, timestamp#1-T3000ms]
            +- EventTimeWatermark timestamp#1: timestamp, 3 seconds
               +- Project [timestamp#1]
                  +- Filter Contains(split(logs#5, ,, -1)[4], gif)
                     +- 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@7eafcdef, 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@3e8c9c4c
+- *(6) HashAggregate(keys=[window#822-T3000ms], functions=[count(1)], output=[window#810-T3000ms, gif_count#825L])
   +- StateStoreSave [window#822-T3000ms], state info [ checkpoint = file:/tmp/temporary-a4a518b2-a3ff-4163-8a29-c5a18d39c201/state, runId = 22115180-3cbd-43f4-9c89-7ad00fbe0238, opId = 0, ver = 0, numPartitions = 200], Update, 0, 2
      +- *(5) HashAggregate(keys=[window#822-T3000ms], functions=[merge_count(1)], output=[window#822-T3000ms, count#831L])
         +- StateStoreRestore [window#822-T3000ms], state info [ checkpoint = file:/tmp/temporary-a4a518b2-a3ff-4163-8a29-c5a18d39c201/state, runId = 22115180-3cbd-43f4-9c89-7ad00fbe0238, opId = 0, ver = 0, numPartitions = 200], 2
            +- *(4) HashAggregate(keys=[window#822-T3000ms], functions=[merge_count(1)], output=[window#822-T3000ms, count#831L])
               +- Exchange hashpartitioning(window#822-T3000ms, 200), true, [id=#775]
                  +- *(3) HashAggregate(keys=[window#822-T3000ms], functions=[partial_count(1)], output=[window#822-T3000ms, count#831L])
                     +- *(3) Project [window#822-T3000ms]
                        +- *(3) Filter (((isnotnull(timestamp#1-T3000ms) AND isnotnull(window#822-T3000ms)) AND (timestamp#1-T3000ms >= window#822-T3000ms.start)) AND (timestamp#1-T3000ms < window#822-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), 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)], [window#822-T3000ms, timestamp#1-T3000ms]
                              +- EventTimeWatermark timestamp#1: timestamp, 3 seconds
                                 +- *(2) Project [timestamp#1]
                                    +- *(2) Filter Contains(split(logs#5, ,, -1)[4], gif)
                                       +- 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