Building the Spark UI that Structured Streaming should have had

When we moved 400+ jobs on our Spark framework from DStreams to Structured Streaming, the hardest part was not the code. It was answering simple questions while a job was running. Is it keeping up? How far behind Kafka is it, in minutes? Which part of the batch is slow? What happened at 3 a.m., before it crashed?

The Spark UI could answer almost none of them.

What the stock UI gets wrong for streaming

• It forgets. The Structured Streaming tab lives as long as the query does. When a query stops, fails or restarts, its per-batch history goes with it. The History Server does not keep per-batch durations, phase timings or watermark movement either, so a post-mortem on a crash is mostly guesswork.
• It doesn’t show lag the way operators think about it. Newer Spark versions do report avgOffsetsBehindLatest and maxOffsetsBehindLatest for Kafka sources, but only buried in the progress JSON, and only as offsets. Nobody on call thinks in offsets. They want to know if the pipeline is two minutes behind or forty.
• Row counts can mislead. In some of our configurations numInputRows did not match what the Kafka offsets said had been read. When the two disagree, the offsets are the truth.

We could have bolted on an external APM agent, but our clusters are air-gapped, and I didn’t want another moving part in every job. The framework library is already on every driver, so the UI went there.

Getting inside the Spark UI

Spark has a documented extension point for this kind of thing: SparkPlugin. You name your class in spark.plugins and Spark calls it when the SparkContext starts, on the driver and on every executor.

The catch is that the pieces you need to add a UI tab, SparkUI.attachTab and the WebUITab / WebUIPage classes, are private[spark]. The standard way around that is to put your code in a package under org.apache.spark, which Scala’s package-private visibility then allows. So the extension lives in org.apache.spark.ssui:

package org.apache.spark.ssui

class StructuredStreamingHistoryPlugin extends SparkPlugin {
  override def driverPlugin(): DriverPlugin = new DriverPlugin {
    override def init(sc: SparkContext, ctx: PluginContext): java.util.Map[String, String] = {
      // A broken UI must never take down a production job.
      try sc.ui.foreach(ui => ui.attachTab(new StructuredStreamingHistoryTab(ui)))
      catch { case NonFatal(e) => /* log and carry on */ }
      java.util.Collections.emptyMap()
    }
  }
  override def executorPlugin(): ExecutorPlugin = null
}

The try is the most important line in the file. Observability code that can crash the job it observes is worse than none.

A small detail that makes it feel native: Spark creates its own SQL and Structured Streaming tabs lazily, after the first query starts. The listener waits for that, optionally hides the built-in streaming tab, and re-attaches the Streaming Stats tab so it sits at the end of the nav bar instead of in the middle of it.

Collecting the data

Data comes from a StreamingQueryListener, registered through spark.sql.streaming.streamingQueryListeners so jobs don’t need any code changes:

class StructuredStreamingHistoryListener extends StreamingQueryListener {
  override def onQueryStarted(e: QueryStartedEvent): Unit      = Store.start(e.id, e.runId, e.name)
  override def onQueryProgress(e: QueryProgressEvent): Unit    = Store.append(e.progress)
  override def onQueryTerminated(e: QueryTerminatedEvent): Unit = Store.terminate(e.runId, e.exception)
}

Every completed micro-batch produces a StreamingQueryProgress: batch ID, input and processing rates, a durationMs map with the time spent in each phase, and, for each source, the start, end and latest offsets as JSON.

These go into an in-memory store keyed by query and run, each with a bounded ring buffer. By default it keeps the last 2,000 batches per query and the last 50 terminated queries. The bounds matter: these jobs run for months, and an unbounded history is just a slow memory leak. When a query restarts, its old run stays visible next to the new one, which is exactly what you want at 3 a.m.

Lag you can act on

The source offsets are where the useful numbers are. For a Kafka source they look like {"topic":{"0":1520,"1":1498}}, so a few lines turn them into ground truth:

val trueInputRows = partitions.map(p => end(p) - start(p)).sum    // what was actually read
val consumerLag   = partitions.map(p => latest(p) - end(p)).sum   // what is still waiting
val lagSeconds    = if (processedRowsPerSecond > 0) consumerLag / processedRowsPerSecond else 0

• True input rows come from offset deltas, not from task metrics, so they are right regardless of how the batch was executed.
• Consumer lag is the number of messages still waiting in Kafka when the batch finished.
• Lag in time divides that by the current processing rate, which turns “4.2 million messages” into “about three minutes behind”. That number goes on the page in traffic-light colours: real-time, under two minutes, two to five, and over five.

A single lag reading is noisy, so the page also shows direction. It compares the average lag of the last few batches against the few before them and labels the query catching up, falling behind, steady, or online when lag is zero. That label answers “should I worry?” faster than any chart.

Where the time goes

durationMs breaks every batch into phases:

• latestOffset: asking Kafka where the end of each partition is
• queryPlanning: Catalyst analysing and optimising the plan
• addBatch: actually running the batch
• walCommit: writing the offset log

Stacked per batch, these tell you what kind of problem you have before you open a single log file. Growing queryPlanning is a plan problem; the deep withColumn plans from the migration showed up exactly like that, planning time climbing to seconds per batch. Spikes in walCommit point at the checkpoint storage. And if addBatch dominates, the job is simply doing too much work per batch.

Charts without JavaScript

The air-gap ruled out charting libraries from a CDN, and I didn’t want to vendor a large one into a library that ships with every job. So the charts are SVG, generated in Scala on the server, the same way Spark’s own pages are built:

• batch duration over time, with min, average, max and latest
• input rate against processing rate, which shows headroom at a glance
• Kafka lag over time
• the phase breakdown as stacked bars

Each chart uses a fixed viewBox and scales to the page width. Hover tooltips are plain SVG <title> elements, so there is no client code at all. Below the charts, a sortable table lists every retained batch with its offsets and phase timings, filterable to the last 10, 30, 100 or 500 batches, and every query can be exported as CSV or JSON for anything that wants to scrape it.

What it changed

The difference was less about any single chart and more about the default. Every job on the framework, all 400+ of them, now ships with the same Streaming Stats tab, with no configuration and no agent. During the migration it was how we compared Structured Streaming against DStreams batch for batch. Since then it has become the first place anyone looks when a pipeline misbehaves.

If you build on Spark and run streaming jobs, the takeaway is that most of this is not hard. SparkPlugin and StreamingQueryListener give you everything you need. The offsets in the progress events are the most underused data in Structured Streaming.