tutorial

Monitoring Apache Celeborn with Vigilmon

Apache Celeborn externalizes Spark and Flink shuffle data to dedicated worker nodes — but a master crash or worker disk overload causes silent job failures. Here's how to monitor Celeborn end-to-end with Vigilmon.

Apache Celeborn (top-level Apache project, 2023) solves a pain point that every team running large Spark or Flink jobs at scale eventually hits: shuffle I/O contention on executor nodes. In standard Spark shuffle, map tasks write data to the executor's local disk, and reduce tasks fetch it over the network from those executors. At scale, this creates two problems: executors spend significant CPU time waiting on local disk I/O for shuffle reads and writes, and if an executor crashes mid-job, all the shuffle data on its disk is lost — triggering expensive stage retries. Celeborn externalizes shuffle data to dedicated Celeborn worker nodes. Executors push shuffle data to workers over the network; reduce tasks pull from workers instead of from remote executor disks. This separates computation from shuffle storage, reduces executor disk pressure, and makes jobs resilient to executor failures. But Celeborn introduces its own operational complexity: a master crash blocks new job registrations, a worker disk overflowing at 90% causes push rejections, and orphan partitions from failed jobs silently consume worker disk until free space is exhausted. Vigilmon provides the external monitoring layer to catch these failures before they stall your Spark production workloads.

What You'll Set Up

  • HTTP probe for Celeborn master process health and leadership status
  • Heartbeat monitors for Celeborn worker processes on each node
  • Worker disk usage monitoring with push-rejection threshold alerts
  • Worker JVM heap usage monitoring for GC pressure detection
  • Job registration success rate monitoring
  • Orphan partition cleanup monitoring
  • Spark and Flink integration health via connector heartbeats

Prerequisites

  • Apache Celeborn 0.5+ with at least one Master and two or more Workers deployed
  • Apache Spark 3.3+ or Apache Flink 1.17+ configured with the Celeborn shuffle plugin
  • Celeborn Master running in high-availability mode (two masters with ZooKeeper) for production
  • A free Vigilmon account

Why Monitoring Celeborn Matters

Celeborn sits on the critical path of every shuffle operation in Spark and Flink jobs that use it. Its failure modes are fast and often silent:

  • Master crash — The Celeborn Master coordinates shuffle worker selection for job registrations. A master crash causes all new Spark job registrations to fail. Spark falls back to local shuffle if configured (celeborn.client.spark.shuffle.writer fallback mode), but this fallback happens without operator notification — your jobs continue, but without Celeborn's benefits, and local executor disks start filling up.
  • Worker crash — Celeborn Workers hold in-flight shuffle data for running Spark stages. A worker crash causes all Spark stages that had shuffle data on that worker to fail and retry. Unlike executor failures, a Celeborn worker failure cannot be compensated by Celeborn itself — Spark must re-run the upstream map stage.
  • Worker disk at 90% capacity — Celeborn workers reject push requests when disk usage exceeds their configured threshold (default 90%). Spark map tasks get push-rejection errors, and the stage fails. This happens suddenly as disk fills during a large job rather than gradually.
  • JVM heap pressure on workers — Celeborn buffers shuffle data in JVM heap before flushing to disk. High heap usage (>80%) causes frequent GC pauses, which manifests as write latency spikes. Unchecked, it leads to OutOfMemoryError and a worker crash.
  • Orphan partitions — When Spark jobs fail before cleanup, their shuffle partitions remain on Celeborn workers. Over time, orphan partitions accumulate and consume significant disk space, potentially causing disk pressure for healthy jobs.

Step 1: Monitor the Celeborn Master

The Celeborn Master exposes an HTTP management endpoint. Add a Vigilmon HTTP monitor:

  1. In Vigilmon, click Add Monitor → HTTP / HTTPS.
  2. Enter: http://celeborn-master:9098/metrics/prometheus (or your Celeborn master metrics endpoint).
  3. Set Check interval to 1 minute.
  4. Set Expected HTTP status to 200.
  5. Click Save.

For the master leadership probe (active/standby HA mode), add a TCP port monitor:

  1. Click Add Monitor → TCP Port.
  2. Enter celeborn-master:9097 (the RPC port).
  3. Set Check interval to 1 minute.
  4. Click Save.

For a deeper health check that validates the master's active state, build a lightweight sidecar:

#!/bin/bash
# /opt/monitoring/check-celeborn-master.sh
MASTER_URL="http://celeborn-master:9098"
HEARTBEAT_URL="https://vigilmon.online/heartbeat/YOUR_MASTER_KEY"

# Check master status via the Celeborn master REST API
STATUS=$(curl -sf "$MASTER_URL/api/v1/conf" | python3 -c "
import sys, json
try:
    data = json.load(sys.stdin)
    # If the master responds to conf queries, it's active
    print('active')
except:
    print('down')
" 2>/dev/null)

if [ "$STATUS" = "active" ]; then
  curl -fsS "$HEARTBEAT_URL"
fi

Schedule every 1 minute. Set the Vigilmon heartbeat interval to 3 minutes. A master that has been unreachable for 3 minutes triggers an alert — by that point, any Spark jobs trying to register new shuffles will have failed.

Step 2: Monitor Celeborn Worker Processes

Each Celeborn Worker must be healthy for shuffle data to be available. Create a heartbeat monitor per worker:

  1. In Vigilmon, click Add Monitor → Cron Heartbeat.
  2. Name it: Celeborn Worker - worker1.
  3. Set the expected interval to 2 minutes.
  4. Copy the heartbeat URL.

On each worker host, create a cron job that confirms the worker process is alive and responding:

#!/bin/bash
# /etc/cron.d/celeborn-worker-heartbeat
# Check that the Celeborn worker is running and responding to requests
*/1 * * * * celeborn curl -sf "http://localhost:9096/metrics/prometheus" \
  | grep -q "celeborn_worker" && \
  curl -fsS "https://vigilmon.online/heartbeat/YOUR_WORKER1_KEY" > /dev/null 2>&1

For a consolidated worker fleet check, add a heartbeat that counts healthy workers from the master's worker registry:

#!/bin/bash
# /opt/monitoring/check-celeborn-worker-count.sh
MIN_WORKERS=3
HEARTBEAT_URL="https://vigilmon.online/heartbeat/YOUR_WORKER_COUNT_KEY"

ALIVE_WORKERS=$(curl -sf "http://celeborn-master:9098/api/v1/workers" | \
  python3 -c "
import sys, json
data = json.load(sys.stdin)
alive = [w for w in data.get('workers', []) if w.get('state') == 'Alive']
print(len(alive))
" 2>/dev/null)

if [ "${ALIVE_WORKERS:-0}" -ge "$MIN_WORKERS" ]; then
  curl -fsS "$HEARTBEAT_URL"
fi

Schedule every 2 minutes. Set the heartbeat interval to 5 minutes.

Step 3: Monitor Worker Disk Usage

Worker disk at >90% capacity causes push rejections. Monitor disk usage per worker and alert before the threshold is hit:

#!/bin/bash
# /opt/monitoring/check-celeborn-disk.sh
# Run on each Celeborn worker node

MAX_DISK_PCT=85   # Alert at 85% — before Celeborn's 90% push-rejection threshold
HEARTBEAT_URL="https://vigilmon.online/heartbeat/YOUR_DISK_WORKER1_KEY"

# Check disk usage for each configured Celeborn data directory
CELEBORN_DIRS="${CELEBORN_WORKER_STORAGE_DIRS:-/mnt/disk1/celeborn,/mnt/disk2/celeborn}"

OVER_THRESHOLD=0
for DIR in $(echo "$CELEBORN_DIRS" | tr ',' ' '); do
  USAGE=$(df "$DIR" 2>/dev/null | tail -1 | awk '{print $5}' | tr -d '%')
  if [ "${USAGE:-0}" -gt "$MAX_DISK_PCT" ]; then
    OVER_THRESHOLD=1
    break
  fi
done

if [ "$OVER_THRESHOLD" -eq 0 ]; then
  curl -fsS "$HEARTBEAT_URL"
fi

Run every 5 minutes. Set the Vigilmon heartbeat interval to 10 minutes. Disk filling is typically gradual during large jobs, so a 10-minute detection window gives you time to intervene before the 90% hard threshold triggers push rejections.

Also add a Vigilmon HTTP monitor for the worker metrics endpoint that alerts on slow response (a stressed worker under disk pressure often has degraded metrics responsiveness too):

  1. Click Add Monitor → HTTP / HTTPS.
  2. Enter: http://celeborn-worker1:9096/metrics/prometheus.
  3. Set Check interval to 2 minutes.
  4. Set Response time threshold to 5000ms.
  5. Click Save.

Step 4: Monitor Worker JVM Heap Usage

Celeborn workers buffer shuffle data in JVM heap. Heap >80% causes GC pressure; unchecked it leads to worker crashes. Add a heap usage heartbeat per worker:

#!/bin/bash
# /opt/monitoring/check-celeborn-heap.sh
# Run on each Celeborn worker node

MAX_HEAP_PCT=78   # Alert at 78% — before GC becomes severe at 80%+
HEARTBEAT_URL="https://vigilmon.online/heartbeat/YOUR_HEAP_WORKER1_KEY"

# Query JVM heap usage via the Celeborn Prometheus metrics endpoint
HEAP_METRICS=$(curl -sf "http://localhost:9096/metrics/prometheus" 2>/dev/null)

HEAP_USED=$(echo "$HEAP_METRICS" | grep 'jvm_memory_bytes_used{area="heap"' | \
  awk '{print $2}' | head -1)
HEAP_MAX=$(echo "$HEAP_METRICS" | grep 'jvm_memory_bytes_max{area="heap"' | \
  awk '{print $2}' | head -1)

if [ -n "$HEAP_USED" ] && [ -n "$HEAP_MAX" ] && [ "$HEAP_MAX" -gt 0 ]; then
  HEAP_PCT=$(python3 -c "print(int($HEAP_USED / $HEAP_MAX * 100))" 2>/dev/null)
  if [ "${HEAP_PCT:-100}" -lt "$MAX_HEAP_PCT" ]; then
    curl -fsS "$HEARTBEAT_URL"
  fi
fi

Run every 2 minutes. Set the Vigilmon heartbeat interval to 5 minutes. A heap that has been above 80% for 5 minutes without a heartbeat indicates sustained GC pressure and warrants investigation.

Step 5: Monitor Job Registration Success Rate

Spark jobs register with the Celeborn master at job start. Registration failures cause jobs to either fail immediately or fall back to local shuffle. Monitor registration success with a test registration heartbeat:

#!/bin/bash
# /opt/monitoring/check-celeborn-registration.sh
MASTER_URL="http://celeborn-master:9098"
HEARTBEAT_URL="https://vigilmon.online/heartbeat/YOUR_REGISTRATION_KEY"

# Verify the master's worker registry has enough workers to accept registrations
# (a master with zero alive workers will reject new job registrations)
WORKER_COUNT=$(curl -sf "$MASTER_URL/api/v1/workers" | \
  python3 -c "
import sys, json
data = json.load(sys.stdin)
print(len([w for w in data.get('workers', []) if w.get('state') == 'Alive']))
" 2>/dev/null)

if [ "${WORKER_COUNT:-0}" -ge 1 ]; then
  curl -fsS "$HEARTBEAT_URL"
fi

For a true end-to-end registration test, integrate into a synthetic Spark job that runs on a schedule and verifies Celeborn shuffle registration succeeds:

#!/usr/bin/env python3
# /opt/monitoring/celeborn_smoke_test.py
# Runs a minimal Spark job that exercises Celeborn shuffle registration

from pyspark.sql import SparkSession
import requests, os, sys

def run_celeborn_smoke_test():
    spark = SparkSession.builder \
        .appName("celeborn-health-check") \
        .config("spark.shuffle.manager", "org.apache.spark.shuffle.celeborn.SparkShuffleManager") \
        .config("spark.celeborn.master.endpoints", os.environ["CELEBORN_MASTER"]) \
        .getOrCreate()

    try:
        # A reduceByKey forces a shuffle through Celeborn
        rdd = spark.sparkContext.parallelize(range(1000), 10)
        result = rdd.map(lambda x: (x % 10, 1)).reduceByKey(lambda a, b: a + b).count()

        if result == 10:  # expect 10 distinct keys
            requests.get(os.environ["VIGILMON_CELEBORN_SMOKE_HEARTBEAT"], timeout=5)
    finally:
        spark.stop()

if __name__ == "__main__":
    run_celeborn_smoke_test()

Schedule this smoke test to run every 30 minutes as a cron job. Set the heartbeat interval to 60 minutes. This is your end-to-end Celeborn integration test — it exercises master registration, worker push, and worker fetch in a single run.

Step 6: Monitor Orphan Partition Cleanup

Orphan partitions from failed jobs accumulate on workers and consume disk space. Monitor their presence:

#!/bin/bash
# /opt/monitoring/check-celeborn-orphans.sh
MAX_ORPHAN_AGE_HOURS=24
HEARTBEAT_URL="https://vigilmon.online/heartbeat/YOUR_ORPHAN_KEY"

# Query the Celeborn master's partition management API for orphaned partitions
ORPHAN_COUNT=$(curl -sf "http://celeborn-master:9098/api/v1/partitions/orphan" | \
  python3 -c "
import sys, json
from datetime import datetime, timezone

data = json.load(sys.stdin)
now = datetime.now(timezone.utc).timestamp() * 1000
max_age_ms = $MAX_ORPHAN_AGE_HOURS * 3600 * 1000

old_orphans = [
    p for p in data.get('partitions', [])
    if now - p.get('createdAtMs', now) > max_age_ms
]
print(len(old_orphans))
" 2>/dev/null)

if [ "${ORPHAN_COUNT:-99}" -eq 0 ]; then
  curl -fsS "$HEARTBEAT_URL"
fi

Run every hour. Set the Vigilmon heartbeat interval to 3 hours. Orphan partitions older than 24 hours indicate the cleanup task is not running — add a PagerDuty alert only if disk usage is simultaneously above 70%.

Step 7: Configure Alert Routing

| Monitor | Alert Channel | Severity | Impact | |---|---|---|---| | Celeborn Master TCP / HTTP | PagerDuty + Slack | Critical | New Spark job shuffle registrations fail | | Worker process heartbeat (any node) | PagerDuty + Slack | Critical | In-flight shuffle data loss → Spark stage retries | | Worker disk usage heartbeat | Slack | High | Push rejections → Spark stage failures | | Worker JVM heap heartbeat | Slack | High | GC pressure → write latency spikes | | Job registration heartbeat | Slack | High | Spark shuffle falling back to local disk | | Celeborn smoke test heartbeat | Slack | High | End-to-end shuffle broken | | Orphan partition heartbeat | Email | Medium | Disk space leak from failed jobs |

Set consecutive failures before alert to 1 for the master and worker monitors. Worker failures have immediate job impact. Set it to 3 for disk and heap monitors to smooth over brief measurement spikes.

For clusters running critical batch processing windows (e.g., nightly ETL jobs), configure Vigilmon maintenance windows to suppress non-critical alerts during planned large-shuffle jobs where disk usage is expected to spike temporarily.

Conclusion

Apache Celeborn gives your Spark and Flink shuffle operations independence from executor disk I/O — but it introduces its own operational surface area that needs monitoring. A master crash silently falls back to local shuffle without notification; a worker disk overflow rejects pushes in the middle of a running stage; orphan partitions slowly starve worker nodes of free space. With Vigilmon monitoring the master, workers, disk usage, JVM heap, and job registration success, you have the external visibility to catch these failures before they cascade into production job failures.

Start with the Celeborn Master HTTP probe and worker process heartbeats — these cover the hard failures. Then add disk and heap monitoring to catch the gradual failures that build up during large shuffle workloads. Sign up for a free Vigilmon account to get started.

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