MCP Tools for Apache Spark — SparkUI REST API, stage health, executor state, shuffle spill, and GC pressure

Spark architecture for MCP tool authors

Spark applications have a Driver (the coordinator process that hosts the SparkContext, schedules stages, and runs the SparkUI) and one or more Executors (JVM processes on worker nodes that execute tasks and cache data). In local mode, driver and executors run in the same JVM. In cluster mode (YARN, Kubernetes, Standalone), the driver runs on a cluster node and executors are launched dynamically.

The Spark History Server (default port 18080) serves the same REST API for completed applications. Use http://driver-host:4040 for live monitoring and http://history-server:18080 for post-mortems. In Kubernetes deployments, expose the driver's port 4040 via a headless service or port-forward — it is not exposed by default. The driver pod label is typically spark-role=driver.

async function createSparkClient(sparkUiUrl) {
  const base = sparkUiUrl.replace(/\/$/, '');

  async function get(path) {
    const res = await fetch(`${base}${path}`, {
      headers: { 'Accept': 'application/json' },
      signal: AbortSignal.timeout(15_000),
    });
    if (!res.ok) throw new Error(`Spark API ${path} → ${res.status}`);
    return res.json();
  }

  async function getApplications() {
    return get('/api/v1/applications');
  }

  async function getJobs(appId, status = null) {
    const qs = status ? `?status=${status}` : '';
    return get(`/api/v1/applications/${appId}/jobs${qs}`);
  }

  async function getStages(appId, status = null) {
    const qs = status ? `?status=${status}` : '';
    return get(`/api/v1/applications/${appId}/stages${qs}`);
  }

  async function getExecutors(appId) {
    return get(`/api/v1/applications/${appId}/executors`);
  }

  async function getStreamingStats(appId) {
    return get(`/api/v1/applications/${appId}/streaming/statistics`);
  }

  async function getStructuredStreamingQueries(appId) {
    return get(`/api/v1/applications/${appId}/streams/active`);
  }

  return { getApplications, getJobs, getStages, getExecutors, getStreamingStats, getStructuredStreamingQueries };
}

Job and stage health: the three-layer status model

Spark jobs contain stages, stages contain tasks. A Spark job status can be RUNNING, SUCCEEDED, FAILED, or UNKNOWN. A running job with all its stages queued (status PENDING) means no executor slots are available — check executor count via GET /api/v1/applications/{appId}/executors.

Stage status is the most diagnostic signal: ACTIVE (tasks running), COMPLETE (all tasks done), FAILED (stage failed after retries), PENDING (waiting for previous stage), or SKIPPED (result cached, stage not needed). A stage in FAILED status with numFailedTasks > 0 tells you task-level failures — check failureReason in stage detail for the exception class and message. The most common stage failures are OOM errors (increase executor memory or reduce partition size), shuffle fetch failures (executor was lost mid-job, increase spark.task.maxFailures), and broadcast join timeout (dataset exceeds spark.sql.broadcastTimeout).

