tutorial

How to Monitor Apache Mahout Distributed ML Jobs and Spark Backend Health with Vigilmon

Silent Apache Mahout training job failures and hung Spark executors can kill your ML pipelines without any alert. Learn how to monitor Mahout job liveness, Spark cluster health, and model training heartbeats with Vigilmon.

Apache Mahout is the distributed machine learning library that runs collaborative filtering, clustering, and classification algorithms on Apache Spark (or Hadoop MapReduce) — but when a Mahout training job silently fails because a Spark executor ran out of memory, a Zookeeper session expired mid-job, or the Spark driver lost contact with the cluster manager, your ML model refresh pipeline stalls without any visible alarm. Mahout jobs often run on long intervals (nightly retraining, weekly model updates), meaning a single failure can go undetected for days.

Vigilmon gives you external visibility into Mahout job health through Spark cluster monitoring and heartbeat monitors that confirm each model training cycle completes successfully. This tutorial covers both.

Why Mahout Needs External Monitoring

Mahout's built-in observability (Spark History Server, YARN application tracking, Spark UI on port 4040) requires active polling during job execution. External monitoring with Vigilmon adds:

  • Proactive alerting when Mahout model training jobs stop completing on schedule
  • Spark cluster health checks before jobs are submitted
  • Driver and executor failure detection through Spark REST API probes
  • Pipeline heartbeats that fire only after a training job fully succeeds and model artifacts are written
  • Cross-environment visibility confirming dev, staging, and prod training jobs all complete

These layers matter because Mahout/Spark failures manifest differently: driver OOM crashes surface immediately, executor failures trigger retries that may eventually succeed (masking intermittent issues), and YARN preemption kills jobs silently with a non-obvious exit code.

Step 1: Build a Mahout Ecosystem Health Endpoint

Mahout relies on Spark (or YARN) as its execution engine, so health monitoring targets the cluster infrastructure. Build a sidecar that probes the Spark standalone cluster or YARN ResourceManager, plus optionally the model artifact store.

Python Health Sidecar (Spark REST API)

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

app = Flask(__name__)

# Spark standalone master REST API (port 6066 for v1, 8080 for web UI)
SPARK_MASTER_URL = os.environ.get('SPARK_MASTER_URL', 'http://spark-master.example.com:8080')
SPARK_MASTER_API = os.environ.get('SPARK_MASTER_API', 'http://spark-master.example.com:6066')
YARN_RM_HTTP     = os.environ.get('YARN_RM_HTTP', 'http://resourcemanager.example.com:8088')
MODEL_STORE_PATH = os.environ.get('MODEL_STORE_PATH')  # e.g. /models or S3 path prefix

@app.route('/health/mahout')
def mahout_health():
    errors = []

    # Try Spark standalone master first
    spark_ok = False
    try:
        r = requests.get(f'{SPARK_MASTER_URL}/json/', timeout=5)
        if r.status_code == 200:
            data = r.json()
            status = data.get('status', '')
            if status == 'ALIVE':
                spark_ok = True
            else:
                errors.append(f'spark_master_status_{status}')
        else:
            errors.append(f'spark_master_http_{r.status_code}')
    except Exception as e:
        errors.append(f'spark_master_unreachable:{e}')

    # Fall back to YARN if standalone not configured
    if not spark_ok and YARN_RM_HTTP:
        try:
            r = requests.get(f'{YARN_RM_HTTP}/ws/v1/cluster/info', timeout=5)
            if r.status_code == 200:
                rm_state = r.json().get('clusterInfo', {}).get('state', '')
                if rm_state == 'STARTED':
                    spark_ok = True
                else:
                    errors.append(f'yarn_rm_state_{rm_state}')
            else:
                errors.append(f'yarn_rm_http_{r.status_code}')
        except Exception as e:
            errors.append(f'yarn_rm_unreachable:{e}')

    if errors:
        return jsonify({'status': 'down', 'errors': errors}), 503
    return jsonify({'status': 'ok', 'backend': 'spark' if spark_ok else 'unknown'}), 200

