Summary
- Explore how Real-Time Mode for Apache Spark Structured Streaming on Databricks now supports inner stream-stream joins in update mode with sub-second latency.
- See it in action on a real-time ad click attribution pipeline that joins clicks back to impressions at ~30K events/sec.
- Learn how switching from Micro-Batch to Real-Time Mode is a single trigger change on the same code, delivering ~16× faster p99 latency for stream-stream joins.
Joins are essential for correlating impressions with clicks, orders, and payments in ad revenue attribution. Implementing them with low latency is a major challenge.
Apache Spark Micro-Batch Mode (MBM) reliably delivers 2–5 second end-to-end latency. However, for a growing class of operational use cases where every millisecond drives a decision, seconds are too slow. Real-Time Mode (RTM) was built for exactly this, and it now supports stream-stream joins, one of the most common stateful operations in streaming. In this blog, we'll show it in action with a classic stream-stream join example: ad click attribution pipeline.
Use case overview: real-time ad click attribution
Two streams:
- Impressions: an ad is displayed to a user.
- Clicks: the user clicks that ad.
You join clicks back to impressions on impression_id, within a short time window. In our demo we used 2 minutes. We generate impressions and clicks at a ~10% match rate.
The architecture looks like this:
Why stream-stream join support in RTM matters
Before this, joining two streams in Spark came down to a trade-off between latency and how much you had to build yourself:
- Micro-Batch Mode (MBM): Already supports stream-stream joins, you write the join as a simple DataFrame expression and the engine handles the state. It just doesn't guarantee sub-second latency (typically 2–5 seconds), which is fine for the majority of streaming use cases such as ETL, analytics, dashboards, but not suitable for operational use cases like real-time alerting, fraud detection, and personalization.
- Real-Time Mode (RTM): Delivered the sub-second latency these use cases need, but didn't support stream-stream joins. To join two streams in RTM, teams funneled both topics into a single transformWithState operator and built the join themselves. It works, but it's custom, stateful code you have to write, tune, and maintain.
Now RTM supports stream-stream joins natively. You get sub-second latency with the same join semantics, and none of the hand-rolled transformWithState logic.
Implementation deep dive
A stream-stream join in Real-Time Mode works exactly like it does in micro-batch: the same DataFrame API, the same watermarks, the same join condition. The only difference is latency. Here's what the pipeline does.
Setting the watermarks
Both streams carry an event-time column, and each gets a watermark:
impressions = impressions.withWatermark("impression_time", "5 minutes")
clicks = clicks.withWatermark("click_time", "5 minutes")
A watermark tells the engine how late data is allowed to arrive. For a stream-stream join, the engine tracks a single globalWatermark = the minimum of the two input watermarks. That globalWatermark drives two things: it handles dropping late-arriving records, and it triggers state cleanup (eviction). Because it's the minimum of the two, a stream running ahead can't prematurely drop records or evict state. The slower stream keeps the faster one from evicting the state too early.
The join: how state stays bounded
attributed = (
impressions.alias("impressions").join(
clicks.alias("clicks"),
expr(
"""
impressions.impression_id = clicks.impression_id AND
click_time >= impression_time AND
click_time <= impression_time + interval 2 minutes
"""
),
"inner",
)The 2-minute time bound is what lets the engine keep state bounded. Each row is held only as long as a matching row from the other side could still arrive:
- An impression can still be matched by a click up to 2 minutes later, so it's kept until globalWatermark − 2 minutes, then evicted.
- A click's impression is always in the past, it matches immediately or never, and it's evicted at globalWatermark (no extra hold).
The 2-minute window here is specific to this demo, choose a window that fits your use case. Keep in mind that a wider window holds rows in state longer, so state size grows accordingly.
The Real-Time Mode trigger
attributed.writeStream
.queryName("adclick-attribution")
.format("kafka")
.option("kafka.bootstrap.servers",<KAFKA_BOOTSTRAP>)
.option("topic", output_topic)
.option("checkpointLocation", checkpoint_path)
.outputMode("update")
.trigger(**({"realTime": "5 minutes"} if mode == "RTM" else {"processingTime": "0 seconds"}))
.start()The join runs in update output mode, and the trigger is the only difference between the two modes: realTime : "5 minutes" runs the query continuously in Real-Time Mode, checkpointing the stream every 5 minutes; processingTime : "0 seconds" runs it as micro-batch, starting each new batch as soon as the previous one finishes. Everything else stays the same: the join, the watermarks and the output.
Note: RTM stream-stream joins currently support inner joins in update output mode. See the Real-Time Mode reference for the full list of supported operators, sources, and sinks.
Latency: Real-Time Mode vs Micro-Batch Mode
To make this reproducible, we built a companion repo that generates the data, replays it into Kafka at a controlled rate, and runs the join in either mode, so you can measure the latency difference yourself. We ran it under sustained load for 1 hour at a ~10% click-to-impression ratio, and compared the two.
Throughput
To validate the pipeline under realistic load, we tested with the following sustained throughput:
|
Metric (per second) |
Value |
|
Impressions |
~27k |
|
Clicks |
~2.8k |
|
Combined input rate |
~30k |
|
Attributed clicks |
~2.5k |
Compute: 3 × r6gd.2xlarge workers + 1 driver (24 worker vCPUs), on DBR 18 LTS.
Latency
Latency is measured end to end, from the click's Kafka timestamp to the attributed record landing on the output Kafka topic (attributed_clicks). On the exact same code and cluster, changing only the trigger, Real-Time Mode achieves 167 ms p99 latency, about 16× faster than micro-batch mode.
Why Real-Time Mode is faster
In micro-batch mode, each batch executes its stages sequentially. In this case there are two stages (Kafka read, then shuffle for the join). But the bigger latency problem isn't within a single batch, it's across batches: batch N must fully complete before batch N+1 begins. With each batch taking about 1 second, a record that arrives just after batch N starts isn't picked up until batch N+1, so it can take up to 2 seconds from arriving on the input topic to landing on the output topic. That leads to an end-to-end latency spike.
Real-Time Mode executes the stages in parallel: a record is processed and joined as soon as it arrives. That's the ~16× p99 difference.
Conclusion
Joining two streams is one of the most common patterns in streaming, and now in Real-Time Mode, you can do it with sub-second latency. It is supported for inner joins, in update output mode, starting DBR 18 LTS.
And the pattern isn't limited to ad click attribution. Any workload that correlates two or more streams can run the same way: an inner, time-bounded join, in Real-Time Mode.
If you're running a stream-stream join in micro-batch and need lower latency, or maintaining a custom transformWithState implementation to achieve the same in RTM, try the built-in stream-stream join instead. Switching is a single trigger change, with no rewrites.
Try it yourself:
- Companion repo: the full ad click attribution pipeline with a data generator; run it and compare MBM vs RTM latency yourself.
- Real-Time Mode documentation: setup, execution modes and supported operators.
