How to Monitor Apache YARN Cluster Health with Vigilmon

Apache YARN (Yet Another Resource Negotiator) is the resource management layer of the Hadoop ecosystem. YARN's ResourceManager schedules jobs across NodeManagers, and when it degrades — through failover, queue saturation, or NodeManager loss — Hadoop jobs, Spark on YARN workloads, and Hive queries queue indefinitely or fail silently without surfacing clear errors to end users.

Vigilmon gives you external visibility into YARN cluster health through HTTP probe monitoring on the YARN REST API and heartbeat monitoring for your critical MapReduce and Spark applications. This tutorial covers both.

Why YARN Needs External Monitoring

YARN exposes a rich REST API at port 8088 (ResourceManager) and 8042 (NodeManager). These endpoints reveal cluster capacity, application state, and queue depths — but only if you're actively querying them. External monitoring with Vigilmon adds:

• Proactive alerting when the YARN ResourceManager fails over or becomes unavailable
• NodeManager health visibility to catch disconnected nodes before cluster capacity drops below job requirements
• Queue saturation alerts when application queues fill up and new jobs cannot be submitted
• Heartbeat monitoring for critical Spark and MapReduce jobs that should complete within an SLA window

These layers work together: a YARN ResourceManager can be healthy while NodeManagers disconnect one by one, quietly reducing the cluster to a state where large jobs cannot be allocated enough containers to run.

Step 1: Build YARN Health Endpoints

YARN's REST API can be queried directly or wrapped in a sidecar for semantic health checks.

Node.js YARN Health Sidecar

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

const app = express();
const YARN_RM = process.env.YARN_RM || 'http://localhost:8088';
const MIN_ACTIVE_NODES = parseInt(process.env.YARN_MIN_NODES || '3');
const MAX_QUEUE_PENDING = parseInt(process.env.YARN_MAX_PENDING || '50');