@app.route('/health/mahout/models')
def mahout_model_health():
    """Check that the latest model artifact was written recently."""
    if not MODEL_STORE_PATH:
        return jsonify({'status': 'skipped', 'reason': 'MODEL_STORE_PATH not configured'}), 200
    import subprocess, time
    try:
        result = subprocess.run(
            ['hdfs', 'dfs', '-stat', '%Y', MODEL_STORE_PATH],
            capture_output=True, text=True, timeout=10
        )
        if result.returncode != 0:
            return jsonify({'status': 'down', 'error': 'model_store_missing'}), 503
        mtime_ms = int(result.stdout.strip())
        age_hours = (time.time() * 1000 - mtime_ms) / 3_600_000
        max_age_hours = float(os.environ.get('MODEL_MAX_AGE_HOURS', '26'))
        if age_hours > max_age_hours:
            return jsonify({'status': 'stale', 'age_hours': round(age_hours, 1)}), 503
        return jsonify({'status': 'ok', 'age_hours': round(age_hours, 1)}), 200
    except Exception as e:
        return jsonify({'status': 'down', 'error': str(e)}), 503

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

Node.js Health Sidecar

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

const app = express();

const SPARK_MASTER_URL = process.env.SPARK_MASTER_URL || 'http://spark-master.example.com:8080';
const YARN_RM_HTTP     = process.env.YARN_RM_HTTP     || 'http://resourcemanager.example.com:8088';

app.get('/health/mahout', async (req, res) => {
  const errors = [];

  // Check Spark standalone master
  try {
    const r = await axios.get(`${SPARK_MASTER_URL}/json/`, { timeout: 5000 });
    const status = r.data?.status;
    if (status !== 'ALIVE') errors.push(`spark_master_status_${status}`);
  } catch (e) {
    // Fall back to YARN
    try {
      const rm = await axios.get(`${YARN_RM_HTTP}/ws/v1/cluster/info`, { timeout: 5000 });
      const rmState = rm.data?.clusterInfo?.state;
      if (rmState !== 'STARTED') errors.push(`yarn_rm_state_${rmState}`);
    } catch (e2) {
      errors.push(`cluster_unreachable:${e2.message}`);
    }
  }

  if (errors.length) return res.status(503).json({ status: 'down', errors });
  return res.status(200).json({ status: 'ok' });
});

app.listen(3011, () => console.log('mahout-health listening on 3011'));

Step 2: Configure Vigilmon HTTP Monitor for Mahout

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

Add a second monitor for model freshness:

  • URL: https://your-app.example.com/health/mahout/models
  • Expected: 200, body contains "status":"ok"
  • Interval: 30 minutes
  • Alert channel: ML platform team Slack channel

And a direct Spark master probe:

  • URL: http://spark-master.example.com:8080/json/
  • Expected: 200
  • Interval: 1 minute
  • Alert channel: Spark infrastructure on-call

Step 3: Heartbeat Monitoring for Mahout Training Pipelines

Cluster health checks confirm the infrastructure is alive — but they do not tell you whether your Mahout training jobs are completing. A Spark cluster can be healthy while your collaborative filtering job is stuck in a retry loop due to a corrupt training data partition, or producing models with empty output because of a silent feature extraction failure.

Vigilmon heartbeat monitors detect these silent failures: your training pipeline pings Vigilmon after each successful model 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: mahout-model-retraining
  3. Set the expected interval: 24 hours (or your retraining cadence)
  4. Set the grace period: 3 hours
  5. Save — copy the heartbeat URL, e.g. https://vigilmon.online/heartbeat/abc123xyz

Wire It Into Your Mahout Training Script

#!/bin/bash
# mahout_train.sh — Mahout training job with Vigilmon heartbeat

set -e

VIGILMON_HB="${VIGILMON_HEARTBEAT_URL}"
MAHOUT_HOME="/opt/mahout"
HDFS_INPUT="/user/mahout/ratings"
HDFS_OUTPUT="/user/mahout/models/$(date +%Y%m%d)"

echo "[$(date)] Starting Mahout collaborative filtering training..."

