tutorial

How to Monitor Spark Structured Streaming with Vigilmon

"A complete guide to monitoring Spark Structured Streaming micro-batch jobs—covering query progress, trigger intervals, source lag, sink latency, and Vigilmon integration."

How to Monitor Spark Structured Streaming with Vigilmon

Spark Structured Streaming turns Apache Spark's batch processing engine into a continuous micro-batch processing system. Each micro-batch reads a chunk of data from a source (Kafka, files, Delta Lake), processes it through your transformations, and writes the results to a sink. The simplicity is the appeal—but it hides real operational risk. A streaming query can stall, fall behind on source data, write results too slowly, or fail checkpointing silently.

This tutorial covers how to monitor Spark Structured Streaming queries in production: tracking streaming query progress, ensuring trigger intervals stay within SLA, monitoring source offset lag, measuring sink write latency, and tracking checkpoint duration. You will wire all of it into Vigilmon for automated alerting.

Prerequisites

  • Apache Spark 3.2+ with Structured Streaming jobs running
  • PySpark or Scala Spark access
  • Vigilmon account at vigilmon.online
  • Access to the Spark UI (port 4040) or Spark History Server

How Structured Streaming Works

Spark Structured Streaming runs as a series of micro-batches. Each micro-batch is a mini Spark job that:

  1. Reads new data from the source up to the current offset
  2. Applies your transformations
  3. Writes results to the sink
  4. Saves checkpoint data to mark progress

The trigger interval controls how often micro-batches run. With Trigger.ProcessingTime("10 seconds"), a new micro-batch starts every 10 seconds. If a micro-batch takes longer than 10 seconds to complete, Spark queues the next one, causing lag to accumulate.

Step 1: Access Streaming Query Progress Metrics

Every active Spark Structured Streaming query exposes progress metrics through the StreamingQueryListener API and the Spark UI. The most direct way to access these metrics is via the Spark REST API:

# List active streaming queries (replace host:4040 with your Spark UI address)
curl -sf http://spark-driver:4040/api/v1/applications/app-id/streaming/statistics

For more detail, query each streaming query's recent progress:

# Get streaming query progress
curl -sf http://spark-driver:4040/api/v1/applications/app-id/streaming/batches \
  | jq '.[0:5] | .[] | {
      batchId: .batchId,
      status: .status,
      processingTime: .processingTime,
      triggerExecution: .processingTime
    }'

The processingTime field tells you how long each micro-batch took. Compare it against your trigger interval—if processing time consistently exceeds the trigger interval, the query is falling behind.

Step 2: Track Streaming Query Progress via Listener

The most reliable way to capture Structured Streaming metrics is to register a StreamingQueryListener in your Spark application. This runs in-process and lets you export metrics to any backend:

from pyspark.sql.streaming import StreamingQueryListener
from pyspark.sql import SparkSession
import json
import time

class MetricsExporter(StreamingQueryListener):
    def __init__(self, metrics_file="/tmp/spark_streaming_metrics.json"):
        self.metrics_file = metrics_file

    def onQueryStarted(self, event):
        print(f"Query started: {event.name} ({event.id})")

    def onQueryProgress(self, event):
        progress = event.progress
        metrics = {
            "query_name": progress.name,
            "query_id": str(progress.id),
            "timestamp": progress.timestamp,
            "batch_id": progress.batchId,
            "num_input_rows": progress.numInputRows,
            "input_rows_per_second": progress.inputRowsPerSecond,
            "processed_rows_per_second": progress.processedRowsPerSecond,
            "trigger_execution_ms": progress.triggerExecution,
            "sources": [
                {
                    "description": src.description,
                    "start_offset": str(src.startOffset),
                    "end_offset": str(src.endOffset),
                    "num_input_rows": src.numInputRows,
                    "input_rows_per_second": src.inputRowsPerSecond,
                }
                for src in progress.sources
            ],
            "sink": {
                "description": progress.sink.description,
            },
            "duration_ms": progress.durationMs,
        }

        with open(self.metrics_file, 'w') as f:
            json.dump(metrics, f)

    def onQueryTerminated(self, event):
        if event.exception:
            print(f"Query {event.id} FAILED: {event.exception}")
        else:
            print(f"Query {event.id} stopped cleanly")


