Spark Structured Streaming, KQL Processing and Windowing Functions

What it is

Streaming data never "finishes", so processing works on small batches as they arrive and on time windows rather than whole tables. In Fabric you process streams with Spark structured streaming in a notebook or Spark job definition, or with KQL inside an Eventhouse.

Windowing functions group events into time intervals, such as "count of readings per device every five minutes". The same window types appear in Eventstream, Spark and KQL.

How it works

Spark structured streaming

A streaming DataFrame is read with readStream and written with writeStream. Spark processes new data as a series of micro-batches.

from pyspark.sql import functions as F

events = spark.readStream.table("bronze_telemetry")

per_device = (events
    .withWatermark("event_time", "10 minutes")
    .groupBy(F.window("event_time", "5 minutes"), "device_id")
    .agg(F.avg("temperature").alias("avg_temp"), F.count("*").alias("readings")))

query = (per_device.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "Files/checkpoints/telemetry_5min")
    .trigger(processingTime="1 minute")
    .toTable("silver_telemetry_5min"))

For a stream that must run for days, use a Spark job definition with a retry policy rather than an interactive notebook.

Process data by using KQL

In an Eventhouse, transformation happens at ingestion or as a precomputed view.

An update policy runs a query on every new batch in a source table and writes the result into a target table.

.create-or-alter function ParseTelemetry() {
    RawTelemetry
    | extend DeviceId = tostring(Payload.deviceId),
             Temperature = todouble(Payload.temperature),
             EventTime = todatetime(Payload.ts)
    | project DeviceId, Temperature, EventTime
}

.alter table Telemetry policy update
@'[{"IsEnabled": true, "Source": "RawTelemetry", "Query": "ParseTelemetry()", "IsTransactional": false}]'

A materialized view keeps an aggregation up to date in the background, so queries read precomputed results.

.create materialized-view TelemetryHourly on table Telemetry
{
    Telemetry
    | summarize AvgTemp = avg(Temperature), Readings = count() by DeviceId, bin(EventTime, 1h)
}

Use update policies for row-level parsing and enrichment, and materialized views for aggregates or "latest value per key" with arg_max.

Windowing functions

Syntax in each engine:

In KQL, bin() is the everyday tumbling window. Row functions such as prev(), next(), row_cumsum() and row_window_session() need serialized (sorted) rows.

Clicks
| sort by UserId asc, Timestamp asc
| extend SessionStart = row_window_session(Timestamp, 4h, 30m, UserId != prev(UserId))
| summarize Clicks = count(), Duration = max(Timestamp) - min(Timestamp) by UserId, SessionStart

Here a new session starts after 30 minutes of inactivity, after 4 hours in total, or when the user changes.

A worked example

Temperature readings arrive in a lakehouse bronze table, some up to five minutes late. The team needs five minute averages per device in a silver Delta table, with no duplicate windows after a restart.

A Spark structured streaming job reads the bronze table, sets withWatermark("event_time", "10 minutes"), groups by a five minute window, and writes in append mode with a checkpoint. Late readings within the watermark update their window before it is emitted, and the checkpoint prevents reprocessing after restarts.

How the exam asks about this

• "Count events in fixed, non-overlapping five minute intervals." A tumbling window, bin(Timestamp, 5m) in KQL.
• "Group a user's clicks until 30 minutes of inactivity." A session window.
• "Restart a streaming query without reprocessing data." Set a checkpointLocation.