async function assessSparkJobHealth(spark, appId) {
  const [jobs, stages, executors] = await Promise.all([
    spark.getJobs(appId),
    spark.getStages(appId),
    spark.getExecutors(appId),
  ]);

  // Active executors only (exclude dead ones)
  const activeExecutors = executors.filter(e => e.isActive);
  const deadExecutors = executors.filter(e => !e.isActive);

  // Executor health metrics
  const executorHealth = activeExecutors.map(e => {
    const gcFraction = e.totalGCTime > 0 && e.totalDuration > 0
      ? e.totalGCTime / e.totalDuration
      : 0;
    const memFraction = e.maxMemory > 0 ? e.memoryUsed / e.maxMemory : 0;
    return {
      id: e.id,
      host: e.hostPort,
      gcFraction,
      memFraction,
      failedTasks: e.failedTasks,
      rddBlocks: e.rddBlocks,
      diskUsed: e.diskUsed,
      // GC overhead alert: > 10% of executor time in GC
      gcAlert: gcFraction > 0.1,
      // Memory pressure: > 90% used
      memAlert: memFraction > 0.9,
    };
  });

  // Stage health
  const activeStages = stages.filter(s => s.status === 'ACTIVE');
  const failedStages = stages.filter(s => s.status === 'FAILED');
  const pendingStages = stages.filter(s => s.status === 'PENDING');

  // Shuffle spill detection — spill is not an error but indicates memory pressure
  const stagesWithSpill = stages.filter(s => (s.diskBytesSpilled ?? 0) > 0);
  const totalSpillBytes = stagesWithSpill.reduce((sum, s) => sum + (s.diskBytesSpilled ?? 0), 0);

  // Stuck stages: PENDING for too long (no executors), or ACTIVE with 0 tasks running
  const stuckStages = activeStages.filter(s => s.numActiveTasks === 0);

  // Running jobs
  const activeJobs = jobs.filter(j => j.status === 'RUNNING');
  const failedJobs = jobs.filter(j => j.status === 'FAILED');

  return {
    appId,
    jobs: { active: activeJobs.length, failed: failedJobs.length, total: jobs.length },
    stages: {
      active: activeStages.length,
      failed: failedStages.length,
      pending: pendingStages.length,
      stuck: stuckStages.length,
      withSpill: stagesWithSpill.length,
      totalSpillBytes,
    },
    executors: {
      active: activeExecutors.length,
      dead: deadExecutors.length,
      withGcAlert: executorHealth.filter(e => e.gcAlert).length,
      withMemAlert: executorHealth.filter(e => e.memAlert).length,
      health: executorHealth,
    },
    failedStageDetails: failedStages.map(s => ({
      stageId: s.stageId,
      name: s.name,
      numFailedTasks: s.numFailedTasks,
      failureReason: s.failureReason,
    })),
    healthy: failedJobs.length === 0 && failedStages.length === 0 && stuckStages.length === 0
      && executorHealth.filter(e => e.gcAlert || e.memAlert).length === 0,
  };
}

Executor diagnostics: GC pressure, memory fractions, and shuffle spill

The GET /api/v1/applications/{appId}/executors endpoint returns per-executor metrics that surface the most common Spark performance pathologies. totalGCTime is the cumulative JVM GC time in milliseconds. totalDuration is the cumulative task execution time. When totalGCTime / totalDuration > 0.1, the executor is spending more than 10% of its time in garbage collection — a symptom of heap pressure, typically caused by large dataset broadcasting, high off-heap object retention, or executor memory set too low relative to partition size.

Shuffle spill (diskBytesSpilled on stages, diskUsed on executors) is not reported as an error. Spark records successful task completions even when 100% of shuffle data spilled to disk. Spill to disk is 10–100× slower than in-memory shuffle and typically indicates that executor memory is undersized for the join or aggregation being performed. Alert when total spill across active stages exceeds 10% of total shuffle write bytes.