# Register the listener when creating the SparkSession
spark = SparkSession.builder \
    .appName("MyStreamingApp") \
    .getOrCreate()

spark.streams.addListener(MetricsExporter())

This writes the latest progress to a JSON file after each micro-batch. Your monitoring scripts can read this file directly.

Step 3: Monitor Trigger Interval Compliance

A trigger interval violation means Spark cannot keep up with the incoming data rate. Build a check that reads the exported metrics and compares batch duration to the trigger interval:

#!/usr/bin/env python3
import json
import sys
import time

METRICS_FILE = "/tmp/spark_streaming_metrics.json"
TRIGGER_INTERVAL_MS = 10_000  # 10 seconds
MAX_BATCH_DURATION_RATIO = 1.5  # alert if batch takes 1.5× the trigger interval
MAX_METRICS_AGE_SECONDS = 120  # alert if no metrics in 2 minutes

try:
    with open(METRICS_FILE) as f:
        metrics = json.load(f)
except FileNotFoundError:
    print(f"CRITICAL: Metrics file {METRICS_FILE} not found — query may not be running")
    sys.exit(2)

# Check metrics freshness
metrics_time = time.mktime(time.strptime(
    metrics["timestamp"][:19], "%Y-%m-%dT%H:%M:%S"
))
age_seconds = time.time() - metrics_time

if age_seconds > MAX_METRICS_AGE_SECONDS:
    print(f"CRITICAL: Last metrics update was {age_seconds:.0f}s ago — query stalled or stopped")
    sys.exit(2)

# Check trigger execution time
trigger_ms = metrics.get("trigger_execution_ms", 0)
ratio = trigger_ms / TRIGGER_INTERVAL_MS

if ratio > MAX_BATCH_DURATION_RATIO:
    print(f"WARNING: Micro-batch took {trigger_ms}ms ({ratio:.1f}× trigger interval {TRIGGER_INTERVAL_MS}ms)")
    sys.exit(1)

print(f"OK: Micro-batch completed in {trigger_ms}ms ({ratio:.2f}× trigger interval)")
print(f"    Input rows: {metrics.get('num_input_rows', 0)}")
print(f"    Throughput: {metrics.get('processed_rows_per_second', 0):.0f} rows/sec")
sys.exit(0)

Step 4: Monitor Source Offset Lag

Source offset lag tells you how far behind Spark's consumption is from the latest data in the source. For Kafka sources, this is the difference between the latest Kafka offset and Spark's current read offset.

Extract source lag from the streaming progress metrics:

#!/usr/bin/env python3
import json
import sys
from kafka import KafkaConsumer, KafkaAdminClient
from kafka.structs import TopicPartition

METRICS_FILE = "/tmp/spark_streaming_metrics.json"
KAFKA_BOOTSTRAP = "kafka:9092"
MAX_LAG_PER_PARTITION = 10_000

with open(METRICS_FILE) as f:
    metrics = json.load(f)

for source in metrics.get("sources", []):
    if "kafka" not in source["description"].lower():
        continue

    # Parse the end offset from Spark's progress metrics
    # Format: {"topic": {"partition": offset, ...}}
    try:
        end_offsets = json.loads(source["end_offset"].replace("'", '"'))
    except (json.JSONDecodeError, TypeError):
        print(f"WARNING: Could not parse end offset for {source['description']}")
        continue

    # Get latest offsets from Kafka
    consumer = KafkaConsumer(bootstrap_servers=KAFKA_BOOTSTRAP)
    total_lag = 0
    for topic, partitions in end_offsets.items():
        for partition_str, spark_offset in partitions.items():
            tp = TopicPartition(topic, int(partition_str))
            end = consumer.end_offsets([tp])
            kafka_latest = end.get(tp, 0)
            lag = kafka_latest - int(spark_offset)
            total_lag += lag
            if lag > MAX_LAG_PER_PARTITION:
                print(f"WARNING: {topic}[{partition_str}] lag={lag} (max {MAX_LAG_PER_PARTITION})")

    consumer.close()
    print(f"Total Kafka lag: {total_lag} messages")
    sys.exit(1 if total_lag > MAX_LAG_PER_PARTITION * 5 else 0)

