Set up real-time mode

This page describes how to run real-time mode queries directly with the Structured Streaming APIs on classic compute. For a step-by-step tutorial, see Tutorial: Run a real-time streaming workload. For conceptual information about real-time mode, see Real-time mode concepts.

Real-time mode in Lakeflow pipelines supports managed pipelines on classic and serverless compute, with separate runtime, source, sink, and operator requirements.

Prerequisites

To use real-time mode, you must configure your compute to meet the following requirements:

Use classic compute. Dedicated and standard access modes are supported. Standard access mode is supported for Python only. The direct query configuration on this page is not supported in serverless notebooks or jobs. For serverless real-time processing, use real-time mode in Lakeflow pipelines.
Use Databricks Runtime 16.4 LTS and above.
Turn off autoscaling.
Turn off Photon.
Set spark.databricks.streaming.realTimeMode.enabled to true.
Turn off spot instances to avoid interruptions.

For latency-sensitive workloads with UDFs, Databricks recommends that you use dedicated access mode. See Table functions.

For instructions on creating and configuring classic compute, see Compute configuration reference.

Stream to stream joins

Stream to stream inner joins require additional configuration for real-time mode. Stream-to-stream outer joins aren't supported. See Stream to stream join.

Important

To run a stream to stream join in real-time mode with multiple other streams on the same cluster, you must use Databricks Runtime 18 LTS and above.

To enable stream to stream joins for real-time mode, set the following Spark configurations:

Python

spark.conf.set("spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled", "true")
spark.conf.set("spark.sql.streaming.join.stateFormatVersion", "4")
spark.conf.set("spark.sql.streaming.join.stateFormatV4.enabled", "true")
spark.conf.set("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion", "2")
spark.conf.set("spark.sql.streaming.realTimeMode.controlMessage.enabled", "true")

Scala

spark.conf.set("spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled", "true")
spark.conf.set("spark.sql.streaming.join.stateFormatVersion", "4")
spark.conf.set("spark.sql.streaming.join.stateFormatV4.enabled", "true")
spark.conf.set("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion", "2")
spark.conf.set("spark.sql.streaming.realTimeMode.controlMessage.enabled", "true")

SQL

SET spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled = true;
SET spark.sql.streaming.join.stateFormatVersion = 4;
SET spark.sql.streaming.join.stateFormatV4.enabled = true;
SET spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion = 2;
SET spark.sql.streaming.realTimeMode.controlMessage.enabled = true;

Query configuration

To run a query in real-time mode, you must enable the real-time trigger. Real-time triggers are supported only in update mode.

Python

query = (
    spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("subscribe", input_topic)
        .load()
        .writeStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("topic", output_topic)
        .option("checkpointLocation", checkpoint_location)
        .outputMode("update")
        # In PySpark, the realTime trigger requires specifying the interval.
        .trigger(realTime="5 minutes")
        .start()
)

Scala

import org.apache.spark.sql.streaming.Trigger

val readStream = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokerAddress)
      .option("subscribe", inputTopic).load()
      .writeStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokerAddress)
      .option("topic", outputTopic)
      .option("checkpointLocation", checkpointLocation)
      .outputMode("update")
      .trigger(Trigger.RealTime())
      // Trigger.RealTime can also accept an argument specifying the batch duration.
      // For example, this code sets a batch duration of 5 minutes:
      // .trigger(Trigger.RealTime("5 minutes"))
      .start()

Real-time watermark propagation

In Databricks Runtime 19 and above, real-time mode supports real-time watermark propagation. Each stateful operator advances its watermark, cleans up state, fires event-time timers, and passes the watermark downstream in real time, instead of only at batch boundaries.

Real-time watermark propagation is on by default for new real-time mode queries in recent Databricks Runtime 19 releases. To go back to the previous behavior, in which the watermark advances only at batch boundaries, set this Spark configuration to false before you start the query. You can set it in the session, as the following example shows, or in the Spark configuration of the compute:

Python

