Build a sub-second operational dashboard on live streaming data, starting with a live Kafka stream
This blog is a full end-to-end walkthrough for practitioners on how to build real-time applications using Apache Spark™ Declarative Pipelines, Lakebase, and Databricks Apps. Databricks recently announced exciting news for anyone building streaming pipelines: Real-Time Mode (RTM) for Spark Declarative Pipelines is now in Public Preview. This means you can build continuous pipelines with end-to-end latencies as low as five milliseconds, without the complexity and cost of managing separate engines.
With Spark Declarative Pipelines you simply declare the flows you want, and the framework keeps them running correctly: handling orchestration, checkpoints, retries, and state so your code focuses on business logic rather than streaming plumbing.
Spark Declarative Pipelines (SDP) did not previously offer operational, sub-second latency. The default execution mode is excellent for high-throughput ETL workloads that can tolerate latencies ranging from a few seconds to minutes, but not for in-the-loop workloads where a decision must land in tens of milliseconds, such as fraud scoring, live personalization, or real-time alerting. For those use cases, teams were forced to add a specialized engine, such as Apache Flink, to meet their needs, which inherited a whole stack to run alongside it.
Real-Time Mode eliminates this added complexity. It targets p99 latency in the tens to low hundreds of milliseconds, and now you get it inside the declarative model: the same sinks-and-flows authoring surface, on serverless. You keep "declare what, not how" and gain real-time speed in one engine.
In this blog, we’re diving deep into what this new capability means for practitioners. To show RTM on SDP in action, we’ll walk through a demo that tracks aircraft positions, demonstrating how you can easily build, monitor, and scale sub-second operational workloads inside the Spark Declarative Pipelines you already know.
Demo: An end-to-end operational pipeline from Kafka to Lakebase
In this demo, live aircraft positions stream into Apache Kafka. A Spark Declarative Pipeline running with Real-Time Mode reads the stream, enriches each position (a status column, an alert flag, a zone label), and counts how many aircraft are inside each monitored zone at once, raising an alert when a zone gets crowded. Both outputs are written to Databricks Lakebase (managed Postgres), which a Databricks App reads to render a live map. This is an end-to-end flow: a plane moves, and the dot on the map moves. The aviation details are just the use case; what we are really showcasing is the code.
The programming model: sinks + update flows
Real-Time Mode pipelines are authored with the pyspark.pipelines API (imported as dp). Two primitives do the work, and notice what is absent: no writeStream, no awaitTermination, no checkpoint paths. You declare; the framework runs it.
- Declare an external sink. Real-Time Mode is built for operational delivery, so rather than landing results in a table for analytics, you write straight to the operational system your application reads from. Here we write to Databricks Lakebase through a native external sink:
from pyspark import pipelines as dp # Native Lakebase sink (Private Preview). The configuration API might change # before GA, so the exact options are omitted here. dp.create_sink( name="lakebase_sink", format=..., # native Lakebase sink (Private Preview) options={...}, # connection + target table ) -
Define an update flow that targets the sink. An
@dp.update_flowreturns a streaming DataFrame, and its rows are routed to the named sink. The flow is where you turn RTM on, through the flow-level Spark configpipelines.trigger: "RealTime". The next two sections show two flows that do exactly this, one stateless and one stateful.Three things to internalize:
- pipelines.trigger: "RealTime" is the switch. Combined with the pipeline-level
spark.databricks.streaming.realTimeMode.enabled, this moves the flow onto the continuous engine. - pipelines.trigger.interval is not a micro-batch size. In RTM the batch is long-running, and data is processed as it arrives. The interval governs how often state and source offsets are checkpointed, not how often results appear.
- The pipeline itself is configured
serverless: true,continuous: true, on thePREVIEWchannel.
That is the whole shape: declare sinks, write flows, flip the trigger. Everything that used to be plumbing (checkpoints, recovery, execution order) stays the framework's job. You stay in business logic.
- pipelines.trigger: "RealTime" is the switch. Combined with the pipeline-level
The stateless flow: enrich each event
The first flow reads rows from Kafka, parses the JSON, drops incomplete rows, and adds a few derived columns: a flight-phase label, an alert flag for emergency transponder codes, and a zone label. It is ordinary DataFrame code, and Real-Time Mode does not change how you write it. The essence, with the Kafka boilerplate trimmed:
@dp.update_flow(
name="positions_flow",
target="lakebase_sink", # the Lakebase sink declared above
spark_conf={"pipelines.trigger": "RealTime", # turn RTM on
"pipelines.trigger.interval": "5 minutes"},
)
def positions_flow():
return (
spark.readStream.format("kafka")
.option("subscribe", "raw-flights")
# ... bootstrap servers + SASL auth ...
.load()
.select(from_json(col("value").cast("string"), RAW_SCHEMA).alias("d"))
.select("d.*")
.filter(col("lat").isNotNull() & col("icao24").isNotNull())
# a few stateless transformations
.withColumn("flight_phase",
when(col("vert_rate") > 500, "climbing")
.when(col("vert_rate") < -500, "descending")
.when(col("baro_alt") > 30000, "cruise")
.otherwise("en_route"))
.withColumn("squawk_alert",
when(col("squawk").isin("7500", "7600", "7700"), "emergency"))
.withColumn("zone", zone_label()) # which geofence box, if any
)Nothing here is RTM-specific. The same DataFrame you would write for a micro-batch flow runs unchanged. The only thing that made it real-time was the pipelines.trigger: "RealTime" line in the decorator.
Going stateful: transformWithState in RTM
The second flow answers a question you cannot answer from any single record: how many aircraft are inside each monitored zone at the same time, and when does a zone become too crowded? A zone here is just a geographic box, like a sector an air-traffic controller watches. Answering this means remembering which aircraft are currently in each zone, so it needs state. Real-Time Mode supports the Arbitrary Stateful Processing API, transformWithState, through the StatefulProcessor class.
We built this as a single keyed stateful operator: one shuffle, one operator, one hop on the real-time path. The processor, ZoneCounter, is keyed by zone and keeps a MapState mapping each aircraft id to the last time we saw it in that zone. Each incoming position refreshes that timestamp. A recurring timer then sweeps out any aircraft we have not seen for a while (it has left the zone or gone quiet) and re-emits the updated count, so a busy zone that empties out clears on its own. The essence of the operator and how it is wired in:
class ZoneCounter(StatefulProcessor):
def init(self, handle):
self.handle = handle
self._aircraft = handle.getMapState("aircraft", KEY_SCHEMA, VAL_SCHEMA) # icao24 -> last seen
self._timer = handle.getValueState("sweepTimer", TIMER_SCHEMA)
def handleInputRows(self, key, rows, timerValues):
now_ms = timerValues.getCurrentProcessingTimeInMs()
for r in rows: # called once per row in RTM
self._aircraft.updateValue(Row(icao24=r.icao24), Row(last_ms=now_ms))
active = self._sweep(now_ms) # drop aircraft not seen within the TTL
self._ensure_timer(now_ms) # keep exactly one pending sweep timer
yield self._alert_row(key[0], active, now_ms / 1000.0)
def handleExpiredTimer(self, key, timerValues, expiredTimerInfo):
now_ms = timerValues.getCurrentProcessingTimeInMs()
active = self._sweep(now_ms) # re-emit the refreshed count
if active:
self._ensure_timer(now_ms)
yield self._alert_row(key[0], active, now_ms / 1000.0)
# Wire the operator into a second flow: filter to in-zone positions, group by
# zone, run ZoneCounter, and write the alerts to a (second) Lakebase table.
@dp.update_flow(
name="zone_congestion_flow",
target="zone_alerts_sink", # a second Lakebase sink, declared like the first
spark_conf={"pipelines.trigger": "RealTime", # turn RTM on
"pipelines.trigger.interval": "5 minutes"},
)
def zone_congestion_flow():
in_zone = (
spark.readStream.format("kafka")
.option("subscribe", "raw-flights")
# ... bootstrap servers + SASL auth ...
.load()
.select(from_json(col("value").cast("string"), RAW_SCHEMA).alias("d"))
.select("d.*")
.withColumn("zone", zone_label())
.filter(col("zone").isNotNull()) # only in-zone positions feed the counter
.select("zone", "icao24", "poller_ts")
)
# one shuffle: groupBy(zone) -> one stateful operator -> congestion alerts
return (
in_zone.groupBy("zone")
.transformWithState(
statefulProcessor = ZoneCounter(),
outputStructType = ALERT_SCHEMA,
outputMode = "update",
timeMode = "processingTime",
)
)Two patterns here are worth lifting into your own code:
Idempotent timers. _ensure_timer registers a sweep timer only if one is not already pending. Because handleInputRows is invoked per row in RTM, a naive "register a timer on every row" would pile up thousands of redundant timers. Keeping exactly one pending timer in a ValueState is the safe pattern. When a zone empties, the timer stops re-arming, and the next arriving position re-arms it.
Staleness-based eviction. An aircraft that crosses out of a zone simply stops appearing in that zone's stream; there is no "leave" event to react to. So the sweep evicts any aircraft not seen within a short TTL and re-emits the count on a recurring processing-time timer. This keeps the entire computation in a single stateful operator on the real-time path, exactly the kind of design that plays to Real-Time Mode's strengths.
Measuring latency and monitoring
Real-Time Mode reports per-record latency at p50, p90, p95, and p99 (docs). Three numbers matter:
processingLatencyMs: from reading a record to writing it downstream.sourceQueuingLatencyMs: from message-bus write (e.g. Kafka log-append) to first read.e2eLatencyMs: from message-bus write to writing downstream.
These are emitted in the streaming query progress, which the pipeline writes to its driver logs as JSON on each reporting interval. Search the driver log for latencies and you will find a block like this, from a run of our demo pipeline:
"latencies": {
"processingLatencyMs": {
"P0": 2,
"P50": 134,
"P90": 135,
"P95": 135,
"P99": 135
},
"sourceQueuingLatencyMs": {
"P0": 0,
"P50": 0,
"P90": 0,
"P95": 1,
"P99": 1
},
"e2eLatencyMs": {
"P0": 2,
"P50": 134,
"P90": 135,
"P95": 135,
"P99": 135
}
}e2eLatencyMs covers the engine path from the message bus to the query’s downstream write. Watch the percentiles, not the average; Real-Time Mode delivers, particularly at p99.
Getting started with Real-Time Mode on SDP
- Create the pipeline. A serverless Lakeflow Spark Declarative Pipeline on the
PREVIEWchannel withcontinuous: true. Setspark.databricks.streaming.realTimeMode.enabled: true - Point it at Apache Kafka. Your streaming source is a Kafka topic; read it with
spark.readStream.format("kafka")inside the flow. - Declare a sink and a flow.
dp.create_sink(...)for your destination, then an@dp.update_flow(target=...)withpipelines.trigger: "RealTime". - Add state only where you need it, with
transformWithStateand theStatefulProcessorclass. The idempotent-timer and staleness-eviction patterns above keep stateful logic in a single operator on the real-time path. - Measure with the latency percentiles before and after, so you can prove the win.
The full flight-tracker pipeline (both flows and the Lakebase sink) is on GitHub.
Conclusion
Real-Time Mode delivers sub-second, continuous processing to Spark Declarative Pipelines without sacrificing the declarative model. You keep the authoring surface you already use, declare what, not how, and gain real-time speed all on one engine. The flight tracker shows the whole arc: a stateless enrichment flow, a stateful operator with a real-time-aware timer pattern, and external delivery to an operational store. Try it on a workload where seconds are too slow, and watch your p99.
Ready to break the microbatch barrier? Check out these resources:
- Real-Time Mode docs
- Breaking the micro-batch barrier: the architecture of Real-Time Mode
- Why data engineering needs to become end-to-end declarative
- Demo: "Getting started with Real-Time Mode"
The external sink that writes to Databricks Lakebase in this demo is currently in Private Preview. Its configuration API might change before general availability, which is why the sink's exact options are omitted from the code above. If you'd like to try the native sink while it's in Private Preview, reach out to Databricks through your account team.
