How do I monitor streaming lag and throughput?

Summary

• Every Structured Streaming progress report carries inputRowsPerSecond (the aggregate rate of data arriving) and processedRowsPerSecond (the aggregate rate Spark is processing it). If input stays above processing, the query is falling behind.
• Read progress with query.lastProgress for a quick look, with a StreamingQueryListener to push every update to your own system, or in the Structured Streaming tab of the Spark web UI (micro-batch mode).
• Lag has more than one meaning. Kafka sources report minOffsetsBehindLatest, maxOffsetsBehindLatest and avgOffsetsBehindLatest in each source's metrics, which count offsets. Event-time lag shows up as eventTime.watermark and the web UI's Global Watermark Gap.
• To feed Prometheus, JMX or Graphite, set spark.sql.streaming.metricsEnabled to true (the default is false) and configure a metrics sink. The REST streaming/* endpoints belong to the legacy DStream API, not to Structured Streaming.

How do I monitor streaming lag and throughput?

Every micro-batch of a Structured Streaming query produces a progress report with its input rate, processing rate, timings and, for Kafka sources, how many offsets the query trails. Read that report from query.lastProgress, push each update to your own system with a StreamingQueryListener, or watch the same numbers in the Spark web UI. "Lag" is not one number. It can mean unread source data, slow batches, or an event-time watermark that trails the clock, and each has its own signal.

Which signals measure streaming lag and throughput?

Signal	Where to read it	What it tells you
Input rate	inputRowsPerSecond in the progress report, "Input Rate" in the web UI	The aggregate rate of data arriving across all sources
Processing rate	processedRowsPerSecond, "Process Rate" in the web UI	The aggregate rate at which Spark is processing data
Batch time	durationMs, "Batch Duration" in the web UI	Time spent per batch and per operation
Source backlog (Kafka)	minOffsetsBehindLatest, maxOffsetsBehindLatest, avgOffsetsBehindLatest in each source's metrics	How many offsets the query is behind the latest available offset
Event-time lag	eventTime.watermark, "Global Watermark Gap" in the web UI	The gap between the batch timestamp and the global watermark

Compare the first two rows first: when the input rate stays above the processing rate, unprocessed data piles up at the source. Backlog metrics count offsets, not seconds. The watermark is the maximum event time the engine has seen minus the late threshold you pass to withWatermark(), so the gap includes that threshold even when the query keeps up. Judge it against your own threshold, not against zero.

How do I read progress from a running query?

lastProgress returns a dict, or None if there have been no progress updates yet.

from pyspark.sql.functions import window

events = (spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
    .option("subscribe", "topic1")
    .load())

query = (events.withWatermark("timestamp", "10 minutes")
    .groupBy(window("timestamp", "5 minutes")).count()
    .writeStream.queryName("events_per_window")
    .outputMode("update").format("console")
    .option("checkpointLocation", "/tmp/checkpoints/events_per_window")
    .start())

p = query.lastProgress
if p:
    print(p["inputRowsPerSecond"], p["processedRowsPerSecond"])
    print(p.get("eventTime", {}).get("watermark"))
    for s in p["sources"]:
        print(s["description"], s.get("metrics"))

To push every update instead of polling, attach a listener to the SparkSession. It receives onQueryProgress after each batch.

from pyspark.sql.streaming import StreamingQueryListener

class LagListener(StreamingQueryListener):
    def onQueryStarted(self, event): pass
    def onQueryIdle(self, event): pass
    def onQueryTerminated(self, event): pass
    def onQueryProgress(self, event):
        p = event.progress
        print(p.name, p.batchId, p.inputRowsPerSecond, p.processedRowsPerSecond)
        for s in p.sources:
            print(s.description, s.metrics)

spark.streams.addListener(LagListener())

How do I send streaming metrics to Prometheus or another monitoring system?

Structured Streaming metrics reach Dropwizard sinks only after you opt in. Set spark.sql.streaming.metricsEnabled in the SparkSession, and queries started afterward report to whatever sinks you have configured. Spark documents these metrics in the spark.streaming namespace: eventTime-watermark, inputRate-total, latency, processingRate-total, states-rowsTotal and states-usedBytes.

spark.sql.streaming.metricsEnabled=true
spark.metrics.conf.*.sink.prometheusServlet.class=org.apache.spark.metrics.sink.PrometheusServlet
spark.metrics.conf.*.sink.prometheusServlet.path=/metrics/prometheus

The driver then serves Prometheus-format metrics at /metrics/prometheus/ on its web UI port, 4040 by default. PrometheusServlet is marked experimental. Other documented sinks include ConsoleSink, CSVSink, JmxSink, MetricsServlet, GraphiteSink, Slf4jSink and StatsdSink.
The REST path /api/v1/applications/[app-id]/streaming/statistics and its receivers and batches siblings are easy to mistake for a Structured Streaming monitor. The docs describe them as statistics for the streaming context and a list of streaming receivers, and they are defined in Spark's streaming module for DStreams, which its guide calls a legacy project. For Structured Streaming, use the progress report, a listener or the metrics system.

How does Databricks monitor streaming queries?

Databricks provides built-in monitoring for Structured Streaming through the Streaming tab in the Spark UI. Give every stream a unique name with .queryName(<query-name>) so you can tell which metrics belong to which stream. To push metrics to an external service for alerting or dashboards, use the StreamingQueryListener interface, available in Python and Scala on Databricks Runtime 11.3 LTS and above. On compute with dedicated access mode, a listener needs Databricks Runtime 15.1 or above to use credentials or interact with Unity Catalog objects.
Backlog metrics appear in sources.metrics of the progress object. For Kafka, the docs list avgOffsetsBehindLatest, maxOffsetsBehindLatest, minOffsetsBehindLatest and estimatedTotalBytesBehindLatest. For Delta Lake and Auto Loader sources they list numBytesOutstanding and numFilesOutstanding. Keep listener logic light: the docs warn that processing latency in listeners can significantly affect query processing speed.

FAQs

Why is query.lastProgress None?

The PySpark docs say it returns None if there were no progress updates. That is the state right after start() and before the first trigger completes, so check again after a batch has run.

Why does the Structured Streaming tab not show my query?

The Spark web UI documents the tab for Structured Streaming jobs running in micro-batch mode. The web UI listens on port 4040 by default.

Is a large watermark gap a sign the query is behind?

Not by itself. The watermark is the maximum event time seen minus the late threshold, and the gap is measured from the batch timestamp, so your threshold is part of it. Check the input and processing rates and the offsets behind latest before you add resources.