digraph G {
0 [labelType="html" label="<br><b>WriteToDataSourceV2</b><br><br>"];
subgraph cluster1 {
isCluster="true";
label="WholeStageCodegen (7)\n \nduration: 35 ms";
2 [labelType="html" label="<b>Sort</b><br><br>sort time: 0 ms<br>peak memory: 64.0 KiB<br>spill size: 0.0 B"];
}
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))\n554 ms (0 ms, 1 ms, 13 ms (stage 5.0: task 209))";
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 5.0: task 208))<br>peak memory total (min, med, max (stageId: taskId))<br>100.0 MiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 5.0: task 208))"];
}
6 [labelType="html" label="<b>StateStoreSave</b><br><br>memory used by state total (min, med, max (stageId: taskId))<br>81.3 KiB (208.0 B, 208.0 B, 208.0 B (stage 5.0: task 208))<br>estimated size of state only on current version total (min, med, max (stageId: taskId))<br>25.0 KiB (64.0 B, 64.0 B, 64.0 B (stage 5.0: task 208))<br>time to commit changes total (min, med, max (stageId: taskId))<br>18.3 s (29 ms, 35 ms, 188 ms (stage 5.0: task 301))<br>time to remove total (min, med, max (stageId: taskId))<br>0 ms (0 ms, 0 ms, 0 ms (stage 5.0: task 208))<br>time to update total (min, med, max (stageId: taskId))<br>610 ms (0 ms, 1 ms, 7 ms (stage 5.0: task 284))"];
subgraph cluster7 {
isCluster="true";
label="WholeStageCodegen (5)\n \nduration: total (min, med, max (stageId: taskId))\n618 ms (0 ms, 1 ms, 8 ms (stage 5.0: task 284))";
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 5.0: task 208))<br>peak memory total (min, med, max (stageId: taskId))<br>100.0 MiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 5.0: task 208))"];
}
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))\n2.4 s (1 ms, 4 ms, 116 ms (stage 5.0: task 209))";
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 5.0: task 208))<br>peak memory total (min, med, max (stageId: taskId))<br>100.0 MiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 5.0: task 208))"];
}
12 [labelType="html" label="<br><b>Exchange</b><br><br>"];
subgraph cluster13 {
isCluster="true";
label="WholeStageCodegen (3)\n \nduration: total (min, med, max (stageId: taskId))\n228 ms (61 ms, 167 ms, 167 ms (stage 4.0: task 6))";
14 [labelType="html" label="<b>HashAggregate</b><br><br>time in aggregation build total (min, med, max (stageId: taskId))<br>4 ms (0 ms, 4 ms, 4 ms (stage 4.0: task 6))<br>peak memory total (min, med, max (stageId: taskId))<br>512.0 KiB (256.0 KiB, 256.0 KiB, 256.0 KiB (stage 4.0: task 7))"];
15 [labelType="html" label="<br><b>Project</b><br><br>"];
}
16 [labelType="html" label="<br><b>EventTimeWatermark</b><br><br>"];
subgraph cluster17 {
isCluster="true";
label="WholeStageCodegen (2)\n \nduration: total (min, med, max (stageId: taskId))\n329 ms (89 ms, 240 ms, 240 ms (stage 4.0: task 6))";
18 [labelType="html" label="<br><b>Project</b><br><br>"];
19 [labelType="html" label="<br><b>Filter</b><br><br>"];
}
20 [labelType="html" label="<br><b>Generate</b><br><br>"];
subgraph cluster21 {
isCluster="true";
label="WholeStageCodegen (1)\n \nduration: total (min, med, max (stageId: taskId))\n382 ms (100 ms, 282 ms, 282 ms (stage 4.0: task 6))";
22 [labelType="html" label="<br><b>Project</b><br><br>"];
}
23 [labelType="html" label="<br><b>MicroBatchScan</b><br><br>"];
2->0;
3->2;
5->3;
6->5;
8->6;
9->8;
11->9;
12->11;
14->12;
15->14;
16->15;
18->16;
19->18;
20->19;
22->20;
23->22;
}
24
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@5d2c170a
Sort [Correct_count#124L DESC NULLS LAST], true, 0
WholeStageCodegen (7)
Exchange rangepartitioning(Correct_count#124L DESC NULLS LAST, 200), true, [id=#359]
HashAggregate(keys=[hostname#14], functions=[count(1)])
WholeStageCodegen (6)
StateStoreSave [hostname#14], state info [ checkpoint = file:/tmp/temporary-f9fb39ce-1caa-4bde-a068-b38acb808fb3/state, runId = 61e1d4d7-e682-42ec-879a-83a32ac478c6, opId = 0, ver = 0, numPartitions = 200], Complete, 0, 2
HashAggregate(keys=[hostname#14], functions=[merge_count(1)])
WholeStageCodegen (5)
StateStoreRestore [hostname#14], state info [ checkpoint = file:/tmp/temporary-f9fb39ce-1caa-4bde-a068-b38acb808fb3/state, runId = 61e1d4d7-e682-42ec-879a-83a32ac478c6, opId = 0, ver = 0, numPartitions = 200], 2
HashAggregate(keys=[hostname#14], functions=[merge_count(1)])
WholeStageCodegen (4)
Exchange hashpartitioning(hostname#14, 200), true, [id=#347]
HashAggregate(keys=[hostname#14], functions=[partial_count(1)])
Project [split(logs#5, ,, -1)[1] AS hostname#14]
WholeStageCodegen (3)
EventTimeWatermark timestamp#1: timestamp, 3 seconds
Project [logs#5, timestamp#1]
Filter ((split(logs#5, ,, -1)[3] = GET) AND (split(logs#5, ,, -1)[5] = 200))
WholeStageCodegen (2)
Generate explode(split(value#0, , -1)), [timestamp#1], false, [logs#5]
Project [value#0, timestamp#1]
WholeStageCodegen (1)
MicroBatchScan[value#0, timestamp#1] class org.apache.spark.sql.execution.streaming.sources.TextSocketTable$$anon$1
== Parsed Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@5d2c170a
+- Sort [Correct_count#124L DESC NULLS LAST], true
+- Project [window#110, count#121L AS Correct_count#124L]
+- Aggregate [hostname#14], [hostname#14 AS window#110, count(1) AS count#121L]
+- Filter ((method#25 = GET) AND (responsecode#40 = 200))
+- 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@782fdeaa, 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@5d2c170a
+- Sort [Correct_count#124L DESC NULLS LAST], true
+- Project [window#110, count#121L AS Correct_count#124L]
+- Aggregate [hostname#14], [hostname#14 AS window#110, count(1) AS count#121L]
+- Filter ((method#25 = GET) AND (responsecode#40 = 200))
+- 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@782fdeaa, 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@5d2c170a
+- Sort [Correct_count#124L DESC NULLS LAST], true
+- Aggregate [hostname#14], [hostname#14 AS window#110, count(1) AS Correct_count#124L]
+- Project [split(logs#5, ,, -1)[1] AS hostname#14]
+- EventTimeWatermark timestamp#1: timestamp, 3 seconds
+- Project [logs#5, timestamp#1]
+- Filter ((split(logs#5, ,, -1)[3] = GET) AND (split(logs#5, ,, -1)[5] = 200))
+- Generate explode(split(value#0, , -1)), [0], false, [logs#5]
+- StreamingDataSourceV2Relation [value#0, timestamp#1], org.apache.spark.sql.execution.streaming.sources.TextSocketTable$$anon$1@782fdeaa, 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@5d2c170a
+- *(7) Sort [Correct_count#124L DESC NULLS LAST], true, 0
+- Exchange rangepartitioning(Correct_count#124L DESC NULLS LAST, 200), true, [id=#359]
+- *(6) HashAggregate(keys=[hostname#14], functions=[count(1)], output=[window#110, Correct_count#124L])
+- StateStoreSave [hostname#14], state info [ checkpoint = file:/tmp/temporary-f9fb39ce-1caa-4bde-a068-b38acb808fb3/state, runId = 61e1d4d7-e682-42ec-879a-83a32ac478c6, opId = 0, ver = 0, numPartitions = 200], Complete, 0, 2
+- *(5) HashAggregate(keys=[hostname#14], functions=[merge_count(1)], output=[hostname#14, count#133L])
+- StateStoreRestore [hostname#14], state info [ checkpoint = file:/tmp/temporary-f9fb39ce-1caa-4bde-a068-b38acb808fb3/state, runId = 61e1d4d7-e682-42ec-879a-83a32ac478c6, opId = 0, ver = 0, numPartitions = 200], 2
+- *(4) HashAggregate(keys=[hostname#14], functions=[merge_count(1)], output=[hostname#14, count#133L])
+- Exchange hashpartitioning(hostname#14, 200), true, [id=#347]
+- *(3) HashAggregate(keys=[hostname#14], functions=[partial_count(1)], output=[hostname#14, count#133L])
+- *(3) Project [split(logs#5, ,, -1)[1] AS hostname#14]
+- EventTimeWatermark timestamp#1: timestamp, 3 seconds
+- *(2) Project [logs#5, timestamp#1]
+- *(2) Filter ((split(logs#5, ,, -1)[3] = GET) AND (split(logs#5, ,, -1)[5] = 200))
+- Generate explode(split(value#0, , -1)), [timestamp#1], false, [logs#5]
+- *(1) Project [value#0, timestamp#1]
+- MicroBatchScan[value#0, timestamp#1] class org.apache.spark.sql.execution.streaming.sources.TextSocketTable$$anon$1
