Validating pointer-based Delta comparison architecture using flatMapGroupsWithState in Structured St

[VamsiDatabricks]

Hi everyone,

I’m leading an implementation where we’re comparing events from two real-time streams — a Source and a Target — in Databricks Structured Streaming (Scala).

Our goal is to identify and emit “delta” differences between corresponding records from both sides based on a common naturalId.

Here’s the high-level architecture we’ve designed:

• Both Source and Target streams (from Kafka/Event Hubs) are read as structured streaming datasets.

• Each event is parsed, hashed (SHA-256), and persisted as full JSON to Delta Lake (for durability, auditability, and replay).

• Only lightweight metadata (key, hash, timestamp, Delta pointer) is kept in Spark state.

• We use flatMapGroupsWithState with event-time timeout + watermarking to hold state per key until both sides arrive.

• Once both Source and Target events for a given key are available, we fetch their corresponding JSONs from Delta using the stored pointers, perform the comparison, emit a DeltaRecord, and clear the state.

• Late or missing events are automatically handled via watermark expiry, and deltas are written back to Delta for downstream consumption.

Here’s a simplified pseudocode snippet:

keyed.flatMapGroupsWithState[StateValue, DeltaRecord](
OutputMode.Append(),
GroupStateTimeout.EventTimeTimeout()
)(handleKey)

Could you please validate this approach for ,

• Any hidden pitfalls in production especially around Delta I/O under load, event skew, or watermarking.

• Whether others have adopted similar pointer-based approaches for large-scale streaming comparisons, and any tuning lessons learned.

Appreciate any feedback, design critiques, or optimization suggestions from those who’ve run this pattern at scale 🙏

Thanks,
Vamsi

#StructuredStreaming, #flatMapGroupsWithState, and #DeltaLake

[iyashk-DB]

The overall pattern is sound, but there are a few real production risks worth calling out.

Delta point reads inside the state function are your biggest bottleneck. When flatMapGroupsWithState (or its replacement transformWithState) fetches a JSON from Delta using a stored pointer, that read happens synchronously on the executor thread for that key. You lose all parallelism within a micro-batch for that group. At scale, if your comparison logic triggers even moderate Delta I/O latency, micro-batch processing times will blow up and you'll fall behind Kafka. A better pattern is to let the state function emit a "ready to compare" record containing both Delta pointers, then do the actual Delta point reads in a downstream join or flatMap where Spark can execute them in parallel across the full task pool.

The global watermark will be set by the slower of your two streams. Databricks (and Spark generally) uses the minimum watermark across all inputs by default. If your Source stream is fast and Target is even slightly delayed, all state timeouts and output are held until Target catches up. This is the correct behavior to avoid dropping records, but the effect is that your end-to-end latency is bounded by whichever stream is slower, not by your watermark threshold. If you're seeing chronic latency in one direction, you'll want to monitor this closely.

Event skew within a key is a real risk. If one key gets a burst of events (say, a high-traffic shipment ID), the state for that key accumulates until the timeout fires. The comparison is still correct, but the state for that one key can grow large, and on RocksDB (which is the default from DBR 17.3+ anyway) that means more disk I/O per checkpoint for those partitions.

flatMapGroupsWithState is the legacy API on your runtime. You're on DBR 17.3 LTS, where transformWithState is available and RocksDB is the default state store. flatMapGroupsWithState still works, but if you're designing this from scratch, transformWithState gives you better state type flexibility and is the direction Databricks is investing in. Not a blocker, just worth knowing before you're deep in production.

One more thing on watermark expiry and late events: the behavior you're describing (emit a mismatch record when only one side arrives before the watermark) is correct, but test your timeout logic carefully. EventTimeTimeout fires during micro-batches where the watermark advances past the key's timeout timestamp. If you have a very low-event period where no new data arrives to advance the watermark, timeouts don't fire. This is a common source of "stuck state" in low-traffic environments.