A streaming pipeline can look correct until the first event arrives late. If a customer order occurred at 10:02 but did not reach Fabric until 10:18, a five-minute reporting window may have closed, remained open, or been updated depending on how event time, watermarks and state are configured. Microsoft DP-700 tests streaming engine choices and windowing functions because a data engineer must explain that behavior, not merely connect an Eventstream to a destination. This lab isolates Spark Structured Streaming’s stateful part of the problem from the message routing and real-time storage responsibilities of Eventstream and Eventhouse. It supports the Microsoft DP-700 exam.
Start by making two clocks explicit
Every test event should carry both event_time and arrival_time. Event time is when the activity occurred in the originating system; arrival time is when the data processing service received or observed it. They are not interchangeable. A point-of-sale device may remain offline and upload an order later; a telemetry gateway may buffer sensor records; a producer may resend the same event after an acknowledgement timeout. When users ask for revenue in five-minute periods or devices active during a shift, they normally mean business event time, not the arrival clock. An engineer must preserve both to explain backfill and delay.
Build a compact input fixture with a device identifier, unique event identifier and an ISO timestamp. Example records include D1 at 10:01 and 10:03, D2 at 10:04, another D1 record at 10:09, and a delayed D1 event whose event time is 10:02 but arrives after later windows have already been observed. Write the expected event-time group for each record before opening Spark. This is more useful than a stream of random numbers, because you can prove whether the aggregation respects the time encoded in the event.
Choose the streaming layer that actually solves the task
Eventstream receives, routes and optionally shapes incoming event streams. Eventhouse stores and queries event-oriented records using KQL. Spark Structured Streaming executes programmable incremental transformations, including stateful aggregations, joins and window functions. These layers can coexist, but an exam answer should identify which layer needs to perform the business rule. A requirement to route telemetry to a managed analytical store is not automatically a reason to deploy stateful Spark. A requirement to maintain a keyed aggregation across batches with explicit checkpoint recovery often is.
Eventstream and Eventhouse handle the live event path and persistent analytical storage; a Spark stateful stream has a different responsibility. Its event-time watermark governs when late observations can no longer update an aggregation, and its checkpoint persists progress and intermediate state across compatible restarts. If the business asks why a valid late order missed a five-minute window, inspect the event timestamp, the query’s watermark and the checkpoint rather than treating it as an Eventhouse retention failure. SQL and KQL help investigate stored events, but choosing a query language alone does not change stateful Spark window semantics.
Build a five-minute window with a ten-minute watermark
The following illustrative PySpark pattern assumes a Fabric lakehouse and an input directory containing JSON records that match the declared schema. It is a lab sketch that must be adapted to your actual lakehouse path, table names and runtime. The five-minute window groups events by their event timestamps. The ten-minute watermark expresses how long state should remain eligible for late data, relative to the maximum event time Spark has observed. Those two time intervals have different jobs and must not be confused.
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, TimestampType
schema = StructType([
StructField("device_id", StringType(), False),
StructField("event_id", StringType(), False),
StructField("event_time", TimestampType(), False)
])
events = (spark.readStream
.schema(schema)
.json("Files/stream-lab/incoming"))
counts = (events
.withWatermark("event_time", "10 minutes")
.groupBy(F.window("event_time", "5 minutes"), F.col("device_id"))
.count())
query = (counts.writeStream
.format("delta")
.outputMode("append")
.option("checkpointLocation", "Files/stream-lab/checkpoints/device-counts")
.toTable("device_window_counts"))Append output mode for a windowed aggregation emits a finished window once Spark has enough watermark progress to determine that the window is ready. If the stream receives no newer events, the watermark may not advance and the output might wait. This does not mean the pipeline is broken. To validate the lab, continue sending records whose event times advance the maximum observed time beyond a window’s end and the chosen delay. Compare output windows against the fixture and record when each window becomes final. Never assert that “ten minutes after wall-clock time” always produces the final row; watermark progress is based on observed event-time data.
What the watermark does — and does not — guarantee
A watermark lets the engine control how long it must retain state for potentially late data. With a ten-minute delay, records within the tracked lateness bound can still update an open window. Older events can be dropped from a closed stateful result; exact behavior depends on progress and the operator, and Microsoft cautions that rows older than the watermark do not receive the same guarantee. A longer delay retains more late arrivals but also keeps more state, increasing memory, checkpoint size and the delay before append-mode output is finalized. A short delay reduces state while risking missed legitimate late data.
Use three late-data experiments. First, send a normal event in a new window and verify it is included. Next, delay one record by a few minutes in event time but send it while its window is still retained. Finally, advance later event times enough that the old window is finalized, and then send a record timestamped far in the past. Capture whether it changes the result. Do not draw a general conclusion from a single batch: the stream’s maximum event time and the order of records matter. Your goal is a written rule for which lateness the business accepts.
Keep every stateful query’s checkpoint independent
The checkpointLocation is durable operational state. It stores information about processing progress and, for a stateful query, state-store data needed to resume after a restart. Do not reuse one checkpoint directory for two different streaming queries, even if they read the same source. Sharing locations can corrupt recovery metadata or cause confusing duplicates and skipped work. Do not place the checkpoint on ephemeral driver disk: the query cannot reliably resume after a restarted environment if its state disappeared with that process.
Stop the query after several microbatches, then restart a compatible version using its original checkpoint location. The expected result is resumed processing without replaying already committed work into the Delta sink. Checkpoint support and exactly-once guarantees are conditional on the source/sink and operation; custom foreachBatch output logic still needs idempotency because a batch can be retried. If you change the state schema, grouping keys, source or sink type substantially, test whether the old checkpoint remains compatible before deploying the change. Incompatible state should be migrated or restarted by an intentional plan, not by blindly deleting a checkpoint in production.
Use idempotent identifiers for downstream writes
Suppose a stream writes five-minute totals to a serving table, and a retry repeats the same microbatch. Your target should not contain two totals for the same device and time window merely because the write ran twice. If you use foreachBatch to perform an upsert, construct a deterministic merge key from the logical group and window start/end; include a mechanism to make repeated batch processing safe. Separate event deduplication by event_id from aggregate idempotency. The former determines which source records count; the latter determines whether a repeated write corrupts the serving result.
Batch and incremental data loading connects streaming silver-layer output with scheduled ingestion, replay safety and downstream serving requirements. A source replay may be correct, a stateful calculation may be correct, and the final reporting table can still be wrong if its merge semantics are not. Use a deliberately interrupted test and compare the final count with the same fixture processed without interruption. That is stronger evidence than a screenshot of a green streaming-job icon.
Choose the state store and control its growth
For stateful Spark workloads, Microsoft recommends considering a RocksDB state store instead of the default state kept principally in JVM-managed memory. RocksDB moves active state into native memory and local executor storage, while durable snapshots or changelogs are maintained in the stream checkpoint. This can reduce garbage-collection pressure with large keyed states, but it does not remove the need to bound the number of keys, time windows or buffered join rows. A ten-minute watermark and a five-minute grouping interval can still produce many state entries when the incoming data has millions of distinct device identifiers.
Monitor processing rate, input rate, microbatch duration and state size during the test. When a stream falls behind, trace source reads, shuffles, state updates, small-file writes and downstream waits against the same load curve. Short microbatch intervals may lower latency but generate excessive Delta files; longer intervals improve batching while sacrificing freshness. Fabric performance tuning also evaluates bottlenecks across compute, storage and queries; this experiment isolates stateful streaming behavior.
What changes with stream-stream joins
The state problem becomes more demanding when two live streams must be correlated. An order and a payment can arrive independently, sometimes with different delays. A safe join usually needs event-time bounds as well as watermarks on the appropriate streams, so Spark eventually knows which unmatched rows no longer need to be retained. Without a time-range condition, a high-volume join can accumulate state indefinitely. For the exam, identify the business correlation window, which timestamps represent meaningful event times, and what should happen when a payment arrives after the allowed interval.
Consider an order made at 10:00 and a payment received at 10:14 when the business permits a 30-minute payment interval. The payment’s system arrival time may be later still. Test both valid delayed joins and intentionally too-late events, and make the intended exclusion observable. Distinguish a stream-stream join from joining a stream to a relatively static dimension. They have different state and correctness implications; treating every join as a batch SQL join hides the operating cost and the late-data rules.
Failure drill — the query restarts with missing or incompatible state
Introduce a controlled restart during the lab and preserve the query’s last progress information. Then attempt an intentionally incompatible change such as altering stateful grouping keys. If the running query refuses to restore the old state, do not immediately remove the checkpoint and rerun against live production sources. Investigate the compatibility boundary, decide how much replay is available, and plan a new checkpoint and target recovery only when supported. If you need to replay retained event data into a new serving result, give that recovery run its own destination or deterministic merge behavior so it cannot silently double the original counts.
Monitoring, alerts and error recovery need the failed microbatch identifier, source offsets, event-time watermark, checkpoint path, destination version and before/after counts. A durable checkpoint can restore offsets and state for a compatible query, but it does not by itself guarantee an exactly-once external side effect; custom sinks and foreachBatch logic should be idempotent. Restart only with a checkpoint that matches the stateful plan, quarantine malformed records for review and demonstrate the expected business result rather than merely a cleared error banner.
A short streaming readiness test
- Produce events containing event time, arrival time, a stable event ID, and a deliberate out-of-order sequence. State the intended five-minute grouping results.
- Build a stateful aggregate with a justified watermark, durable checkpoint and reliable Delta sink. Verify when output windows become final.
- Restart the query using the same compatible state, and compare the result to an uninterrupted run; retry a write and prove the output is not duplicated.
- Inject a late record beyond the agreed bound and record the intentional policy: rejection, quarantine, correction flow or another agreed business treatment.
- Compare Spark structured streaming with Eventstream routing and Eventhouse analytical query requirements. Choose the simplest component that owns the actual rule.
This exercise covers more than remembering withWatermark syntax. It asks you to distinguish event time from processing time, define which business facts are late, decide how much state to retain, and make failure recovery provable. Those are the practical reasons Spark Structured Streaming appears in the DP-700 syllabus. If a candidate can predict the outcome of an out-of-order event and explain what a checkpoint does after a restart, the underlying principle is understood rather than memorized.