tutorial

Monitoring Hamilton (DAGWorks) Data Pipelines with Vigilmon

Hamilton is a lightweight Python library for defining data and ML pipelines as DAGs — but silent transform failures, stale dataflow graphs, and long-running nodes degrade model freshness and data quality without raising exceptions. Here's how to monitor Hamilton pipeline execution, node health, dataflow freshness, and driver process availability with Vigilmon.

Hamilton is a lightweight Python library from DAGWorks that lets you define data transformations and ML feature pipelines as directed acyclic graphs (DAGs) by writing annotated Python functions. Each function becomes a node; Hamilton resolves dependencies automatically and executes the graph. When Hamilton pipelines run on a schedule — feeding feature stores, refreshing ML model inputs, or powering ETL workflows — silent failures are the most dangerous failure mode: a node that raises an unhandled exception, a driver process that exits without logging, or a dataflow graph that produces stale outputs without any alerting. Vigilmon gives you external monitoring for Hamilton pipeline execution health, node-level failure detection, dataflow freshness, and driver process availability so your data and ML teams catch pipeline failures before they reach downstream consumers.

What You'll Set Up

  • Hamilton driver process health via cron heartbeats
  • Pipeline execution success and failure detection
  • Node-level error monitoring within dataflows
  • Dataflow output freshness verification
  • Resource utilization during pipeline execution

Prerequisites

  • Hamilton 1.x+ installed (pip install sf-hamilton)
  • Pipeline runs on a schedule (cron, Airflow, or similar)
  • Write access to pipeline output storage (files, database, or S3)
  • A free Vigilmon account

Step 1: Monitor Pipeline Driver Process Health