app.get('/health/yarn', async (req, res) => {
  try {
    // YARN ResourceManager info endpoint
    const info = await axios.get(`${YARN_RM}/ws/v1/cluster/info`, { timeout: 10000 });
    const cluster = info.data.clusterInfo;

    if (cluster.state !== 'STARTED') {
      return res.status(503).json({
        status: 'down',
        reason: 'resourcemanager_not_started',
        state: cluster.state,
      });
    }

    // Check cluster metrics for node health
    const metrics = await axios.get(`${YARN_RM}/ws/v1/cluster/metrics`, { timeout: 10000 });
    const m = metrics.data.clusterMetrics;

    if (m.activeNodes < MIN_ACTIVE_NODES) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'insufficient_active_nodes',
        active_nodes: m.activeNodes,
        min_required: MIN_ACTIVE_NODES,
        lost_nodes: m.lostNodes,
        unhealthy_nodes: m.unhealthyNodes,
      });
    }

    if (m.appsPending > MAX_QUEUE_PENDING) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'application_queue_saturated',
        apps_pending: m.appsPending,
        max_allowed: MAX_QUEUE_PENDING,
        active_nodes: m.activeNodes,
      });
    }

    return res.status(200).json({
      status: 'ok',
      active_nodes: m.activeNodes,
      lost_nodes: m.lostNodes,
      unhealthy_nodes: m.unhealthyNodes,
      apps_running: m.appsRunning,
      apps_pending: m.appsPending,
      available_vcores: m.availableVirtualCores,
      available_mb: m.availableMB,
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.get('/health/yarn/nodes', async (req, res) => {
  try {
    const nodes = await axios.get(`${YARN_RM}/ws/v1/cluster/nodes`, { timeout: 10000 });
    const allNodes = nodes.data.nodes.node || [];

    const running = allNodes.filter(n => n.state === 'RUNNING');
    const lost = allNodes.filter(n => n.state === 'LOST');
    const unhealthy = allNodes.filter(n => n.state === 'UNHEALTHY');
    const decommissioned = allNodes.filter(n => n.state === 'DECOMMISSIONED');

    if (running.length < MIN_ACTIVE_NODES) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'insufficient_running_nodes',
        running: running.length,
        lost: lost.length,
        unhealthy: unhealthy.length,
        lost_nodes: lost.map(n => ({ id: n.id, rack: n.rack })),
        unhealthy_nodes: unhealthy.map(n => ({ id: n.id, healthReport: n.healthReport })),
      });
    }

    return res.status(200).json({
      status: 'ok',
      running_nodes: running.length,
      lost_nodes: lost.length,
      unhealthy_nodes: unhealthy.length,
      decommissioned_nodes: decommissioned.length,
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.get('/health/yarn/queues', async (req, res) => {
  const watchedQueue = process.env.YARN_QUEUE || 'default';

  try {
    const scheduler = await axios.get(`${YARN_RM}/ws/v1/cluster/scheduler`, { timeout: 10000 });
    const queues = scheduler.data.scheduler.schedulerInfo.queues?.queue || [];

    const target = queues.find(q => q.queueName === watchedQueue);
    if (!target) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'queue_not_found',
        queue: watchedQueue,
        available_queues: queues.map(q => q.queueName),
      });
    }

    const usedCapacity = parseFloat(target.usedCapacity || 0);
    if (usedCapacity > 90) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'queue_near_capacity',
        queue: watchedQueue,
        used_capacity_pct: usedCapacity,
        num_applications: target.numApplications,
      });
    }

    return res.status(200).json({
      status: 'ok',
      queue: watchedQueue,
      used_capacity_pct: usedCapacity,
      capacity_pct: parseFloat(target.capacity || 0),
      num_applications: target.numApplications,
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.listen(3009);

Python YARN Health Sidecar

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

app = Flask(__name__)
YARN_RM = os.environ.get('YARN_RM', 'http://localhost:8088')
MIN_ACTIVE_NODES = int(os.environ.get('YARN_MIN_NODES', '3'))

@app.route('/health/yarn')
def yarn_health():
    try:
        info = requests.get(f'{YARN_RM}/ws/v1/cluster/info', timeout=10).json()
        state = info['clusterInfo']['state']
        if state != 'STARTED':
            return jsonify({'status': 'down', 'reason': 'rm_not_started', 'state': state}), 503

        metrics = requests.get(f'{YARN_RM}/ws/v1/cluster/metrics', timeout=10).json()
        m = metrics['clusterMetrics']
        active = m['activeNodes']
        pending = m['appsPending']

        if active < MIN_ACTIVE_NODES:
            return jsonify({
                'status': 'degraded',
                'reason': 'insufficient_active_nodes',
                'active_nodes': active,
                'min_required': MIN_ACTIVE_NODES,
                'lost_nodes': m['lostNodes'],
            }), 503

        if pending > 50:
            return jsonify({
                'status': 'degraded',
                'reason': 'queue_saturated',
                'apps_pending': pending,
            }), 503

        return jsonify({
            'status': 'ok',
            'active_nodes': active,
            'apps_running': m['appsRunning'],
            'apps_pending': pending,
            'available_vcores': m['availableVirtualCores'],
        })
    except Exception as e:
        return jsonify({'status': 'down', 'error': str(e)}), 503

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

Direct YARN API Probing

# ResourceManager health (returns HTTP 200 if healthy)
curl http://yarn-rm:8088/ws/v1/cluster/info

# Cluster metrics (active nodes, pending apps, available resources)
curl http://yarn-rm:8088/ws/v1/cluster/metrics | jq '.clusterMetrics | {activeNodes, lostNodes, appsPending, appsRunning}'

# NodeManager health on individual agent
curl http://yarn-nm-01:8042/ws/v1/node/info | jq '.nodeInfo.healthStatus'

Step 2: Configure Vigilmon HTTP Monitor for YARN

1. Log in to vigilmon.online and go to Monitors → New Monitor
2. Choose HTTP / HTTPS
3. Set the URL to your YARN health endpoint: https://your-cluster.example.com/health/yarn
4. Set the check interval to 1 minute
5. Under Expected response, configure:
   • Status code: 200
   • Response body contains: "status":"ok"
   • Response time threshold: 10000ms
6. Under Alert channels, assign your Slack or PagerDuty channel
7. Save the monitor

Add monitors for node and queue health:

• URL: https://your-cluster.example.com/health/yarn/nodes

• Expected: 200, body contains "status":"ok"

• Interval: 2 minutes

• Alert channel: Slack + PagerDuty (node loss is capacity-critical)

• URL: https://your-cluster.example.com/health/yarn/queues

• Expected: 200, body contains "status":"ok"

• Interval: 2 minutes

• Alert channel: data engineering Slack channel

Vigilmon's multi-region probe consensus prevents brief ResourceManager HA failovers (which typically resolve in 30–60 seconds) from generating false-positive P1 alerts.

Step 3: Heartbeat Monitoring for YARN Applications

YARN cluster health doesn't tell you whether your Spark or MapReduce applications are making progress. A Spark job can be in RUNNING state from YARN's perspective while all its executors are stuck waiting for a shuffle write that will never complete.

Vigilmon heartbeat monitors detect these silent application stalls.

Set Up the Heartbeat Monitor

1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
2. Set the name: spark-on-yarn-etl-job
3. Set the expected interval: 30 minutes (match your job SLA window)
4. Set the grace period: 15 minutes
5. Save — copy the heartbeat URL, e.g. https://vigilmon.online/heartbeat/abc123xyz

Wire It Into a PySpark on YARN Job

# spark_yarn_etl.py
from pyspark.sql import SparkSession
import requests, os

spark = SparkSession.builder \
    .appName('etl-pipeline') \
    .master('yarn') \
    .getOrCreate()

heartbeat_url = os.environ.get('VIGILMON_HEARTBEAT_URL')

def ping_vigilmon():
    if heartbeat_url:
        try:
            requests.get(heartbeat_url, timeout=5)
        except Exception:
            pass

def process_batch(input_path, output_path):
    df = spark.read.parquet(input_path)
    result = df.groupBy('event_type').count()
    result.write.mode('overwrite').orc(output_path)
    return result.count()

# Main pipeline
try:
    count = process_batch(
        os.environ['INPUT_PATH'],
        os.environ['OUTPUT_PATH']
    )
    print(f'Processed {count} records')

    # Ping Vigilmon only on successful completion
    ping_vigilmon()
except Exception as e:
    print(f'Job failed: {e}')
    raise  # Let YARN see the failure; no heartbeat ping
finally:
    spark.stop()

Submit Script with Heartbeat

#!/bin/bash
# submit_spark_job.sh
set -e

HEARTBEAT_URL="${VIGILMON_HEARTBEAT_URL}"

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --conf spark.yarn.submit.waitAppCompletion=true \
  --conf spark.executor.instances=10 \
  --conf spark.executor.memory=4g \
  --conf spark.executor.cores=2 \
  --name "ETL Pipeline" \
  --class com.example.EtlJob \
  etl-job.jar

# Ping Vigilmon only if spark-submit exits 0
if [ -n "$HEARTBEAT_URL" ]; then
  curl -sf "$HEARTBEAT_URL" > /dev/null
fi

MapReduce Job with Heartbeat (Java)

// MRJobRunner.java
import org.apache.hadoop.mapreduce.Job;
import java.net.http.*;
import java.net.URI;

public class MRJobRunner {
    public static void main(String[] args) throws Exception {
        Job job = Job.getInstance();
        // ... configure job ...

        boolean success = job.waitForCompletion(true);
        if (success) {
            String heartbeatUrl = System.getenv("VIGILMON_HEARTBEAT_URL");
            if (heartbeatUrl != null) {
                HttpClient.newHttpClient().send(
                    HttpRequest.newBuilder(URI.create(heartbeatUrl)).GET().build(),
                    HttpResponse.BodyHandlers.discarding()
                );
            }
            System.exit(0);
        } else {
            System.exit(1);
        }
    }
}

Step 4: Alert Routing for YARN Cluster Failures

YARN failures cascade from resource management into every job running on the cluster. Route alerts by the impact scope:

| Monitor | Alert Channel | Priority | |---|---|---| | YARN ResourceManager /health/yarn | Slack + PagerDuty | P1 | | NodeManager availability /health/yarn/nodes | Slack + PagerDuty | P1 | | Queue saturation /health/yarn/queues | Slack + email | P2 | | Heartbeat: Spark ETL job | Slack + email | P2 | | Heartbeat: scheduled MapReduce jobs | Email | P3 |

Set response time thresholds as early warning signals:

• Alert at 5000ms for the ResourceManager health endpoint (slow YARN API responses signal RM overload)
• Alert at 8000ms for node checks (slow NodeManager responses signal network or disk pressure)
• Alert at 3000ms for queue checks (should be fast unless the scheduler is backlogged)

For clusters running SLA-critical batch jobs (end-of-day reports, nightly aggregations), set heartbeat intervals to match the SLA window and use a tighter grace period — this ensures YARN-side stalls surface before the business SLA is breached.

Summary

YARN failures are scheduling failures — they cause jobs to queue indefinitely, executors to stall silently, and NodeManager capacity to shrink until critical workloads can no longer be allocated. External monitoring catches these before your batch SLAs are breached:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/yarn | ResourceManager state, active node count, queue depth | | HTTP monitor on /health/yarn/nodes | NodeManager health, lost and unhealthy node detection | | HTTP monitor on /health/yarn/queues | Queue capacity utilization, application backlog | | Heartbeat monitor | Spark/MapReduce job completion within SLA window |

Get started free at vigilmon.online — your first YARN cluster monitor is running in under two minutes.