# Run Mahout distributed ALS training on Spark
spark-submit \
  --class org.apache.mahout.sparkbindings.als.ALS \
  --master yarn \
  --deploy-mode cluster \
  --num-executors 8 \
  --executor-memory 4g \
  --driver-memory 2g \
  "$MAHOUT_HOME/lib/mahout-spark-*.jar" \
  --input "$HDFS_INPUT" \
  --output "$HDFS_OUTPUT" \
  --rank 20 \
  --lambda 0.065 \
  --iterations 10

EXIT_CODE=$?

if [ $EXIT_CODE -eq 0 ]; then
  echo "[$(date)] Training complete. Pinging Vigilmon heartbeat..."
  curl -s --max-time 10 "$VIGILMON_HB" > /dev/null
  echo "[$(date)] Done."
else
  echo "[$(date)] Training FAILED (exit $EXIT_CODE). Heartbeat NOT sent."
  exit $EXIT_CODE
fi

PySpark / Mahout Python API with Heartbeat

# mahout_train.py — Mahout-on-Spark via SparkContext + Vigilmon heartbeat
from pyspark import SparkContext, SparkConf
import requests
import os

VIGILMON_HB = os.environ['VIGILMON_HEARTBEAT_URL']
HDFS_INPUT  = os.environ.get('HDFS_INPUT', 'hdfs:///user/mahout/ratings')
HDFS_OUTPUT = os.environ.get('HDFS_OUTPUT', 'hdfs:///user/mahout/models/latest')

conf = SparkConf().setAppName('mahout-als-training')
sc   = SparkContext(conf=conf)

try:
    ratings = sc.textFile(HDFS_INPUT)

    # Example: parse and train using Mahout's distributed ALS via Spark RDD API
    parsed = ratings.map(lambda line: line.split(',')) \
                    .map(lambda parts: (int(parts[0]), int(parts[1]), float(parts[2])))

    # ... ALS training logic via org.apache.mahout integration ...

    parsed.saveAsTextFile(HDFS_OUTPUT)

    # Ping Vigilmon only after model artifacts are successfully written
    requests.get(VIGILMON_HB, timeout=10)
    print("Model written to HDFS. Vigilmon heartbeat sent.")
finally:
    sc.stop()

Airflow DAG Integration

# dags/mahout_training_dag.py
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

with DAG(
    'mahout_als_retraining',
    schedule_interval='0 2 * * *',
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    train = BashOperator(
        task_id='run_mahout_training',
        bash_command='/opt/pipelines/mahout_train.sh',
        env={'VIGILMON_HEARTBEAT_URL': '{{ var.value.vigilmon_mahout_hb }}'},
    )

Step 4: Alert Routing for Mahout Failures

Mahout failures have distinct urgency levels: Spark cluster outages block all training immediately, while stale model artifacts degrade recommendation quality gradually.

Configure alert routing in Vigilmon:

| Monitor | Alert Channel | Priority | |---|---|---| | Spark master web UI (port 8080) | Slack + PagerDuty | P1 | | Mahout cluster health /health/mahout | Slack + PagerDuty | P1 | | Model freshness check /health/mahout/models | Slack | P2 | | Heartbeat: nightly ALS retraining | Slack + email | P2 | | Heartbeat: weekly clustering job | Email | P3 |

Set response time thresholds for early warning:

  • Alert at 6000ms for the cluster health endpoint (slow Spark REST responses signal driver memory pressure)
  • Alert at 15000ms for the model store check (slow HDFS stat calls signal DataNode or NameNode load)

For recommendation systems with SLA requirements, configure a model staleness budget: if a model artifact is older than 30 hours (exceeding a 24-hour retraining cycle plus a 6-hour grace window), recommendation quality degradation becomes measurable — escalate immediately rather than waiting for the next scheduled check.

Summary

Mahout failures span Spark/YARN infrastructure (immediate impact) and model pipeline execution (gradual quality degradation). External monitoring catches both:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on Spark master web UI | Spark cluster availability, master liveness | | HTTP monitor on cluster health endpoint | Spark + YARN reachability, backend selection | | HTTP monitor on model freshness endpoint | Model artifact age, HDFS write confirmation | | Heartbeat monitor | Training job completion, ML pipeline liveness |

Get started free at vigilmon.online — your first Mahout training pipeline 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