Hamilton pipelines execute synchronously within a Python driver process. When the driver crashes mid-execution — due to an OOM kill, an unhandled exception in the top-level script, or a transient import error — the pipeline produces no output and leaves no signal unless you have external monitoring. Start with a heartbeat that only fires on successful completion:

  1. In Vigilmon, click Add Monitor → Cron Heartbeat.
  2. Set Expected ping interval to match your pipeline schedule (e.g., 10 minutes for a pipeline that runs every 10 minutes).
  3. Copy the heartbeat URL (e.g., https://vigilmon.online/heartbeat/abc123).

Wrap your Hamilton driver invocation:

#!/usr/bin/env python3
# run_pipeline.py

import requests
import sys
import traceback
from hamilton import driver
import my_pipeline_module  # Your Hamilton module with transform functions

HEARTBEAT_URL = "https://vigilmon.online/heartbeat/abc123"

def run():
    config = {
        "environment": "production",
        "date": "2024-01-15",
    }

    dr = driver.Builder().with_modules(my_pipeline_module).with_config(config).build()

    # Execute the DAG and request final outputs
    output_vars = ["feature_a", "feature_b", "model_input"]
    inputs = {"raw_data_path": "/data/raw/latest.parquet"}

    results = dr.execute(output_vars, inputs=inputs)

    # Ping Vigilmon only on successful completion
    requests.get(HEARTBEAT_URL, timeout=10)
    print(f"Pipeline complete. Outputs: {list(results.keys())}")
    return results

if __name__ == "__main__":
    try:
        run()
    except Exception as e:
        print(f"Pipeline failed: {e}", file=sys.stderr)
        traceback.print_exc()
        sys.exit(1)

If the heartbeat stops arriving, Vigilmon alerts you within one missed interval — far faster than discovering stale model features in production hours later.

Step 2: Detect Node-Level Failures in the Dataflow Graph

Hamilton catches exceptions in individual nodes and propagates them up through the DAG. However, some failures are silent: a node that returns None when it should return a DataFrame, a transform that produces an empty result due to an upstream data issue, or a validation step that short-circuits without raising an exception. Add node-level validation within your pipeline module:

# my_pipeline_module.py

import pandas as pd
from hamilton.function_modifiers import check_output

def raw_events(raw_data_path: str) -> pd.DataFrame:
    """Load raw event data from storage."""
    df = pd.read_parquet(raw_data_path)
    if df.empty:
        raise ValueError(f"raw_events: empty DataFrame loaded from {raw_data_path}")
    return df

def filtered_events(raw_events: pd.DataFrame) -> pd.DataFrame:
    """Filter events to the last 24 hours."""
    result = raw_events[raw_events["timestamp"] > pd.Timestamp.now() - pd.Timedelta("24h")]
    if len(result) < 100:
        raise ValueError(
            f"filtered_events: suspiciously low row count ({len(result)}). "
            f"Expected >= 100 rows in the last 24 hours."
        )
    return result

@check_output(data_type=pd.Series, range=(0.0, 1.0))
def conversion_rate(filtered_events: pd.DataFrame) -> pd.Series:
    """Compute per-user conversion rate."""
    return filtered_events.groupby("user_id")["converted"].mean()

Capture node-level failures in the driver wrapper and report them:

import requests
import json
from datetime import datetime

HEARTBEAT_URL = "https://vigilmon.online/heartbeat/abc123"
FAILURE_LOG = "/var/log/hamilton-failures.jsonl"

def run_with_node_tracking():
    errors = []

    try:
        dr = driver.Builder().with_modules(my_pipeline_module).with_config({}).build()
        results = dr.execute(["conversion_rate"], inputs={"raw_data_path": "/data/raw/latest.parquet"})
        requests.get(HEARTBEAT_URL, timeout=10)
    except Exception as e:
        # Log structured failure for downstream alerting
        failure = {
            "timestamp": datetime.utcnow().isoformat(),
            "pipeline": "feature_pipeline",
            "error": str(e),
            "node": getattr(e, "__cause__", None) and str(e.__cause__),
        }
        with open(FAILURE_LOG, "a") as f:
            f.write(json.dumps(failure) + "\n")
        raise

The structured JSONL log feeds into your log aggregator (Datadog, Loki, CloudWatch) for node-level failure analysis alongside the Vigilmon heartbeat miss for the operational alert.

Step 3: Monitor Dataflow Output Freshness

Hamilton pipelines that feed feature stores, model training jobs, or dashboards must produce fresh outputs on schedule. Even when the pipeline driver completes successfully, the outputs might not be written to storage — a silent write failure, a disk full condition, or a misconfigured output path leaves downstream consumers reading stale data. Monitor the freshness of pipeline outputs directly:

#!/bin/bash
# /usr/local/bin/hamilton-freshness-check.sh

HEARTBEAT_URL="https://vigilmon.online/heartbeat/def456"
OUTPUT_PATH="/data/features/conversion_rate.parquet"
MAX_AGE_MINUTES=15   # Alert if output is older than 15 minutes

if [ ! -f "$OUTPUT_PATH" ]; then
  echo "Output file missing: $OUTPUT_PATH"
  exit 1
fi

FILE_AGE_MINUTES=$(( ($(date +%s) - $(stat -c %Y "$OUTPUT_PATH")) / 60 ))

if [ "$FILE_AGE_MINUTES" -le "$MAX_AGE_MINUTES" ]; then
  echo "Output fresh: ${FILE_AGE_MINUTES} min old (max: ${MAX_AGE_MINUTES} min)"
  curl -s "$HEARTBEAT_URL"
else
  echo "Output stale: ${FILE_AGE_MINUTES} min old (max: ${MAX_AGE_MINUTES} min)"
  exit 1
fi

For S3-backed outputs:

#!/bin/bash
HEARTBEAT_URL="https://vigilmon.online/heartbeat/def456"
S3_PATH="s3://your-feature-store/hamilton/conversion_rate/latest.parquet"
MAX_AGE_MINUTES=15

LAST_MODIFIED=$(aws s3 ls "$S3_PATH" 2>/dev/null | awk '{print $1 " " $2}')

if [ -z "$LAST_MODIFIED" ]; then
  echo "S3 output not found: $S3_PATH"
  exit 1
fi

FILE_AGE_MINUTES=$(python3 -c "
from datetime import datetime, timezone
lm = datetime.strptime('$LAST_MODIFIED', '%Y-%m-%d %H:%M:%S')
lm = lm.replace(tzinfo=timezone.utc)
age = (datetime.now(timezone.utc) - lm).total_seconds() / 60
print(int(age))
")

if [ "$FILE_AGE_MINUTES" -le "$MAX_AGE_MINUTES" ]; then
  curl -s "$HEARTBEAT_URL"
else
  echo "S3 output stale: ${FILE_AGE_MINUTES} min old"
  exit 1
fi

Set the Vigilmon heartbeat interval to match your pipeline's expected output frequency. Freshness monitoring catches write failures independently of execution monitoring — a pipeline can appear to complete successfully but fail silently during the output write phase.

Step 4: Track Execution Time per Pipeline Run

Hamilton DAGs can grow complex as more nodes are added over time. A pipeline that previously took 2 minutes can silently degrade to 15 minutes due to a new upstream data size increase, a slow join on a growing table, or a node that materializes an unexpectedly large intermediate result. Track execution time to catch performance regressions early:

#!/usr/bin/env python3
import time
import requests
import json
from datetime import datetime

HEARTBEAT_URL = "https://vigilmon.online/heartbeat/ghi789"
MAX_EXECUTION_SECONDS = 300   # Alert if pipeline takes more than 5 minutes
METRICS_LOG = "/var/log/hamilton-metrics.jsonl"

def run_timed():
    start = time.monotonic()

    dr = driver.Builder().with_modules(my_pipeline_module).with_config({}).build()
    results = dr.execute(["conversion_rate", "feature_a", "feature_b"],
                         inputs={"raw_data_path": "/data/raw/latest.parquet"})

    elapsed = time.monotonic() - start

    # Log execution metrics
    metric = {
        "timestamp": datetime.utcnow().isoformat(),
        "pipeline": "feature_pipeline",
        "duration_seconds": round(elapsed, 2),
        "outputs": list(results.keys()),
    }
    with open(METRICS_LOG, "a") as f:
        f.write(json.dumps(metric) + "\n")

    if elapsed <= MAX_EXECUTION_SECONDS:
        requests.get(HEARTBEAT_URL, timeout=10)
        print(f"Pipeline OK: {elapsed:.1f}s")
    else:
        print(f"Pipeline too slow: {elapsed:.1f}s > {MAX_EXECUTION_SECONDS}s", flush=True)
        # Don't ping — let Vigilmon alert on the missed heartbeat
        raise RuntimeError(f"Execution time {elapsed:.1f}s exceeded limit {MAX_EXECUTION_SECONDS}s")

Combine with Hamilton's execution tracker for per-node timing:

from hamilton.plugins import h_tqdm

# Use h_tqdm adapter to get per-node timing feedback during development
dr = (driver.Builder()
      .with_modules(my_pipeline_module)
      .with_config({})
      .with_adapters(h_tqdm.ProgressBar())
      .build())

Set the Vigilmon heartbeat to MAX_EXECUTION_SECONDS + pipeline_interval. For a pipeline expected to run in under 5 minutes every 10 minutes, set the heartbeat expected interval to 15 minutes — enough buffer that a single slow run doesn't false-alarm, but a consistently slow or stuck pipeline still alerts.

Step 5: Monitor the Scheduler or Orchestrator Driving Hamilton

Hamilton itself has no built-in scheduler — it relies on an external trigger: a cron job, Airflow DAG, Prefect flow, or custom scheduler. If the scheduler stops triggering the Hamilton driver, the pipeline produces no output and no errors — it simply stops running silently. Monitor the orchestration layer independently:

#!/bin/bash
# /usr/local/bin/hamilton-scheduler-check.sh
# Run this from a DIFFERENT host than the pipeline host

HEARTBEAT_URL="https://vigilmon.online/heartbeat/jkl012"
PIPELINE_HOST="pipeline-worker-01.internal"
PIPELINE_LOG="/var/log/hamilton-metrics.jsonl"
MAX_SILENCE_MINUTES=20  # Alert if no pipeline run in 20 minutes

# SSH to pipeline host and check last execution timestamp
LAST_RUN=$(ssh -o ConnectTimeout=10 "$PIPELINE_HOST" \
  "tail -1 $PIPELINE_LOG 2>/dev/null | python3 -c \"import sys, json; d = json.load(sys.stdin); print(d['timestamp'])\" 2>/dev/null")

if [ -z "$LAST_RUN" ]; then
  echo "Cannot read pipeline log from $PIPELINE_HOST"
  exit 1
fi

AGE_MINUTES=$(python3 -c "
from datetime import datetime, timezone
last = datetime.fromisoformat('$LAST_RUN').replace(tzinfo=timezone.utc)
age = (datetime.now(timezone.utc) - last).total_seconds() / 60
print(int(age))
")

if [ "$AGE_MINUTES" -le "$MAX_SILENCE_MINUTES" ]; then
  echo "Scheduler OK: last run ${AGE_MINUTES} min ago"
  curl -s "$HEARTBEAT_URL"
else
  echo "Scheduler silent: no pipeline run in ${AGE_MINUTES} min (max: ${MAX_SILENCE_MINUTES} min)"
  exit 1
fi

This check runs on a different host intentionally — if the pipeline worker itself goes down, an SSH-based check from a different machine catches both the scheduler failure and host failure at once.

Step 6: Set Up Alert Channels

Configure Vigilmon to route Hamilton pipeline alerts to the right teams:

  1. Go to Alert Channels in Vigilmon.
  2. Add a Slack webhook to #data-platform-alerts — data engineers need pipeline failure context immediately.
  3. Add PagerDuty for the pipeline execution and freshness monitors — stale features in production are severity-1 for ML-dependent products.
  4. Add email notifications for execution time monitors — performance regressions are important but not on-call worthy.
  5. Set Consecutive failures before alert to 2 for freshness monitors — a single delayed pipeline run during deployment doesn't warrant a page.

Add maintenance windows when deploying schema changes or reprocessing historical data:

curl -X POST https://vigilmon.online/api/maintenance \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "monitor_id": "HAMILTON_FRESHNESS_MONITOR_ID",
    "duration_minutes": 60,
    "reason": "Historical backfill - pipeline will produce non-real-time outputs"
  }'

Summary

| Monitor | Target | What It Catches | |---|---|---| | Cron heartbeat (execution) | Driver process completion | Pipeline crash, OOM kill, import error | | Structured logging | Node-level failures | Empty outputs, validation failures, bad data | | Cron heartbeat (freshness) | Output file/S3 modification time | Write failures, disk full, wrong output path | | Cron heartbeat (latency) | Pipeline execution duration | Data volume growth, slow joins, perf regression | | SSH-based check (scheduler) | Last execution timestamp | Cron/Airflow/Prefect not triggering pipeline |

Hamilton's simplicity — plain Python functions, no server, no UI — means failures are equally simple to miss. There's no dashboard showing the last successful DAG run, no built-in alerting when a pipeline stops executing, and no freshness SLA enforcement. Vigilmon's external heartbeat monitoring fills that gap: you know about Hamilton pipeline failures in minutes, not when a downstream model silently degrades and a data scientist asks why features haven't updated.

Start monitoring for free at vigilmon.online.

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