spark.conf.set("spark.databricks.streaming.realTimeMode.realTimeWatermark.enabled", "false")

Scala

spark.conf.set("spark.databricks.streaming.realTimeMode.realTimeWatermark.enabled", "false")

SQL

SET spark.databricks.streaming.realTimeMode.realTimeWatermark.enabled = false;

Important

Turning on real-time watermark propagation by default doesn't affect existing queries. A query keeps the setting it started with, including after a restart from the same checkpoint. To change the setting for an existing query, restart the query from a new checkpoint.

For new stateful queries, you might see more aggressive late-event filtering and shorter state retention than with the previous semantics. Validate in a pre-production environment before you run new queries in production.

For how real-time watermark propagation affects stateful operators, see transformWithState in real-time mode and Multi-step stateful queries.

Min-of-max watermark

When real-time watermark propagation is on, real-time mode computes the watermark with min-of-max semantics for all sources that real-time mode supports. For each withWatermark, Databricks tracks the maximum event time seen in each source partition. The watermark is the minimum of these per-partition maximums, minus the watermark delay. The watermark advances only as fast as the slowest source partition, so events from a slower partition aren't dropped as late because a faster partition moved the watermark ahead.

For most sources, each Spark input partition is a source partition. Apache Kafka sources are an exception: Databricks tracks each Kafka topic partition as a source partition, even when one Spark input partition reads several of them. For example, if you set maxPartitions lower than the number of Kafka topic partitions, a slow topic partition still holds back the watermark when it shares a Spark input partition with faster ones.

Databricks persists only one watermark value between batches, not one value for each source partition. If a source partition receives no data during a batch, the watermark doesn't advance in that batch.

When real-time watermark propagation is off, real-time mode uses max-of-max semantics instead, as micro-batch mode does. The watermark is the maximum event time across all source partitions, minus the watermark delay.

If a stateful operator has more than one withWatermark input, such as a stream to stream join, Databricks combines their watermarks with the multiple watermarks policy.

Query shapes that aren't supported

When real-time watermark propagation is on, Databricks rejects the following query shapes when it plans a batch, before any tasks run:

A shuffle before withWatermark, such as repartition(n).withWatermark(...). The query fails with the SHUFFLE_NOT_SUPPORTED_BEFORE_WATERMARK error.
More than one shuffle between a stateful operator and the closest withWatermark or stateful operator before it in the query. Databricks checks each input of a stateful operator separately, so each input of a join can have one shuffle. For example, withWatermark(...), then repartition(2), then groupByKey(...).transformWithState(...) has two shuffles: the one that repartition adds and the one that groupByKey adds. The query fails with the MULTIPLE_SHUFFLES_NOT_SUPPORTED error.

If your workload needs one of these query shapes, contact Azure Databricks support.

Compute sizing

You can run one real-time job per compute resource if the compute has enough task slots.

To run in low-latency mode, the total number of available task slots must be greater than or equal to the number of tasks across all query stages.

Slot calculation examples

Pipeline type	Configuration	Required slots
Single-stage stateless (Kafka source + sink)	`maxPartitions` = 8	8 slots
Two-stage stateful (Kafka source + shuffle)	`maxPartitions` = 8, shuffle partitions = 20	28 slots (8 + 20)
Three-stage (Kafka source + shuffle + repartition)	`maxPartitions` = 8, two shuffle stages of 20 each	48 slots (8 + 20 + 20)

If you don't set maxPartitions, use the number of partitions in the Kafka topic.

Additional resources

Σχόλια

Ήταν χρήσιμη αυτή η σελίδα;

Τελευταία ενημέρωση στις 2026-11-10

Set up real-time mode

Prerequisites

Stream to stream joins

Python

Scala

SQL

Query configuration

Python

Scala

Real-time watermark propagation

Python

Scala

SQL

Min-of-max watermark

Query shapes that aren't supported

Compute sizing

Slot calculation examples

Additional resources

Σχόλια

Πρόσθετοι πόροι