Step 5: Measure Sink Write Latency

Sink write latency is captured in the durationMs section of the streaming progress object. This is a map of phase names to milliseconds spent in that phase:

{
  "durationMs": {
    "latestOffset": 12,
    "getBatch": 8,
    "queryPlanning": 45,
    "walCommit": 23,
    "commitOffsets": 15,
    "triggerExecution": 890,
    "addBatch": 750
  }
}

The addBatch duration represents actual data write time to the sink. Build a latency check:

#!/usr/bin/env python3
import json
import sys

METRICS_FILE = "/tmp/spark_streaming_metrics.json"
MAX_ADD_BATCH_MS = 5000  # 5 seconds for sink writes

with open(METRICS_FILE) as f:
    metrics = json.load(f)

duration_map = metrics.get("duration_ms", {})
add_batch_ms = duration_map.get("addBatch", 0)
total_ms = duration_map.get("triggerExecution", 0)

if add_batch_ms > MAX_ADD_BATCH_MS:
    print(f"WARNING: Sink write (addBatch) took {add_batch_ms}ms — exceeds {MAX_ADD_BATCH_MS}ms threshold")
    print(f"  Total micro-batch time: {total_ms}ms")
    sys.exit(1)

print(f"OK: Sink write latency {add_batch_ms}ms (total batch: {total_ms}ms)")
sys.exit(0)

High addBatch time suggests the sink is slow to accept writes. For Kafka sinks, check broker throughput. For file sinks (Delta, Parquet), check HDFS or S3 write throughput.

Step 6: Monitor Checkpoint Duration

Checkpointing in Structured Streaming writes offset and state information to durable storage after each micro-batch. Slow checkpoints mean slow micro-batches.

The walCommit and commitOffsets phases in durationMs capture checkpoint write time. Monitor them together:

#!/usr/bin/env python3
import json
import sys

METRICS_FILE = "/tmp/spark_streaming_metrics.json"
MAX_CHECKPOINT_MS = 3000  # 3 seconds

with open(METRICS_FILE) as f:
    metrics = json.load(f)

duration_map = metrics.get("duration_ms", {})
wal_ms = duration_map.get("walCommit", 0)
commit_ms = duration_map.get("commitOffsets", 0)
checkpoint_total = wal_ms + commit_ms

if checkpoint_total > MAX_CHECKPOINT_MS:
    print(f"WARNING: Checkpoint took {checkpoint_total}ms (walCommit={wal_ms}ms, commitOffsets={commit_ms}ms)")
    print("  Check checkpoint storage (HDFS/S3) performance")
    sys.exit(1)

print(f"OK: Checkpoint completed in {checkpoint_total}ms")
sys.exit(0)

Step 7: Build a Health API for Vigilmon

Consolidate all checks into a single HTTP endpoint that Vigilmon polls:

#!/usr/bin/env python3
from flask import Flask, jsonify
import json
import time
import os

app = Flask(__name__)
METRICS_FILE = "/tmp/spark_streaming_metrics.json"
TRIGGER_INTERVAL_MS = 10_000
MAX_METRICS_AGE_SECONDS = 120

