digraph G {
0 [labelType="html" label="<br><b>WriteToDataSourceV2</b><br><br>"];
subgraph cluster1 {
isCluster="true";
label="WholeStageCodegen (7)";
2 [labelType="html" label="<br><b>Sort</b><br><br>"];
}
3 [labelType="html" label="<br><b>Exchange</b><br><br>"];
subgraph cluster4 {
isCluster="true";
label="WholeStageCodegen (6)\n \nduration: total (min, med, max (stageId: taskId))\n6 ms (0 ms, 0 ms, 3 ms (stage 28.0: task 1610))";
5 [labelType="html" label="<b>HashAggregate</b><br><br>time in aggregation build total (min, med, max (stageId: taskId))<br>0 ms (0 ms, 0 ms, 0 ms (stage 28.0: task 1610))<br>peak memory total (min, med, max (stageId: taskId))<br>512.0 KiB (0.0 B, 0.0 B, 256.0 KiB (stage 28.0: task 1610))"];
}
6 [labelType="html" label="<b>StateStoreSave</b><br><br>memory used by state total (min, med, max (stageId: taskId))<br>752.0 B (0.0 B, 0.0 B, 376.0 B (stage 28.0: task 1610))<br>count of cache hit on states cache in provider: 4<br>estimated size of state only on current version total (min, med, max (stageId: taskId))<br>176.0 B (0.0 B, 0.0 B, 88.0 B (stage 28.0: task 1610))<br>time to commit changes total (min, med, max (stageId: taskId))<br>155 ms (0 ms, 0 ms, 84 ms (stage 28.0: task 1614))<br>time to remove total (min, med, max (stageId: taskId))<br>0 ms (0 ms, 0 ms, 0 ms (stage 28.0: task 1610))<br>time to update total (min, med, max (stageId: taskId))<br>1 ms (0 ms, 0 ms, 1 ms (stage 28.0: task 1614))"];
subgraph cluster7 {
isCluster="true";
label="WholeStageCodegen (5)\n \nduration: total (min, med, max (stageId: taskId))\n1 ms (0 ms, 0 ms, 1 ms (stage 28.0: task 1614))";
8 [labelType="html" label="<b>HashAggregate</b><br><br>time in aggregation build total (min, med, max (stageId: taskId))<br>0 ms (0 ms, 0 ms, 0 ms (stage 28.0: task 1610))<br>peak memory total (min, med, max (stageId: taskId))<br>512.0 KiB (0.0 B, 0.0 B, 256.0 KiB (stage 28.0: task 1610))"];
}
9 [labelType="html" label="<br><b>StateStoreRestore</b><br><br>"];
subgraph cluster10 {
isCluster="true";
label="WholeStageCodegen (4)\n \nduration: total (min, med, max (stageId: taskId))\n9 ms (0 ms, 0 ms, 6 ms (stage 28.0: task 1614))";
11 [labelType="html" label="<b>HashAggregate</b><br><br>time in aggregation build total (min, med, max (stageId: taskId))<br>0 ms (0 ms, 0 ms, 0 ms (stage 28.0: task 1610))<br>peak memory total (min, med, max (stageId: taskId))<br>512.0 KiB (0.0 B, 0.0 B, 256.0 KiB (stage 28.0: task 1610))"];
}
12 [labelType="html" label="<b>Exchange</b><br><br>shuffle records written: 19<br>shuffle write time total (min, med, max (stageId: taskId))<br>11 ms (5 ms, 5 ms, 5 ms (stage 27.0: task 1405))<br>data size total (min, med, max (stageId: taskId))<br>1352.0 B (592.0 B, 760.0 B, 760.0 B (stage 27.0: task 1405))<br>shuffle bytes written total (min, med, max (stageId: taskId))<br>1996.0 B (878.0 B, 1118.0 B, 1118.0 B (stage 27.0: task 1405))"];
subgraph cluster13 {
isCluster="true";
label="WholeStageCodegen (3)\n \nduration: total (min, med, max (stageId: taskId))\n24 ms (11 ms, 13 ms, 13 ms (stage 27.0: task 1605))";
14 [labelType="html" label="<b>HashAggregate</b><br><br>time in aggregation build total (min, med, max (stageId: taskId))<br>6 ms (3 ms, 3 ms, 3 ms (stage 27.0: task 1405))<br>peak memory total (min, med, max (stageId: taskId))<br>128.5 MiB (64.3 MiB, 64.3 MiB, 64.3 MiB (stage 27.0: task 1405))<br>number of output rows: 19<br>avg hash probe bucket list iters (min, med, max (stageId: taskId)):<br>(1, 1, 1 (stage 27.0: task 1405))"];
15 [labelType="html" label="<br><b>Project</b><br><br>"];
16 [labelType="html" label="<b>Filter</b><br><br>number of output rows: 100"];
17 [labelType="html" label="<b>Expand</b><br><br>number of output rows: 100"];
18 [labelType="html" label="<br><b>Project</b><br><br>"];
}
19 [labelType="html" label="<br><b>EventTimeWatermark</b><br><br>"];
subgraph cluster20 {
isCluster="true";
label="WholeStageCodegen (2)\n \nduration: total (min, med, max (stageId: taskId))\n15 ms (7 ms, 8 ms, 8 ms (stage 27.0: task 1605))";
21 [labelType="html" label="<br><b>Project</b><br><br>"];
}
22 [labelType="html" label="<b>Generate</b><br><br>number of output rows: 50"];
subgraph cluster23 {
isCluster="true";
label="WholeStageCodegen (1)\n \nduration: total (min, med, max (stageId: taskId))\n18 ms (8 ms, 10 ms, 10 ms (stage 27.0: task 1605))";
24 [labelType="html" label="<br><b>Project</b><br><br>"];
}
25 [labelType="html" label="<b>MicroBatchScan</b><br><br>number of output rows: 48"];
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;
}
26
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@39c45ae2
Sort [total_bytes#99L DESC NULLS LAST], true, 0
WholeStageCodegen (7)
Exchange rangepartitioning(total_bytes#99L DESC NULLS LAST, 200), true, [id=#1001]
HashAggregate(keys=[window#100-T3000ms, hostname#14], functions=[sum(cast(bytes#49 as bigint))])
WholeStageCodegen (6)
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 = 1, numPartitions = 200], Complete, 0, 2
HashAggregate(keys=[window#100-T3000ms, hostname#14], functions=[merge_sum(cast(bytes#49 as bigint))])
WholeStageCodegen (5)
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 = 1, numPartitions = 200], 2
HashAggregate(keys=[window#100-T3000ms, hostname#14], functions=[merge_sum(cast(bytes#49 as bigint))])
WholeStageCodegen (4)
Exchange hashpartitioning(window#100-T3000ms, hostname#14, 200), true, [id=#989]
HashAggregate(keys=[window#100-T3000ms, hostname#14], functions=[partial_sum(cast(bytes#49 as bigint))])
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]
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
== Parsed Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@39c45ae2
+- 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, 47
== Analyzed Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@39c45ae2
+- 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, 47
== Optimized Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@39c45ae2
+- 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, 47
== Physical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@39c45ae2
+- *(7) Sort [total_bytes#99L DESC NULLS LAST], true, 0
+- Exchange rangepartitioning(total_bytes#99L DESC NULLS LAST, 200), true, [id=#1001]
+- *(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 = 1, 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 = 1, 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=#989]
+- *(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
