Tutorial: Monitor a Structured Streaming query with StreamingQueryListener
A StreamingQueryListener receives progress metrics from a running Structured Streaming query at the end of every micro-batch, such as the number of input rows, the processing rate, and the batch duration. This tutorial guides you through attaching a listener that checks whether the query keeps up with its input, warns when the query falls behind, and records each batch's health to a Unity Catalog table that you query with SQL. The example runs on serverless compute and on Unity Catalog-enabled compute.
For a complete reference of the metrics and event fields that a listener reports, see Monitoring Structured Streaming queries on Databricks.
This tutorial includes the following steps:
- Create a notebook
- Create the Unity Catalog objects
- Define the listener
- Define the periodic writer
- Start the query with monitoring
- Query the recorded health
- Stop the query and clean up
Requirements
To complete this tutorial, you must meet the following requirements:
- Your workspace must have Unity Catalog enabled. See What is Unity Catalog?.
- You must have access to serverless compute or a Unity Catalog-enabled cluster. For the Databricks Runtime versions and access modes that
StreamingQueryListenersupports, see Monitoring Structured Streaming queries on Databricks. - You must have permission to create a schema, a volume, and tables in a Unity Catalog catalog. This tutorial uses a catalog named
main. Substitute a catalog that you can write to.
Step 1: Create a notebook
You use a Databricks notebook to define the listener and run the streaming query.
To create a notebook:
- Click New in the sidebar, then click
Notebook. - In the compute drop-down menu, select serverless compute or your Unity Catalog-enabled cluster.
- Select Python as the default language.
Step 2: Create the Unity Catalog objects
The listener records per-batch health to a Unity Catalog managed table, the streaming query writes to a second managed table, and the query stores its checkpoint in a Unity Catalog volume. To create the schema and volume and define the object names, run:
catalog = "main"
schema = "streaming_monitoring"
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.{schema}")
spark.sql(f"CREATE VOLUME IF NOT EXISTS {catalog}.{schema}.checkpoints")
health_table = f"{catalog}.{schema}.streaming_health"
target_table = f"{catalog}.{schema}.rate_target"
checkpoint_path = f"/Volumes/{catalog}/{schema}/checkpoints/rate_target"
Step 3: Define the listener
The listener keeps each callback lightweight. For every micro-batch, it compares the processing rate to the input rate, prints a warning when the query falls behind, and appends the batch's health to an in-memory buffer. To define the listener, run:
import threading
from pyspark.sql.streaming import StreamingQueryListener
class HealthListener(StreamingQueryListener):
def __init__(self):
self._lock = threading.Lock()
self._buffer = []
def onQueryStarted(self, event):
print(f"Query started: {event.id}")
def onQueryProgress(self, event):
progress = event.progress
input_rate = progress.inputRowsPerSecond
processed_rate = progress.processedRowsPerSecond
# The query is falling behind when it processes data slower than it arrives.
falling_behind = input_rate > 0 and processed_rate < input_rate
if falling_behind:
print(f"WARNING batch {progress.batchId}: falling behind "
f"({processed_rate:.1f} processed/s < {input_rate:.1f} input/s)")
with self._lock:
self._buffer.append((int(progress.batchId), int(progress.numInputRows),
float(input_rate), float(processed_rate),
bool(falling_behind)))
def drain(self):
with self._lock:
records, self._buffer = self._buffer, []
return records
def onQueryIdle(self, event):
pass
def onQueryTerminated(self, event):
print(f"Query terminated: {event.id}")
The listener overrides the following event methods:
Method | Description |
|---|---|
| Runs when the query starts. |
| Runs at the end of each micro-batch. This example compares the processing rate to the input rate, warns when the query falls behind, and buffers the batch's health. |
| Runs when the query is idle and waiting for new data. |
| Runs when the query stops, with or without an error. |
The drain method returns the buffered records and clears the buffer. A lock guards the buffer because the listener appends to it from the listener thread while the writer in the next step reads from a different thread.
onQueryProgress runs on the listener thread, so heavy work there delays progress reporting for every query in the session. This is why the listener only buffers in the callback and writes from a separate thread. See the note in Monitoring Structured Streaming queries on Databricks.
Step 4: Define the periodic writer
A background thread writes the buffered records to the health table at a fixed interval, so the listener never runs a Spark job on its own thread. The publish function is the single place that writes the records. To send the metrics to another system, such as Apache Kafka, replace the body of publish. To define the writer and its interval loop, run:
def publish(records):
if not records:
return
(spark.createDataFrame(records,
"batchId long, numInputRows long, inputRowsPerSecond double, "
"processedRowsPerSecond double, fallingBehind boolean")
.write.mode("append").saveAsTable(health_table))
stop_event = threading.Event()
def flush_loop(listener, interval_seconds=10):
while not stop_event.wait(interval_seconds):
publish(listener.drain())
The loop writes a batch of records every 10 seconds until stop_event is set. Each write appends many rows at a time, which avoids the small files that a per-batch write would create.
Step 5: Start the query with monitoring
Register the listener, start the writer thread, then start a query that reads from the built-in rate source and writes to the target table. The rate source generates rows at a fixed rate, so the listener reports steady progress without an external data source. To start monitoring and the query, run:
listener = HealthListener()
spark.streams.addListener(listener)
flusher = threading.Thread(target=flush_loop, args=(listener,), daemon=True)
flusher.start()
query = (
spark.readStream.format("rate").option("rowsPerSecond", 5).load()
.writeStream.format("delta")
.option("checkpointLocation", checkpoint_path)
.toTable(target_table)
)
Let the query run for a minute or two so that several flush intervals pass. Each micro-batch triggers onQueryProgress, which prints a warning when the query falls behind and buffers the batch's health. The writer appends the buffered records to the table every 10 seconds, so you can query them while the stream runs.
Step 6: Query the recorded health
Read the health table to see, for each batch, how fast data arrived, how fast the query processed it, and whether the query fell behind. To display the records, run:
display(spark.sql(f"""
SELECT batchId, numInputRows,
round(inputRowsPerSecond, 1) AS input_per_second,
round(processedRowsPerSecond, 1) AS processed_per_second,
fallingBehind
FROM {health_table}
ORDER BY batchId
"""))
The preceding query returns one row per completed batch, such as this:
batchId numInputRows input_per_second processed_per_second fallingBehind
0 2 0.0 1.9 false
1 5 4.8 5.7 false
2 5 5.7 4.6 true
3 5 4.6 6.0 false
4 4 4.8 5.0 false
5 4 5.0 4.4 true
6 5 5.5 6.1 false
7 4 4.8 4.1 true
A healthy query processes data at least as fast as it arrives, so processed_per_second stays at or above input_per_second and fallingBehind is false. Individual batches often dip below the input rate, as batches 2, 5, and 7 show. This jitter is normal. If fallingBehind stays true across several consecutive batches, the query can't keep up with its input. Investigate by adding compute, reducing the work in each batch, or lowering the input rate.
For the full set of source, sink, and state metrics available on each progress event, see Monitoring Structured Streaming queries on Databricks.
Step 7: Stop the query and clean up
Stop the query, stop the writer thread, write any records still in the buffer, and drop the objects you created so that they don't continue to incur costs. To clean up, run:
query.stop()
stop_event.set()
flusher.join()
publish(listener.drain())
spark.streams.removeListener(listener)
spark.sql(f"DROP SCHEMA IF EXISTS {catalog}.{schema} CASCADE")