function diagnoseExecutorProblems(executors) {
  const active = executors.filter(e => e.isActive);

  return active.map(e => {
    const problems = [];

    // GC overhead
    const gcFraction = e.totalDuration > 0 ? e.totalGCTime / e.totalDuration : 0;
    if (gcFraction > 0.1) {
      problems.push({
        type: 'gc_overhead',
        severity: gcFraction > 0.3 ? 'critical' : 'warning',
        detail: `GC consuming ${(gcFraction * 100).toFixed(1)}% of executor time`,
        remediation: 'Increase spark.executor.memory or reduce partition size with repartition()',
      });
    }

    // Memory pressure
    const memFraction = e.maxMemory > 0 ? e.memoryUsed / e.maxMemory : 0;
    if (memFraction > 0.9) {
      problems.push({
        type: 'memory_pressure',
        severity: 'critical',
        detail: `${(memFraction * 100).toFixed(1)}% of executor memory used (${formatBytes(e.memoryUsed)} / ${formatBytes(e.maxMemory)})`,
        remediation: 'Increase spark.executor.memory or spark.memory.fraction, or unpersist cached RDDs',
      });
    }

    // Disk spill
    if (e.diskUsed > 1_073_741_824) {  // > 1 GB spill on this executor
      problems.push({
        type: 'shuffle_spill',
        severity: 'warning',
        detail: `${formatBytes(e.diskUsed)} shuffled to disk (in-memory shuffle unavailable)`,
        remediation: 'Increase spark.executor.memory, reduce broadcast threshold, or increase partition count',
      });
    }

    // Failed tasks — task failures on this executor
    if (e.failedTasks > 0) {
      problems.push({
        type: 'task_failures',
        severity: e.failedTasks > 10 ? 'critical' : 'warning',
        detail: `${e.failedTasks} task failures on executor ${e.id} (${e.hostPort})`,
        remediation: 'Check executor logs for OOM, network errors, or corrupt data',
      });
    }

    return {
      executorId: e.id,
      host: e.hostPort,
      problems,
      healthy: problems.length === 0,
    };
  });
}

function formatBytes(bytes) {
  if (bytes < 1024) return `${bytes} B`;
  if (bytes < 1_048_576) return `${(bytes / 1024).toFixed(1)} KB`;
  if (bytes < 1_073_741_824) return `${(bytes / 1_048_576).toFixed(1)} MB`;
  return `${(bytes / 1_073_741_824).toFixed(1)} GB`;
}

Structured Streaming query health

Spark Structured Streaming runs continuous micro-batch queries inside a Spark application. Each query is identified by a UUID and has a name if queryName was set. The key health signal is whether the query is keeping up with input rate — if inputRowsPerSecond consistently exceeds processedRowsPerSecond, the query is falling behind and trigger interval accumulation will eventually stall the query.

The endpoint GET /api/v1/applications/{appId}/streams/active returns a list of active Structured Streaming queries. Each has status with isActive, message (current state message), and isTriggerActive (true when a micro-batch is running). Check lastProgress for the most recent batch: batchId, numInputRows, inputRowsPerSecond, processedRowsPerSecond, batchDuration, and sink/source information.

async function streamingQueryHealth(spark, appId) {
  let queries;
  try {
    const result = await spark.getStructuredStreamingQueries(appId);
    queries = result.streams ?? result ?? [];
  } catch (err) {
    // Not a Structured Streaming app
    return { hasStreamingQueries: false };
  }

  const queryHealth = queries.map(q => {
    const progress = q.lastProgress;
    const inputRate = progress?.inputRowsPerSecond ?? 0;
    const processRate = progress?.processedRowsPerSecond ?? 0;
    const batchDurationMs = progress?.batchDuration ?? 0;
    const batchId = progress?.batchId ?? -1;

    // Processing lag: input rate exceeds processing rate by >20% consistently
    const lagRatio = processRate > 0 ? inputRate / processRate : null;
    const isLagging = lagRatio !== null && lagRatio > 1.2;

    // Watermark lag for event-time queries
    const eventTimeStats = progress?.eventTime;
    const watermarkMs = eventTimeStats?.watermark ? new Date(eventTimeStats.watermark).getTime() : null;
    const watermarkLagMs = watermarkMs ? Date.now() - watermarkMs : null;

    // Sources and sinks
    const sources = (progress?.sources ?? []).map(s => ({
      description: s.description,
      numInputRows: s.numInputRows,
      startOffset: s.startOffset,
      endOffset: s.endOffset,
    }));

    return {
      id: q.id,
      name: q.name ?? '(unnamed)',
      isActive: q.status?.isActive ?? false,
      statusMessage: q.status?.message ?? null,
      isTriggerActive: q.status?.isTriggerActive ?? false,
      lastBatchId: batchId,
      inputRowsPerSecond: inputRate,
      processedRowsPerSecond: processRate,
      batchDurationMs,
      isLagging,
      lagRatio,
      watermarkLagMs,
      sources,
      // Alert conditions
      alerts: [
        ...(isLagging ? [`Processing lag: input ${inputRate.toFixed(0)} rows/s, processing ${processRate.toFixed(0)} rows/s`] : []),
        ...(watermarkLagMs !== null && watermarkLagMs > 300_000 ? [`Watermark lag: ${Math.round(watermarkLagMs / 1000)}s`] : []),
        ...(!q.status?.isActive ? ['Query is not active'] : []),
      ],
    };
  });

  return {
    hasStreamingQueries: queries.length > 0,
    queries: queryHealth,
    laggingQueries: queryHealth.filter(q => q.isLagging),
    inactiveQueries: queryHealth.filter(q => !q.isActive),
    healthy: queryHealth.every(q => q.isActive && !q.isLagging && q.alerts.length === 0),
  };
}

