Databricks Spark Developer Associate: A Streaming Job That Never Catches Up - and Silently Drops Data

A Streaming Job That Never Catches Up - and Silently Drops Data

A Structured Streaming job reads a 12-partition Kafka topic peaking at 9,000 messages/s and writes to Delta through foreachBatch, running a MERGE on message id. Configuration: trigger(processingTime='10 seconds'), no maxOffsetsPerTrigger, spark.sql.shuffle.partitions=200, and withWatermark('event_time', '30 minutes') followed by dropDuplicates('event_id') that the team believes performs one-hour deduplication. Each micro-batch takes about 48 seconds to process, numInputRows per batch grows roughly 8% per hour, and the state store on disk has reached 40 GB and keeps climbing. Events arriving more than 30 minutes late never reach the target table even though they are present in Kafka. No exceptions appear in the logs.

1. Work out whether this job can ever catch up, and give the levers you would change in priority order with the reasoning behind each.

Read the worked answer

It cannot catch up, and the trigger interval is a fiction.

Arithmetic: the trigger asks for a batch every 10 s, but a batch takes 48 s. Structured Streaming runs micro-batches serially - the next batch starts only after the previous commits. Effective cadence is ~48 s, not 10 s. Meanwhile input arrives continuously at 9,000/s, so each batch must consume ~430k records and the backlog that accumulated while it was processing. Growing numInputRows per batch (+8%/h) is the backlog being pulled in, not a change in the source rate.

Lever order:

Make the trigger honest. Set trigger(processingTime='60 seconds') so cadence matches achievable latency; this stops the pointless per-batch overhead of starting work you cannot finish, and makes the metrics interpretable.
Bound the batch with maxOffsetsPerTrigger. Pick it from measured throughput, not from the source rate: if a batch can be processed in 60 s, cap at (records/s sustainable x 60). This converts an unbounded, growing batch into a stable one - the job then lags by a predictable amount instead of diverging.
Attack the real parallelism ceiling. The source read stage has at most 12 tasks, one per Kafka partition - spark.sql.shuffle.partitions=200 does nothing for the read. If 12 tasks cannot carry the rate, no amount of shuffle tuning helps; you must add Kafka partitions (or accept the lag).
Make the sink cheap per batch. A foreachBatch MERGE over 200 arbitrary shuffle partitions rewrites many small files. repartition inside foreachBatch on the MERGE key/partition column before merging reduces files touched and rewrites - a change that often halves batch duration with no added hardware.

Raise capacity only after 1-4; otherwise you buy throughput and give it straight back to overhead.

2. Why do events more than 30 minutes late disappear, and does the current code really deduplicate over a one-hour window?

Read the worked answer

No, it does not - the watermark defines both the lateness tolerance and the dedup window.

withWatermark('event_time', '30 minutes') maintains a global watermark at (max event time seen - 30 min). Any record with event time below that bound is treated as too late and is dropped by stateful operators downstream - which is exactly the observed behaviour: the data is in Kafka, it just never survives the query.
dropDuplicates('event_id') under a watermark keeps deduplication state only until the watermark passes. State older than the watermark is expired, so an event_id seen 45 minutes ago is no longer remembered - the "one-hour dedup" is really a 30-minute dedup plus silent data loss beyond 30 minutes.

Options, with tradeoffs:

Widen to '1 hour' if the business genuinely tolerates hour-old events: dedup then covers an hour, at the cost of roughly double the dedup state.
Deduplicate in the sink (the more robust answer given a MERGE already exists): drop the streaming dropDuplicates, keep a watermark sized to the sink's ability to handle late arrivals, and let MERGE on event_id be the authoritative idempotency barrier. Then lateness policy is explicit and durable, and state stops being a dedup history store.
Never set the watermark purely to shrink state if the semantic requirement is a full hour of coverage - those two goals conflict, and the watermark always wins silently.

3. The state store is 40 GB and growing. Given this code, what are the likely causes and what would you inspect to distinguish them?

Read the worked answer

Three candidate causes, distinguishable by inspection:

Watermark stalls at the slowest partition. The global watermark is derived from the minimum across partitions. If one Kafka partition goes quiet or is assigned to a slow task, the watermark stops advancing, state expires never, and the store grows while the job looks healthy. Check: per-partition numInputRows and the offsets/watermark in StreamingQueryProgress; look for a partition reporting zero rows while others progress.
The watermark is not on the path of the stateful operator. If withWatermark is applied to a different DataFrame than the one dropDuplicates/aggregation runs on, or after an aggregation, no event-time bound reaches the state store and retention is unbounded. Check: df.explain(True) - confirm the watermark operator sits below the stateful operator (StateStoreRestore / Deduplicate), not above or on a sibling branch.
The key space is genuinely huge. 40 GB may simply be the honest size of event_id state for your rate and watermark. Check: state rows and per-partition state size in the Streaming tab; divide by distinct event_ids per 30-minute window.

Also verify the provider: with the RocksDB state store, on-disk size is expected to be larger than the in-memory figure and compaction lag can inflate it. Only after ruling out (1) and (2) is the answer 'add capacity or shorten the watermark'.