== 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