AliveMCP integration for Spark monitoring

Register AliveMCP probes at two levels for Spark: a driver availability probe (60-second interval, 5-second timeout, checks GET /api/v1/applications) and a job + executor health probe (every 2 minutes, 30-second timeout, fans out to jobs + stages + executors endpoints). For Structured Streaming applications, add a third streaming query lag probe (60-second interval).

async function sparkClusterHealthProbe(sparkUiUrl, appId) {
  const spark = createSparkClient(sparkUiUrl);

  // Driver liveness
  let apps;
  try {
    apps = await spark.getApplications();
  } catch (err) {
    return { healthy: false, layer: 'driver_unavailable', error: err.message };
  }

  if (!apps.find(a => a.id === appId)) {
    return { healthy: false, layer: 'app_not_found', appId };
  }

  // Job + stage + executor health
  const [jobHealth, streamingHealth] = await Promise.allSettled([
    assessSparkJobHealth(spark, appId),
    streamingQueryHealth(spark, appId),
  ]);

  const jobResult = jobHealth.status === 'fulfilled' ? jobHealth.value : null;
  const streamResult = streamingHealth.status === 'fulfilled' ? streamingHealth.value : null;

  const alerts = [];
  if (jobResult) {
    if (jobResult.jobs.failed > 0) alerts.push(`${jobResult.jobs.failed} failed Spark job(s)`);
    if (jobResult.stages.failed > 0) alerts.push(`${jobResult.stages.failed} failed stage(s)`);
    if (jobResult.stages.stuck > 0) alerts.push(`${jobResult.stages.stuck} stage(s) stuck (ACTIVE with 0 running tasks)`);
    if (jobResult.executors.withGcAlert > 0) alerts.push(`${jobResult.executors.withGcAlert} executor(s) with GC overhead > 10%`);
    if (jobResult.executors.withMemAlert > 0) alerts.push(`${jobResult.executors.withMemAlert} executor(s) with memory > 90%`);
    if (jobResult.stages.totalSpillBytes > 1_073_741_824) {
      alerts.push(`${formatBytes(jobResult.stages.totalSpillBytes)} shuffled to disk (spill)`);
    }
  }
  if (streamResult?.hasStreamingQueries) {
    if (streamResult.inactiveQueries.length > 0) alerts.push(`${streamResult.inactiveQueries.length} streaming query(ies) not active`);
    if (streamResult.laggingQueries.length > 0) alerts.push(`${streamResult.laggingQueries.length} streaming query(ies) lagging behind input rate`);
  }

  return {
    healthy: alerts.length === 0,
    alerts,
    job: jobResult,
    streaming: streamResult,
  };
}

Related guides

• MCP tools for Apache Flink — streaming job health, checkpoint failures, backpressure detection, and TaskManager state
• MCP tools for dbt — model run status, test failures, source freshness, and run artifact parsing
• MCP tools for Apache Kafka — consumer group lag, partition health, and broker availability
• MCP tools for ClickHouse — ReplicatedMergeTree health, system.parts, and replication queue monitoring
• MCP server health check patterns — composite endpoints and failure classification