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

On Databricks a stream is the same PySpark code with 3 changes: tables named in Unity Catalog, the checkpoint in a volume, and a trigger the compute allows. Serverless compute, which is all Free Edition has, runs only availableNow and Once, so a stream there runs as a scheduled job that loads what is new and stops. Kafka is read with the kafka format and a service credential, or in SQL with read_kafka in a streaming table. A stream that never stops runs on classic compute or in a continuous pipeline, as a continuous job that Databricks restarts when it fails; real-time mode cuts latency below a second on dedicated compute.

- Page: https://tecyfy.com/data-engineering/course/streaming/on-databricks
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 9: Streaming, Streaming on Databricks
- Parts of this module: Kafka into a Delta table (https://tecyfy.com/data-engineering/course/streaming); Event time, windows and watermarks (https://tecyfy.com/data-engineering/course/streaming/event-time); Exactly-once and foreachBatch (https://tecyfy.com/data-engineering/course/streaming/exactly-once); Streaming on Databricks (https://tecyfy.com/data-engineering/course/streaming/on-databricks); Build it: orders from Kafka to an hourly table (https://tecyfy.com/data-engineering/course/streaming/build-it)

At the end of this part you can:

- Move a local Structured Streaming job to Databricks, with Unity Catalog tables, a checkpoint in a volume and a trigger serverless compute allows
- Read Kafka on Databricks in Python with a service credential, and in SQL with read_kafka in a streaming table
- Choose between availableNow on a schedule, a stream that never stops and real-time mode, from the latency the data needs and what each costs

Parts [1](/data-engineering/course/streaming) to [3](/data-engineering/course/streaming/exactly-once) 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](/data-engineering/course/unity-catalog).
2. **The checkpoint goes in a volume**, such as `/Volumes/<catalog>/silver/checkpoints/orders`, 1 folder for each stream, as [module 6](/data-engineering/course/getting-data-in) 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:

```python
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:

```python
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:

```sql
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](/data-engineering/course/medallion-pipelines), 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

| Trigger | What it does | Where |
| --- | --- | --- |
| `availableNow=True` | Reads what is there, in as many batches as it needs, and stops | Everywhere; the one Databricks recommends on serverless |
| `processingTime="1 minute"` | Starts a batch every minute, for ever | Classic compute |
| none | `processingTime` with an interval of 0: checks for new data every few milliseconds | Classic compute; fails on serverless |
| `realTime="5 minutes"` | Real-time mode: latency below a second | Dedicated compute, and Lakeflow pipelines |

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](/data-engineering/kafka-sizing) 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:

(An interactive demo goes here on the page: https://tecyfy.com/data-engineering/course/streaming/on-databricks)

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](/data-engineering/course/streaming/build-it) puts the module together: Kafka to bronze, silver and an hourly gold table, replayed from scratch.

## Check yourself

1. A notebook on Free Edition starts hourly_sales.py's stream with no trigger, as in many examples online. What happens?

   A. It runs every few milliseconds until someone stops it
   B. It runs once an hour
   C. It runs as availableNow
   D. It fails with INFINITE_STREAMING_TRIGGER_NOT_SUPPORTED, because a stream with no trigger runs without stopping, which serverless compute does not allow

2. The shop's dashboard is read each morning, and the fraud check must stop a payment within a second. Which fits each?

   A. The dashboard runs availableNow on a schedule; the fraud check needs a stream that never stops, in real-time mode, on dedicated compute
   B. Both should be real-time streams
   C. Both should run nightly
   D. The dashboard needs real-time mode; the fraud check runs availableNow every hour

### Answers

1. D. With no trigger, Spark uses processingTime with an interval of 0, a stream that never stops. Serverless compute supports only availableNow and Once, and raises that error for any other trigger. On classic compute the same code would run, and check for new data every few milliseconds, which Databricks warns can mean many cloud storage calls and their charges.
2. A. A dashboard read once a day loses nothing by loading new events on a schedule and stopping, and costs only the minutes it runs. A decision within a second cannot wait for a scheduled run; Databricks names fraud detection as a use for real-time mode, which needs dedicated compute running all the time.

## Questions

### Which Structured Streaming triggers work on Databricks serverless compute?

Only availableNow, which Databricks recommends, and Once, which is deprecated. processingTime and continuous are not supported, and a stream without a trigger fails with INFINITE_STREAMING_TRIGGER_NOT_SUPPORTED, because Spark's default is processingTime with an interval of 0. For a stream that never stops on serverless, Databricks points to a Lakeflow pipeline in continuous mode.

### How do I read Kafka in Databricks SQL?

With the read_kafka table-valued function, from Databricks Runtime 13.3 LTS. A streaming query with it, such as CREATE OR REFRESH STREAMING TABLE ... AS SELECT ... FROM STREAM read_kafka(bootstrapServers => '...', subscribe => '...'), works only in a Lakeflow pipeline or as a streaming table in Databricks SQL. The key and value come as bytes; value::string:field reads a field of a JSON value.

## Sources

- [Connect to Apache Kafka](https://docs.databricks.com/aws/en/connect/streaming/kafka)
- [read_kafka table-valued function](https://docs.databricks.com/aws/en/sql/language-manual/functions/read_kafka)
- [Configure Structured Streaming trigger intervals](https://docs.databricks.com/aws/en/structured-streaming/triggers)
- [Serverless compute limitations](https://docs.databricks.com/aws/en/compute/serverless/limitations)
- [Real-time mode in Structured Streaming](https://docs.databricks.com/aws/en/structured-streaming/real-time)
- [Production considerations for Structured Streaming](https://docs.databricks.com/aws/en/structured-streaming/production)
- [Streaming tables](https://docs.databricks.com/aws/en/ldp/streaming-tables)
- [Databricks Free Edition limitations](https://docs.databricks.com/aws/en/getting-started/free-edition-limitations)