Structured Streaming on Databricks: Kafka, triggers, real-time mode and streaming tables

Parts 1 to 3 ran Kafka and 3 streams on your computer. This part moves them to Databricks; its code follows Databricks' docs, and none of it needs the local Kafka running.

The same code, 3 changes

A stream on Databricks is the PySpark you have already written. Only 3 things change:

1. Tables have Unity Catalog names, such as silver.orders, in the schemas of module 5.
2. The checkpoint goes in a volume, such as /Volumes/<catalog>/silver/checkpoints/orders, 1 folder for each stream, as module 6 did for files.
3. The trigger is one the compute allows. Serverless compute, which is all Free Edition has, runs only availableNow and the deprecated Once.

So part 3's latest_orders.py becomes, in a notebook, its merge_batch function unchanged and this:

catalog = spark.catalog.currentCatalog()
spark.sql("CREATE VOLUME IF NOT EXISTS silver.checkpoints")

query = (
    spark.readStream.table("bronze.order_events")
    .writeStream.foreachBatch(merge_batch)
    .option("checkpointLocation", f"/Volumes/{catalog}/silver/checkpoints/orders")
    .trigger(availableNow=True)
    .start()
)

with MERGE INTO silver.orders in merge_batch, and no stream_spark import, since a notebook has spark already. Leave out the trigger and the stream fails with INFINITE_STREAMING_TRIGGER_NOT_SUPPORTED: Spark's default is processingTime with an interval of 0, a stream that never stops, which serverless compute refuses.

Kafka on Databricks

A Kafka cluster that Databricks can reach is read with the same kafka format as part 1. What changes is how it signs in: Databricks supports Unity Catalog service credentials, SASL/SSL, and the cloud's own options for Amazon MSK, Azure Event Hubs and Google Cloud Managed Kafka. With a service credential:

orders = (
    spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "<bootstrap server>:9092")
    .option("subscribe", "orders")
    .option("databricks.serviceCredential", "<service credential name>")
    .option("startingOffsets", "earliest")
    .load()
)

From there, part 1's from_json and toTable are unchanged. Free Edition restricts outbound internet access to a limited set of trusted domains, so a Kafka cluster of your own may not be reachable from it, and the Kafka on your computer, listening on 127.0.0.1, never is. To practise there, stream from a Delta table, as the code above does: the checkpoint, the watermark and the MERGE behave the same whatever the source.

In SQL, the same read is read_kafka in a streaming table, which Databricks runs and checkpoints for you:

CREATE OR REFRESH STREAMING TABLE bronze.order_events AS
SELECT
  CAST(value::string:order_id AS INT) AS order_id,
  value::string:customer_id AS customer_id,
  value::string:status AS status,
  CAST(value::string:amount AS DECIMAL(10,2)) AS amount,
  to_timestamp(value::string:event_time) AS event_time
FROM STREAM read_kafka(
  bootstrapServers => '<bootstrap server>:9092',
  subscribe => 'orders',
  serviceCredential => '<service credential name>'
);

value::string:order_id reads a field of the JSON value. Streaming SQL like this runs only in a Lakeflow pipeline, as in module 7, or as a streaming table in Databricks SQL. A streaming table handles each input row once and appends it; for a source that updates or deletes records, Databricks' docs point to AUTO CDC, which plays the part of part 3's MERGE.

Triggers, and what they cost

The empty row is the costly one. Databricks warns that a stream with no trigger can make a high volume of cloud storage calls a day, and charges from the cloud with them. Real-time mode is for decisions that cannot wait a second, such as stopping a fraudulent payment; Databricks says it needs dedicated compute, so anything that can wait seconds or minutes costs less without it. Spark's own continuous trigger is not supported on Databricks, which points to real-time mode instead.

A stream that runs as a job

A stream that never stops runs as a Lakeflow job with the continuous schedule, on classic jobs compute, since serverless refuses a trigger that never stops. Databricks recommends it for every such stream: it allows only 1 run at a time, starts a new run when one fails, and waits longer between each retry. That restart is part 3's replay: the batch planned before the failure runs again, which is why every write in foreachBatch has to be safe to run twice. In a job, leave out awaitTermination(); the job already waits for the stream, and Databricks says the call disrupts its backlog metrics and notifications.

To see whether a Kafka stream keeps up, read its source metrics in the query's progress: avgOffsetsBehindLatest, maxOffsetsBehindLatest and minOffsetsBehindLatest say how many offsets the partitions are behind the newest event, and estimatedTotalBytesBehindLatest the bytes. A lag that grows run after run means the stream needs more compute, or more partitions, which the Kafka sizing calculator works out.

On Free Edition, with serverless only and at most 5 job tasks at once, the same stream runs as a job on a schedule with availableNow: every 15 minutes, or every hour, it loads what is new and stops.

When a stream is worth it

Most data does not need a stream that never stops. Play 1 day of orders through a nightly batch, an hourly one and a stream, and watch how far behind the dashboard runs:

Try itPlay a day of orders

The pipeline runs

NightlyHourlyAs data arrives

Play

00:00One day, midnight to midnight

00:0006:0012:0018:0024:00

Orders arrived

0

On the dashboard

0

Arrived, waitingOn the dashboardA batch run

An hourly availableNow job is the same code as a stream that never stops, with the same checkpoint and the same exactly-once, at a fraction of the compute. Choose a stream that never stops when someone acts on the data within minutes, and real-time mode when a machine must act within a second. You can move between them later by changing only the trigger, since the checkpoint stays the same.

Part 5 puts the module together: Kafka to bronze, silver and an hourly gold table, replayed from scratch.