tutorial

How to Monitor Apache Spark: Driver Availability, Job SLOs, and Pipeline Uptime with Vigilmon

Spark driver failures, executor churn, and runaway shuffle jobs kill batch pipelines silently. Learn how to monitor Spark driver availability, executor health, job duration SLOs, and integrate Vigilmon heartbeats for end-to-end pipeline uptime.

Apache Spark jobs look simple until they don't finish. A driver that restarted under OOM, executors churning through retries, a shuffle service that became unavailable — all of these can let spark-submit hang for hours before anyone notices. Standard Spark monitoring (Spark UI, History Server) requires someone to look at it; you need proactive, external alerting.

Vigilmon gives Spark pipelines external visibility: HTTP monitors for the Spark REST API and shuffle service, heartbeat monitors so you know immediately when a job fails to deliver output on schedule, and response time alerting to catch driver GC pressure before it cascades.

Spark Failure Modes Worth Monitoring

  • Driver OOM or crash — job fails with no retry unless you're using cluster mode with restartPolicy
  • Executor failure rate — high executor churn causes task retries, inflating job duration past SLO without a hard failure
  • Job/stage duration SLO breach — the job runs but takes 3× longer than expected, meaning downstream is late
  • Shuffle service unavailability — in yarn/external shuffle service setups, an unavailable shuffle service causes all pending stages to hang
  • Dynamic allocation starvation — job waits for executor allocation that never arrives due to cluster capacity limits

Step 1: Build a Spark Health Endpoint

Spark exposes a REST API on the driver (port 4040 for running apps, or History Server port 18080 for completed apps).

Probe a Running Driver

# Running app metrics via Spark UI REST API (driver port 4040)
GET http://spark-driver:4040/api/v1/applications/{appId}/stages
GET http://spark-driver:4040/api/v1/applications/{appId}/executors

Node.js Health Sidecar

Build a health sidecar that wraps the Spark REST API and exposes a clean 200/503 endpoint for Vigilmon:

// spark-health.js
const express = require('express');
const axios = require('axios');

const app = express();
const SPARK_UI = process.env.SPARK_UI_URL || 'http://localhost:4040';
const APP_ID = process.env.SPARK_APP_ID;  // or read from first /applications response
const MAX_EXECUTOR_FAILURE_RATE = parseFloat(process.env.MAX_EXECUTOR_FAILURE_RATE || '0.2');
const MAX_FAILED_TASKS = parseInt(process.env.MAX_FAILED_TASKS || '100');

async function getAppId() {
  const resp = await axios.get(`${SPARK_UI}/api/v1/applications`, { timeout: 5000 });
  return resp.data[0]?.id;
}

app.get('/health/spark', async (req, res) => {
  try {
    const appId = APP_ID || await getAppId();

    const [execResp, stageResp] = await Promise.all([
      axios.get(`${SPARK_UI}/api/v1/applications/${appId}/executors`, { timeout: 5000 }),
      axios.get(`${SPARK_UI}/api/v1/applications/${appId}/stages`, { timeout: 5000 }),
    ]);

    const executors = execResp.data.filter(e => e.id !== 'driver');
    const activeExecutors = executors.filter(e => e.activeTasks >= 0 && !e.isBlacklisted);
    const failedTasks = executors.reduce((sum, e) => sum + e.failedTasks, 0);

    const issues = [];

    if (executors.length === 0) {
      issues.push('no executors allocated');
    }

    const failureRate = executors.length > 0
      ? (executors.filter(e => e.isBlacklisted).length / executors.length)
      : 0;

    if (failureRate > MAX_EXECUTOR_FAILURE_RATE) {
      issues.push(`executor failure rate ${(failureRate * 100).toFixed(0)}%`);
    }

    if (failedTasks > MAX_FAILED_TASKS) {
      issues.push(`${failedTasks} failed tasks`);
    }

    if (issues.length > 0) {
      return res.status(503).json({ status: 'degraded', issues, appId });
    }

    return res.status(200).json({
      status: 'ok',
      appId,
      executors: activeExecutors.length,
      failed_tasks: failedTasks,
    });
  } catch (err) {
    // If we can't reach the driver, it's down
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.get('/health/spark/shuffle', async (req, res) => {
  // External shuffle service exposes a metrics servlet on port 7337
  const SHUFFLE_URL = process.env.SHUFFLE_SERVICE_URL || 'http://localhost:7337';
  try {
    const resp = await axios.get(`${SHUFFLE_URL}/metrics/json`, { timeout: 3000 });
    return res.status(200).json({ status: 'ok', shuffle_metrics: Object.keys(resp.data).length });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.listen(3011, () => console.log('Spark health sidecar on :3011'));

Python Health Probe (for Spark-on-Kubernetes)

# spark_health.py
import os, requests
from flask import Flask, jsonify

app = Flask(__name__)
SPARK_UI = os.environ.get('SPARK_UI_URL', 'http://localhost:4040')

@app.route('/health/spark')
def spark_health():
    try:
        apps = requests.get(f'{SPARK_UI}/api/v1/applications', timeout=5).json()
        if not apps:
            return jsonify(status='down', reason='no_active_applications'), 503
        app_id = apps[0]['id']

        executors = requests.get(
            f'{SPARK_UI}/api/v1/applications/{app_id}/executors', timeout=5
        ).json()

        workers = [e for e in executors if e['id'] != 'driver']
        failed = sum(e['failedTasks'] for e in workers)
        blacklisted = sum(1 for e in workers if e.get('isBlacklisted', False))

        if not workers:
            return jsonify(status='degraded', reason='no_executors'), 503
        if blacklisted / max(len(workers), 1) > 0.2:
            return jsonify(status='degraded', reason='high_executor_failure', blacklisted=blacklisted), 503

        return jsonify(status='ok', executors=len(workers), failed_tasks=failed)
    except Exception as e:
        return jsonify(status='down', error=str(e)), 503

if __name__ == '__main__':
    app.run(port=3011)

Step 2: Job Duration SLO Monitoring via Heartbeats

HTTP monitors tell you whether Spark is running. Heartbeat monitors tell you whether it's delivering output on time — which is what your downstream SLAs actually care about.

Configure a Heartbeat in Vigilmon

  1. Monitors → New Monitor → Heartbeat
  2. Name: spark-etl-nightly-batch
  3. Expected interval: match your job's scheduled completion time (e.g., 60 minutes for a nightly job that should finish within an hour of start)
  4. Grace period: 15 minutes
  5. Alert channel: PagerDuty

Wire the Heartbeat Into Your Spark Job

# At the end of your PySpark job, ping Vigilmon on successful completion
import requests, os

def ping_vigilmon_heartbeat():
    url = os.environ.get('VIGILMON_HEARTBEAT_URL')
    if url:
        try:
            requests.get(url, timeout=5)
        except Exception:
            pass  # don't let monitoring failure kill the job

# In your Spark driver:
if __name__ == '__main__':
    spark = SparkSession.builder.appName('nightly-etl').getOrCreate()
    try:
        run_etl(spark)
        ping_vigilmon_heartbeat()
    finally:
        spark.stop()

For Scala/Java Spark jobs:

// Add to your driver main()
import java.net.{HttpURLConnection, URL}

def pingVigilmon(url: String): Unit = {
  try {
    val conn = new URL(url).openConnection().asInstanceOf[HttpURLConnection]
    conn.setRequestMethod("GET")
    conn.setConnectTimeout(3000)
    conn.setReadTimeout(3000)
    conn.getResponseCode
    conn.disconnect()
  } catch {
    case _: Exception => // monitoring failure must never kill the job
  }
}

// At end of successful job execution:
pingVigilmon(sys.env.getOrElse("VIGILMON_HEARTBEAT_URL", ""))

Stage-Level Heartbeats for Long-Running Jobs

For jobs with distinct stages (ingest → transform → load), send a heartbeat at each stage boundary. This lets you identify which stage is hanging:

def with_stage_heartbeat(stage_name: str, fn):
    result = fn()
    url = os.environ.get(f'VIGILMON_HEARTBEAT_{stage_name.upper()}')
    if url:
        requests.get(url, timeout=5)
    return result

with_stage_heartbeat('ingest', lambda: run_ingest(spark))
with_stage_heartbeat('transform', lambda: run_transform(spark))
with_stage_heartbeat('load', lambda: run_load(spark))

Step 3: Configure Vigilmon HTTP Monitors

  1. Monitors → New Monitor → HTTP / HTTPS
  2. URL: https://your-spark-host.example.com/health/spark
  3. Interval: 2 minutes
  4. Expected: 200, body contains "status":"ok"
  5. Response time threshold: 8000ms
  6. Alert channel: PagerDuty (P1 for active job failures)

Add a second monitor for shuffle service:

  • URL: https://your-spark-host.example.com/health/spark/shuffle
  • Interval: 2 minutes
  • Alert channel: Slack (shuffle outages degrade jobs but don't fail them immediately)

Step 4: Alert Routing for Spark Pipelines

| Monitor | Trigger | Channel | Priority | |---|---|---|---| | HTTP: /health/spark | Driver down or executor failure >20% | PagerDuty | P1 | | HTTP: /health/spark/shuffle | Shuffle service unavailable | Slack | P2 | | Heartbeat: job completion | Job didn't finish within SLO | PagerDuty | P1 | | Heartbeat: ingest stage | Stage hung for >30 min | Slack | P2 | | Heartbeat: load stage | Output not delivered | PagerDuty | P1 |

Summary

Spark jobs fail in three ways that are easy to miss: the driver crashes silently, executor churn inflates job duration past SLO, or the job runs successfully but produces no output due to a logic error. Cover all three:

| Monitor Type | What It Catches | |---|---| | HTTP: driver health | Driver down, executor failures, blacklisted workers | | HTTP: shuffle service | Shuffle service outages causing stage hangs | | Heartbeat: job completion | SLO breach, job crash before output | | Heartbeat: per-stage | Identifies which stage in a multi-stage job is hanging |

Set up Spark monitoring at vigilmon.online in under five minutes.

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