tutorial

How to Monitor Apache Spark GraphX Graph Computation Health and Pipeline Availability with Vigilmon

GraphX graph computation failures silently stall Spark jobs and break graph analytics pipelines. Learn how to monitor Spark GraphX job health, driver availability, and pipeline liveness with Vigilmon HTTP probes and heartbeat monitors.

Apache Spark GraphX is the graph computation API built into Apache Spark — enabling distributed graph analytics, PageRank, connected components, and triangle counting at petabyte scale. But when a GraphX job hangs on a superstep barrier, a Spark driver OOMs processing a large graph RDD, or a graph analytics pipeline stalls mid-iteration, your downstream systems receive no graph updates. Recommendation engines, fraud detection graphs, and knowledge graph pipelines degrade silently while Spark's UI may still show an "active" application that never completes.

Vigilmon gives you external visibility into Spark GraphX job health through HTTP probe monitoring (via Spark's REST API or a custom job health sidecar) and heartbeat monitoring for your graph analytics pipeline applications. This tutorial covers both.

Why Spark GraphX Needs External Monitoring

Spark's built-in tooling (Web UI on port 4040, History Server, Spark REST API on port 4040) provides rich internal diagnostics — but only if someone is watching. External monitoring with Vigilmon adds:

  • Proactive alerting when the Spark driver becomes unreachable (immediately visible to downstream graph consumers as missing or stale graph outputs)
  • Job completion detection by verifying graph computation jobs finish within expected durations rather than looping on retries or hanging on barrier synchronization
  • Heartbeat monitoring so you know immediately when a GraphX pipeline application stops producing graph snapshots or checkpoints
  • Multi-region availability checking from outside your Spark cluster network

These layers work together: the Spark driver can be reachable while a GraphX Pregel iteration is deadlocked, and Spark metrics can look healthy while your graph pipeline has produced no output for hours.

Step 1: Build a GraphX Health Endpoint

Spark exposes a REST API (default port 4040 on the driver) for application, job, and stage status. Wrap it in a health sidecar for precise GraphX pipeline status.

Using the Spark REST API Directly

# List active Spark applications
curl -i http://spark-driver.example.com:4040/api/v1/applications

# Check active jobs for an application
curl -i "http://spark-driver.example.com:4040/api/v1/applications/app-20240101000000-0001/jobs"

# Check stages (useful for detecting GraphX superstep stalls)
curl -i "http://spark-driver.example.com:4040/api/v1/applications/app-20240101000000-0001/stages"

Point a Vigilmon monitor at the applications endpoint for basic driver reachability. For deeper GraphX pipeline status, use a custom sidecar.

Custom Python Health Sidecar

For GraphX-specific health checks — job completion rate, stalled stage detection, and pipeline freshness:

# graphx_health.py
from flask import Flask, jsonify
import requests
import os
import time

app = Flask(__name__)

SPARK_UI_URL = os.environ.get('SPARK_UI_URL', 'http://localhost:4040')
MAX_STALE_SECONDS = int(os.environ.get('MAX_STALE_SECONDS', '3600'))
LAST_COMPLETED_FILE = os.environ.get('LAST_COMPLETED_FILE', '/tmp/graphx_last_completed')

@app.route('/health/graphx')
def graphx_health():
    issues = []

    # Check Spark driver reachability
    try:
        apps_resp = requests.get(f'{SPARK_UI_URL}/api/v1/applications', timeout=5)
        apps_resp.raise_for_status()
        apps = apps_resp.json()
    except requests.RequestException as e:
        return jsonify({
            'status': 'down',
            'reason': 'spark_driver_unreachable',
            'error': str(e),
        }), 503

    if not apps:
        return jsonify({
            'status': 'down',
            'reason': 'no_active_spark_applications',
        }), 503

    app_id = apps[0]['id']
    active_jobs = []
    failed_jobs = []

    try:
        jobs_resp = requests.get(
            f'{SPARK_UI_URL}/api/v1/applications/{app_id}/jobs',
            timeout=5
        )
        jobs_resp.raise_for_status()
        jobs = jobs_resp.json()
        active_jobs = [j for j in jobs if j.get('status') == 'RUNNING']
        failed_jobs = [j for j in jobs if j.get('status') == 'FAILED']
    except requests.RequestException as e:
        issues.append(f'jobs_api_unreachable: {str(e)}')

    # Check for stalled stages (GraphX superstep hangs)
    stalled_stages = []
    try:
        stages_resp = requests.get(
            f'{SPARK_UI_URL}/api/v1/applications/{app_id}/stages',
            timeout=5
        )
        stages_resp.raise_for_status()
        stages = stages_resp.json()
        now_ms = int(time.time() * 1000)
        for stage in stages:
            if stage.get('status') == 'ACTIVE':
                submission_time = stage.get('submissionTime', now_ms)
                if isinstance(submission_time, str):
                    # Parse ISO format if needed
                    pass
                duration_s = (now_ms - submission_time) / 1000 if isinstance(submission_time, int) else 0
                if duration_s > MAX_STALE_SECONDS:
                    stalled_stages.append({
                        'stage_id': stage.get('stageId'),
                        'duration_s': int(duration_s),
                    })
    except requests.RequestException as e:
        issues.append(f'stages_api_unreachable: {str(e)}')

    if stalled_stages:
        issues.append(f'stalled_stages_detected: {stalled_stages}')

    if failed_jobs:
        issues.append(f'failed_jobs: {[j.get("jobId") for j in failed_jobs]}')

    if issues:
        return jsonify({
            'status': 'degraded',
            'issues': issues,
            'active_jobs': len(active_jobs),
            'failed_jobs': len(failed_jobs),
        }), 503

    return jsonify({
        'status': 'ok',
        'app_id': app_id,
        'active_jobs': len(active_jobs),
        'stalled_stages': len(stalled_stages),
    }), 200

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

Scala/JVM Health Sidecar

For teams running GraphX from Scala applications, expose a health endpoint directly from the Spark driver process:

// GraphXHealthServer.scala
import org.http4s._
import org.http4s.dsl.io._
import org.http4s.circe._
import io.circe.syntax._
import cats.effect.IO
import org.apache.spark.SparkContext
import org.apache.spark.graphx._

object GraphXHealthServer {

  def healthRoutes(sc: SparkContext, lastCheckpointTime: () => Long): HttpRoutes[IO] =
    HttpRoutes.of[IO] {
      case GET -> Root / "health" / "graphx" =>
        val staleMs = System.currentTimeMillis() - lastCheckpointTime()
        val staleSeconds = staleMs / 1000

        if (sc.isStopped) {
          val body = Map(
            "status" -> "down",
            "reason" -> "spark_context_stopped"
          ).asJson
          ServiceUnavailable(body)
        } else if (staleSeconds > 3600) {
          val body = Map(
            "status" -> "degraded",
            "reason" -> "graph_checkpoint_stale",
            "stale_seconds" -> staleSeconds.toString
          ).asJson
          ServiceUnavailable(body)
        } else {
          val body = Map(
            "status" -> "ok",
            "spark_app_id" -> sc.applicationId,
            "last_checkpoint_seconds_ago" -> staleSeconds.toString
          ).asJson
          Ok(body)
        }
    }
}

Step 2: Configure Vigilmon HTTP Monitor for GraphX

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your GraphX health endpoint: https://your-app.example.com/health/graphx
  4. Set the check interval to 2 minutes
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 15000ms (graph computation health checks involve RDD metadata queries that can be slow on large graphs)
  6. Under Alert channels, assign your Slack or PagerDuty channel
  7. Save the monitor

Add a second monitor for the Spark REST API directly:

  • URL: http://spark-driver.example.com:4040/api/v1/applications
  • Expected: 200
  • Interval: 5 minutes
  • Alert channel: data-engineering channel

Vigilmon's multi-region probe consensus prevents false positives from transient network blips between probe nodes and your Spark cluster.

Step 3: Heartbeat Monitoring for GraphX Analytics Pipelines

Driver reachability checks are necessary — but not sufficient. A GraphX Pregel algorithm can run indefinitely on a malformed superstep without the driver appearing down. A graph analytics pipeline can stall between computation phases without any visible error. Vigilmon heartbeat monitors detect these silent stalls: your GraphX pipeline pings Vigilmon after each graph computation batch or checkpoint write. If pings stop, Vigilmon fires an alert.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: graphx-analytics-pipeline
  3. Set the expected interval: 60 minutes (adjust to your graph computation batch duration)
  4. Set the grace period: 30 minutes
  5. Save — copy the heartbeat URL, e.g. https://vigilmon.online/heartbeat/abc123xyz

Wire It Into Your GraphX Pipeline (Python / PySpark)

# graphx_pipeline.py
from pyspark.sql import SparkSession
import requests
import os

HEARTBEAT_URL = os.environ['VIGILMON_HEARTBEAT_URL']

def ping_vigilmon():
    try:
        requests.get(HEARTBEAT_URL, timeout=5)
    except Exception:
        pass  # Non-fatal: don't interrupt the graph computation

spark = SparkSession.builder.appName('GraphXPipeline').getOrCreate()
sc = spark.sparkContext

def run_graph_batch(edges_path: str, output_path: str):
    # Load edges RDD
    edges = sc.textFile(edges_path).map(
        lambda line: line.split(',')
    ).map(
        lambda parts: (int(parts[0]), int(parts[1]))
    )

    # Build GraphX graph via GraphFrames (PySpark bridge)
    from graphframes import GraphFrame
    vertices = edges.flatMap(lambda e: [e[0], e[1]]).distinct() \
        .map(lambda v: (v,)).toDF(['id'])
    edges_df = edges.toDF(['src', 'dst'])
    g = GraphFrame(vertices, edges_df)

    # Run PageRank
    results = g.pageRank(resetProbability=0.15, maxIter=10)
    results.vertices.write.mode('overwrite').parquet(output_path)

    # Ping Vigilmon after successful computation
    ping_vigilmon()
    print(f'Graph batch complete, results written to {output_path}')

Wire It Into Your GraphX Pipeline (Scala)

// GraphXPipeline.scala
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
import scala.util.Try
import java.net.URL

object GraphXPipeline {

  val heartbeatUrl = sys.env("VIGILMON_HEARTBEAT_URL")

  def pingVigilmon(): Unit = Try {
    new URL(heartbeatUrl).openStream().close()
  } // Non-fatal: swallow errors so graph computation isn't interrupted

  def runPageRankBatch(graph: Graph[Long, Double]): Unit = {
    val ranks = graph.pageRank(0.0001).vertices
    ranks.saveAsTextFile(s"hdfs:///output/pagerank-${System.currentTimeMillis()}")

    // Ping after successful graph write
    pingVigilmon()
    println("PageRank batch complete")
  }

  def runConnectedComponents(graph: Graph[Long, Double]): Unit = {
    val cc = graph.connectedComponents().vertices
    cc.saveAsTextFile(s"hdfs:///output/cc-${System.currentTimeMillis()}")

    pingVigilmon()
    println("Connected components batch complete")
  }
}

Step 4: Alert Routing for GraphX Failures

GraphX failures cascade: a stalled Pregel iteration blocks all downstream graph consumers; a driver OOM terminates all active graph computations; a checkpoint failure causes graph state loss on the next incremental computation. Alert routing should reflect this cascade.

Configure alert routing in Vigilmon:

| Monitor | Alert Channel | Priority | |---|---|---| | GraphX health /health/graphx | Slack + PagerDuty | P1 | | Spark REST API /api/v1/applications | Slack | P2 | | Heartbeat: graph analytics pipeline | Slack + email | P2 | | Heartbeat: graph export/ETL job | Email | P3 |

Set response time thresholds for early warning:

  • Alert at 10000ms for the health endpoint (slow metadata queries signal memory pressure or GC pauses on the driver)
  • Alert at 60000ms for long-running graph computation batches

For bursty graph workloads, enable two consecutive failures before alerting — Spark drivers can briefly become unresponsive during large graph RDD partition shuffles without representing a full outage.

Summary

GraphX failures are silent at the graph computation layer. External monitoring catches driver crashes, Pregel superstep stalls, and analytics pipeline stalls before they degrade downstream graph consumers or leave recommendation and fraud detection systems running on stale graph data:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/graphx | Driver liveness, stalled stage detection, failed job alerts | | HTTP monitor on Spark REST API | Driver-level availability for direct API callers | | Heartbeat monitor | Graph computation batch progress, checkpoint write liveness |

Get started free at vigilmon.online — your first GraphX monitor is running in under two 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