@app.route('/health')
def health():
    if not os.path.exists(METRICS_FILE):
        return jsonify({'status': 'error', 'reason': 'metrics file missing'}), 503

    with open(METRICS_FILE) as f:
        metrics = json.load(f)

    # Freshness check
    metrics_time = time.mktime(time.strptime(
        metrics["timestamp"][:19], "%Y-%m-%dT%H:%M:%S"
    ))
    age = time.time() - metrics_time
    if age > MAX_METRICS_AGE_SECONDS:
        return jsonify({'status': 'error', 'reason': f'stale metrics ({age:.0f}s old)'}), 503

    # Trigger compliance
    trigger_ms = metrics.get("trigger_execution_ms", 0)
    duration_map = metrics.get("duration_ms", {})
    add_batch_ms = duration_map.get("addBatch", 0)
    checkpoint_ms = duration_map.get("walCommit", 0) + duration_map.get("commitOffsets", 0)

    warnings = []
    if trigger_ms > TRIGGER_INTERVAL_MS * 1.5:
        warnings.append(f"trigger_violation:{trigger_ms}ms")
    if add_batch_ms > 5000:
        warnings.append(f"slow_sink:{add_batch_ms}ms")
    if checkpoint_ms > 3000:
        warnings.append(f"slow_checkpoint:{checkpoint_ms}ms")

    ok = len(warnings) == 0
    return jsonify({
        'status': 'ok' if ok else 'degraded',
        'query_name': metrics.get('query_name'),
        'batch_id': metrics.get('batch_id'),
        'metrics_age_seconds': round(age, 1),
        'trigger_execution_ms': trigger_ms,
        'sink_write_ms': add_batch_ms,
        'checkpoint_ms': checkpoint_ms,
        'input_rows': metrics.get('num_input_rows', 0),
        'warnings': warnings,
    }), 200 if ok else 503

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

Step 8: Configure Vigilmon Monitors

Log in to vigilmon.online and create the following monitors.

Streaming Query Health Monitor

| Field | Value | |-------|-------| | Name | Spark Streaming Health | | Type | HTTP | | URL | http://your-metrics-api:8080/health | | Expected Status | 200 | | Response Contains | "status": "ok" | | Interval | 60 seconds | | Timeout | 15 seconds |

Spark UI Availability

| Field | Value | |-------|-------| | Name | Spark Driver UI | | Type | HTTP | | URL | http://spark-driver:4040 | | Expected Status | 200 | | Interval | 60 seconds | | Timeout | 10 seconds |

Alert Routing

Under Settings → Notifications:

  • Query stalled (stale metrics): PagerDuty immediately
  • Trigger violation: Slack #streaming-ops after 2 consecutive failures
  • Slow sink: Slack #data-platform + email after 5 minutes
  • Slow checkpoint: Slack #streaming-ops after 3 consecutive failures

Monitoring at a Glance

| Metric | Source | Threshold | |--------|--------|-----------| | Micro-batch duration | trigger_execution_ms | ≤ 1.5× trigger interval | | Query freshness | Metrics file age | < 2 minutes | | Source Kafka lag | Kafka end offsets | < 10k/partition | | Sink write time | addBatch duration | < 5 seconds | | Checkpoint time | walCommit + commitOffsets | < 3 seconds | | Input throughput | inputRowsPerSecond | > 0 when data is available | | Spark UI | HTTP check | 200 OK |

Troubleshooting Common Issues

Micro-batch duration exceeds trigger interval:

  • Profile the slow phase using Spark UI's streaming tab
  • Check getBatch time—slow Kafka fetches indicate broker issues
  • Check addBatch time—slow sinks or wide transformations
  • Increase trigger interval temporarily to reduce queue pressure while optimizing

Query stalls (no progress):

  • Check Spark driver logs for unhandled exceptions
  • Look for deadlocks in task scheduling (check active executors in Spark UI)
  • Review checkpoint storage for permission or quota issues

High Kafka source lag:

  • Increase Spark executor count to parallelize reading more partitions
  • Check Kafka partition count—add partitions if source throughput exceeds Spark's ingest rate
  • Check for producer-side bursts that temporarily exceed consumer capacity

Slow checkpointing:

  • Move checkpoint location to faster storage (local NVMe vs. S3)
  • Reduce state retention time with withWatermark and state time-to-live settings
  • Check for checkpoint storage throughput limits or throttling

Conclusion

Spark Structured Streaming delivers the familiarity of Spark's DataFrame API on streaming data, but reliable production operation means tracking micro-batch health at each layer. By exporting progress metrics from the StreamingQueryListener API and feeding them into a health endpoint, you get granular visibility into trigger compliance, source lag, sink latency, and checkpoint performance.

Wire these checks into Vigilmon and you have automated monitoring that catches problems in minutes rather than hours. Start with the query freshness check—it catches the most common failure mode (the query silently stopped)—then add trigger compliance and sink latency checks as your streaming workload grows.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles