How to Monitor Delta Lake Tables and Pipelines with Vigilmon

Delta Lake is the open-source storage layer that adds ACID transactions, scalable metadata handling, and time travel to data lakes built on Apache Spark. Teams running Delta Lake on Databricks, AWS EMR, or open-source Apache Spark rely on Delta pipelines to keep their analytics current — but when a Delta write job fails or a vacuum operation corrupts a table's history, downstream dashboards silently show stale data.

This tutorial walks you through monitoring Delta Lake pipelines, structured streaming jobs, and the Delta transaction log using Vigilmon — the external uptime monitoring tool that catches what Spark job schedulers miss.

Why Delta Lake pipelines need external monitoring

Delta Lake introduces failure modes that traditional job monitoring doesn't surface:

• Checkpoint failures in structured streaming — a streaming Delta writer crashes and Spark restarts it from a stale checkpoint, causing data loss or out-of-order processing that only shows up days later in reconciliation
• Concurrent write conflicts — two batch jobs write to the same Delta table simultaneously; one fails with ConcurrentAppendException, logs the error, and exits cleanly — but no alert fires
• Vacuum removes needed files — a VACUUM runs with retentionHours=0 (common in tutorials), deletes files still referenced by an in-flight read job, and breaks time-travel queries
• Transaction log corruption — the _delta_log directory grows without bound or a partial write leaves the log in an inconsistent state; all reads fail until the log is repaired

External monitoring provides an independent view that checks actual data freshness and endpoint health — not just whether the Spark application submitted successfully.

What you'll need

• A Delta Lake pipeline on Databricks, AWS EMR, or standalone Spark
• Python 3.8+ (for the sidecar health server)
• A free Vigilmon account

Step 1: Build a pipeline health endpoint

Add a small health server to your Spark driver process that Vigilmon can probe every minute.

# delta_health_server.py
from http.server import BaseHTTPRequestHandler, HTTPServer
from delta.tables import DeltaTable
import json, time, threading

_state = {"status": "starting", "last_commit_ts": None, "table_version": None}

class DeltaHealthHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/health":
            now = time.time()
            lag = None
            if _state["last_commit_ts"]:
                lag = int(now - _state["last_commit_ts"])

            body = json.dumps({
                "status": _state["status"],
                "table_version": _state["table_version"],
                "last_commit_age_seconds": lag
            })
            code = 200 if _state["status"] == "ok" else 503
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(body.encode())
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, *args):
        pass

def start_health_server(port: int = 8080):
    httpd = HTTPServer(("0.0.0.0", port), DeltaHealthHandler)
    t = threading.Thread(target=httpd.serve_forever, daemon=True)
    t.start()
    return httpd

Call start_health_server() before your first Delta write, and update _state after each successful commit:

from pyspark.sql import SparkSession
from delta_health_server import start_health_server, _state
import time

spark = SparkSession.builder \
    .appName("delta-ingestion") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

start_health_server(port=8080)

# ... your ingestion logic ...
df.write.format("delta").mode("append").save("/mnt/delta/events")

# Read back the current version to confirm success
from delta.tables import DeltaTable
dt = DeltaTable.forPath(spark, "/mnt/delta/events")
version = dt.history(1).select("version").collect()[0][0]

_state["status"] = "ok"
_state["last_commit_ts"] = time.time()
_state["table_version"] = version

Step 2: Monitor Delta table history via the transaction log

The Delta transaction log (_delta_log/) is the source of truth for table health. You can monitor its freshness without running Spark by checking the latest log file's modification time from S3 or HDFS:

# lambda_delta_health.py — AWS Lambda function for serverless clusters
import boto3, json, time

s3 = boto3.client("s3")

def handler(event, context):
    bucket = "your-datalake-bucket"
    prefix = "delta/events/_delta_log/"

    response = s3.list_objects_v2(
        Bucket=bucket,
        Prefix=prefix,
        MaxKeys=10
    )
    objects = sorted(
        response.get("Contents", []),
        key=lambda o: o["LastModified"],
        reverse=True
    )

    if not objects:
        return {"statusCode": 503, "body": json.dumps({"status": "no_log_files"})}

    latest = objects[0]
    age_seconds = int(time.time() - latest["LastModified"].timestamp())
    max_expected_age = 3600  # expect a commit at least hourly

    status = "ok" if age_seconds < max_expected_age else "stale"
    return {
        "statusCode": 200 if status == "ok" else 503,
        "body": json.dumps({
            "status": status,
            "latest_log_file": latest["Key"],
            "age_seconds": age_seconds
        })
    }

Deploy this as a Lambda function and expose it via API Gateway — Vigilmon gets a stable HTTPS URL to probe even on serverless clusters.

Step 3: Monitor structured streaming Delta checkpoints

Structured streaming jobs write checkpoints to track progress. A stale checkpoint directory means the job has stopped writing:

import os, json, time
from pathlib import Path

def check_streaming_job_health(checkpoint_path: str, max_lag_seconds: int = 300):
    """Check if a Delta streaming job is making progress."""
    commits_dir = Path(checkpoint_path) / "commits"
    if not commits_dir.exists():
        return {"status": "no_checkpoint", "code": 503}

    # Find the most recently modified commit file
    commit_files = sorted(commits_dir.iterdir(), key=lambda f: f.stat().st_mtime)
    if not commit_files:
        return {"status": "no_commits", "code": 503}

    latest_mtime = commit_files[-1].stat().st_mtime
    lag = int(time.time() - latest_mtime)

    if lag > max_lag_seconds:
        return {"status": "stale", "lag_seconds": lag, "code": 503}
    return {"status": "ok", "lag_seconds": lag, "code": 200}

Expose this from your health server's /streaming-health path and add it as a separate Vigilmon monitor.

Step 4: Add Vigilmon HTTP monitoring

With your health endpoints running:

1. Log in to vigilmon.online → Monitors → New Monitor → HTTP / HTTPS
2. URL: http://<spark-driver-host>:8080/health
3. Expected status code: 200
4. Optional keyword check: "status":"ok"
5. Check interval: 1 minute
6. Save

For the streaming checkpoint monitor:

1. New Monitor → HTTP / HTTPS
2. URL: http://<spark-driver-host>:8080/streaming-health
3. Expected status: 200
4. Keyword: "status":"ok"
5. Save

For the Lambda/API Gateway endpoint:

1. New Monitor → HTTP / HTTPS
2. URL: https://<api-gateway-id>.execute-api.<region>.amazonaws.com/prod/delta-health
3. Expected status: 200
4. Save

Step 5: Send pipeline failure alerts directly to Vigilmon

Wire your Spark error handler to push an alert the moment a job fails:

import requests, os

VIGILMON_API_KEY = os.environ["VIGILMON_API_KEY"]
VIGILMON_MONITOR_ID = os.environ["VIGILMON_MONITOR_ID"]

def push_incident(message: str):
    try:
        requests.post(
            f"https://vigilmon.online/api/monitors/{VIGILMON_MONITOR_ID}/incidents",
            headers={"Authorization": f"Bearer {VIGILMON_API_KEY}"},
            json={"message": message, "severity": "critical"},
            timeout=5
        )
    except Exception:
        pass  # best-effort; don't let alerting break the error path

try:
    run_delta_pipeline(spark)
except Exception as e:
    push_incident(f"Delta Lake ingestion failed: {type(e).__name__}: {e}")
    raise

Step 6: Alert on ConcurrentAppendException specifically

Concurrent write conflicts are silent killers in Delta Lake multi-writer setups. Add a specific handler:

from delta.exceptions import ConcurrentAppendException, ConcurrentDeleteReadException

try:
    df.write.format("delta").mode("append").save(table_path)
except ConcurrentAppendException as e:
    push_incident(f"Delta concurrent write conflict on {table_path}: {e}")
    raise
except ConcurrentDeleteReadException as e:
    push_incident(f"Delta concurrent delete-read conflict on {table_path}: {e}")
    raise

Step 7: Create a status page for your data platform

1. In Vigilmon, go to Status Pages → New Status Page
2. Add your Delta Lake monitors (batch health, streaming health, transaction log freshness)
3. Publish the page

Teams consuming your Delta tables can check this page before filing a "data looks wrong" support ticket.

Monitoring checklist for Delta Lake

| What to monitor | Vigilmon type | Alert condition | |---|---|---| | Batch pipeline /health | HTTP | Non-200 or "status":"stale" | | Streaming job checkpoint freshness | HTTP | Non-200 or lag_seconds > 300 | | Transaction log freshness (Lambda) | HTTP | Non-200 or "status":"stale" | | S3 API Gateway for serverless clusters | HTTP | Non-200 response | | Driver host TCP reachability | TCP Port | Connection refused |

Summary

Delta Lake gives your data lake ACID guarantees — but the jobs that maintain those guarantees are as critical as any production service. With Vigilmon you get:

• 1-minute HTTP probes on batch and streaming pipeline health
• Transaction log freshness checks that catch stale data before analysts notice
• Instant webhook alerts when ConcurrentAppendException or checkpointing failures occur
• A public status page your data consumers can bookmark

Start monitoring your Delta Lake pipelines for free →