tutorial

Monitoring whylogs Data Pipelines with Vigilmon

whylogs profiles your ML data to catch drift, nulls, and schema violations — but the profiling jobs themselves can fail silently. Here's how to monitor your whylogs pipeline health, null rate drift, feature distribution shifts, and constraint violations with Vigilmon.

whylogs is an open source data logging library developed by WhyLabs that computes statistical profiles of datasets in a single pass — capturing per-column null rates, numeric distributions, categorical frequencies, and cardinality estimates. Those profiles catch the silent ML model failures that no infrastructure monitor sees: training-serving skew, gradual feature drift, and upstream schema changes that flip a numeric column to strings. But whylogs is only useful when it's actually running — profiling jobs that crash silently, profile writes that fail to disk, or constraint checks that are never evaluated create a data observability blind spot. Vigilmon closes that blind spot by monitoring your whylogs pipeline health, detecting drift beyond thresholds, and alerting on constraint violations.

What You'll Set Up

  • Cron heartbeat for profiling job health
  • Null rate drift alert per critical column
  • Numeric feature distribution drift monitor (PSI-based)
  • Schema violation detection
  • Data volume anomaly alert
  • Constraint violation rate monitoring

Prerequisites

  • whylogs 1.x Python library installed in your ML pipeline
  • A Python ML pipeline or batch scoring script that profiles data
  • A free Vigilmon account

Step 1: Monitor Profiling Job Health with a Cron Heartbeat

The most critical whylogs monitor is the simplest: confirm that profiling actually runs on schedule. A job that crashes during preprocessing, before calling why.log(), produces no profile and generates no alert — unless you're watching for its absence.

  1. In Vigilmon, click Add Monitor → Cron Heartbeat.
  2. Set the expected interval to match your batch pipeline schedule (e.g. 60 minutes for hourly scoring jobs).
  3. Copy the heartbeat URL.

Wrap your profiling call with a success signal:

import whylogs as why
import pandas as pd
import requests
import os

VIGILMON_HEARTBEAT = os.environ["VIGILMON_PROFILING_HEARTBEAT_URL"]

def profile_batch(df: pd.DataFrame, batch_id: str) -> None:
    """Profile inference batch and save results."""
    results = why.log(df)
    profile = results.profile()

    # Save profile to local storage
    profile_path = f"/var/whylogs/profiles/{batch_id}.bin"
    profile.write(path=profile_path)

    # Signal successful profiling to Vigilmon
    requests.get(VIGILMON_HEARTBEAT, timeout=5)

# In your pipeline
df = load_batch()  # your data loading logic
profile_batch(df, batch_id=f"batch_{datetime.utcnow().strftime('%Y%m%d_%H%M')}")

If the job crashes before profile_batch() completes — OOM error, upstream data load failure, disk full — the heartbeat never fires and Vigilmon alerts after one missed interval. Set Consecutive failures before alert to 1 for profiling heartbeats: a single missed profiling run is a genuine blind spot.

Step 2: Detect Null Rate Drift in ML Features

Null rates are the canary in the coal mine for upstream data pipeline failures. When an upstream ETL job starts producing nulls in a previously clean feature column, model predictions silently degrade — the model substitutes mean imputation values that were never seen in training. whylogs captures null counts per column; compare them against your baseline profile to detect drift.

import whylogs as why
import pandas as pd
import requests
import json
import os
from datetime import datetime

BASELINE_PROFILE_PATH = "/var/whylogs/reference/training_profile.bin"
VIGILMON_KEY = os.environ["VIGILMON_API_KEY"]
VIGILMON_NULL_MONITOR_ID = os.environ["VIGILMON_NULL_MONITOR_ID"]
NULL_DRIFT_THRESHOLD_PP = 5.0  # percentage points

def check_null_drift(current_df: pd.DataFrame) -> None:
    """Compare null rates in current batch vs. training baseline."""
    from whylogs.core import DatasetProfile

    # Profile current batch
    current_result = why.log(current_df)
    current_profile = current_result.profile()

    # Load baseline profile
    baseline_profile = DatasetProfile.read(BASELINE_PROFILE_PATH)

    alerts = []
    for col_name in current_df.columns:
        try:
            current_col = current_profile.get_column(col_name)
            baseline_col = baseline_profile.get_column(col_name)

            current_null_pct = (
                current_col.null_count / current_col.count * 100
                if current_col.count > 0 else 0
            )
            baseline_null_pct = (
                baseline_col.null_count / baseline_col.count * 100
                if baseline_col.count > 0 else 0
            )

            drift_pp = current_null_pct - baseline_null_pct
            if drift_pp > NULL_DRIFT_THRESHOLD_PP:
                alerts.append(
                    f"{col_name}: null rate {current_null_pct:.1f}% "
                    f"(baseline {baseline_null_pct:.1f}%, +{drift_pp:.1f}pp)"
                )
        except Exception:
            continue

    if alerts:
        requests.post(
            f"https://vigilmon.online/api/monitors/{VIGILMON_NULL_MONITOR_ID}/report",
            headers={"Authorization": f"Bearer {VIGILMON_KEY}"},
            json={
                "status": "down",
                "message": "Null rate drift detected:\n" + "\n".join(alerts),
            },
            timeout=10,
        )

Call check_null_drift() in your batch pipeline after loading the inference batch but before scoring. A 5 percentage point threshold catches meaningful upstream ETL failures without triggering on normal minor variations.

Step 3: Detect Numeric Feature Distribution Drift (PSI)

Population Stability Index (PSI) measures how much a numeric feature's distribution has shifted relative to a reference (typically your training data distribution). PSI > 0.2 indicates a significant distribution shift that may require model retraining.

whylogs computes the statistical summaries needed for PSI calculation. Here's a monitoring integration using whylogs' profile comparison:

import numpy as np
from whylogs.core import DatasetProfile

def compute_psi(baseline_buckets: list, current_buckets: list, epsilon: float = 1e-6) -> float:
    """Compute Population Stability Index between two distributions."""
    baseline_arr = np.array(baseline_buckets, dtype=float) + epsilon
    current_arr = np.array(current_buckets, dtype=float) + epsilon
    baseline_pct = baseline_arr / baseline_arr.sum()
    current_pct = current_arr / current_arr.sum()
    return float(np.sum((current_pct - baseline_pct) * np.log(current_pct / baseline_pct)))

def check_distribution_drift(current_df, baseline_profile_path: str) -> None:
    from whylogs.viz import NotebookProfileVisualizer
    from whylogs.viz.drift.column_drift_calculator import ColumnDriftCalculator

    baseline_profile = DatasetProfile.read(baseline_profile_path)
    current_result = why.log(current_df)
    current_profile = current_result.profile()

    VIGILMON_KEY = os.environ["VIGILMON_API_KEY"]
    DRIFT_MONITOR_ID = os.environ["VIGILMON_DRIFT_MONITOR_ID"]
    PSI_THRESHOLD = 0.2

    drift_alerts = []
    numeric_cols = current_df.select_dtypes(include="number").columns

    for col in numeric_cols:
        try:
            # Use whylogs drift calculator for histogram comparison
            calc = ColumnDriftCalculator()
            psi = calc.calculate_psi(
                baseline_profile.get_column(col),
                current_profile.get_column(col),
            )
            if psi > PSI_THRESHOLD:
                drift_alerts.append(f"{col}: PSI={psi:.3f} (threshold {PSI_THRESHOLD})")
        except Exception:
            continue

    if drift_alerts:
        requests.post(
            f"https://vigilmon.online/api/monitors/{DRIFT_MONITOR_ID}/report",
            headers={"Authorization": f"Bearer {VIGILMON_KEY}"},
            json={
                "status": "down",
                "message": "Feature distribution drift detected:\n" + "\n".join(drift_alerts),
            },
            timeout=10,
        )

Run after each inference batch completes. PSI > 0.1 is worth logging as a warning; PSI > 0.2 warrants investigation; PSI > 0.25 strongly suggests retraining is needed.

Step 4: Monitor Schema Violations

Schema violations — a new column appearing in inference data, an expected column disappearing, or a numeric column becoming categorical — cause silent model failures where the pipeline either drops the column (missing feature) or passes raw strings to a model expecting floats.

def check_schema(current_df: pd.DataFrame, expected_schema: dict) -> None:
    """
    expected_schema: {"column_name": "numeric" | "categorical" | "string"}
    """
    VIGILMON_KEY = os.environ["VIGILMON_API_KEY"]
    SCHEMA_MONITOR_ID = os.environ["VIGILMON_SCHEMA_MONITOR_ID"]

    violations = []

    # Check for missing columns
    for col, expected_type in expected_schema.items():
        if col not in current_df.columns:
            violations.append(f"Missing column: {col} (expected {expected_type})")
            continue

        # Check type mismatch
        actual_dtype = str(current_df[col].dtype)
        is_numeric = pd.api.types.is_numeric_dtype(current_df[col])
        is_categorical = pd.api.types.is_object_dtype(current_df[col])

        if expected_type == "numeric" and not is_numeric:
            violations.append(f"Type mismatch: {col} expected numeric, got {actual_dtype}")
        elif expected_type == "categorical" and not is_categorical:
            violations.append(f"Type mismatch: {col} expected categorical, got {actual_dtype}")

    # Check for unexpected new columns
    unexpected = [c for c in current_df.columns if c not in expected_schema]
    if unexpected:
        violations.append(f"Unexpected columns added: {', '.join(unexpected)}")

    if violations:
        requests.post(
            f"https://vigilmon.online/api/monitors/{SCHEMA_MONITOR_ID}/report",
            headers={"Authorization": f"Bearer {VIGILMON_KEY}"},
            json={
                "status": "down",
                "message": "Schema violations:\n" + "\n".join(violations),
            },
            timeout=10,
        )

# Define your expected schema at deployment time
EXPECTED_SCHEMA = {
    "user_age": "numeric",
    "purchase_count": "numeric",
    "product_category": "categorical",
    "session_duration_seconds": "numeric",
    "country_code": "categorical",
}

Schema violations are zero-tolerance: any unexpected column change should alert immediately. Set Consecutive failures before alert to 1 on the schema monitor.

Step 5: Monitor Data Volume Anomalies

A batch that arrives with 10% of its expected row count — because an upstream join produced a massive data loss — causes model predictions on too few samples and may indicate a serious upstream ETL failure. A batch that's 5× the expected size may indicate a duplicate ingestion issue.

def check_data_volume(current_df: pd.DataFrame) -> None:
    EXPECTED_ROW_COUNT = int(os.environ.get("EXPECTED_BATCH_SIZE", "10000"))
    LOWER_BOUND_PCT = 50   # alert if batch is < 50% of expected
    UPPER_BOUND_PCT = 200  # alert if batch is > 200% of expected

    VIGILMON_KEY = os.environ["VIGILMON_API_KEY"]
    VOLUME_MONITOR_ID = os.environ["VIGILMON_VOLUME_MONITOR_ID"]

    actual = len(current_df)
    lower = EXPECTED_ROW_COUNT * LOWER_BOUND_PCT // 100
    upper = EXPECTED_ROW_COUNT * UPPER_BOUND_PCT // 100

    if actual < lower or actual > upper:
        requests.post(
            f"https://vigilmon.online/api/monitors/{VOLUME_MONITOR_ID}/report",
            headers={"Authorization": f"Bearer {VIGILMON_KEY}"},
            json={
                "status": "down",
                "message": (
                    f"Batch volume anomaly: {actual} rows "
                    f"(expected {EXPECTED_ROW_COUNT}, "
                    f"bounds [{lower}, {upper}])"
                ),
            },
            timeout=10,
        )

Combine all checks into your pipeline's main entry point:

def run_inference_pipeline(batch_id: str) -> None:
    df = load_batch(batch_id)

    # Run all whylogs-based quality checks
    check_data_volume(df)
    check_schema(df, EXPECTED_SCHEMA)
    check_null_drift(df)
    check_distribution_drift(df, BASELINE_PROFILE_PATH)

    # Score only after quality gate passes
    predictions = model.predict(df)
    save_predictions(predictions, batch_id)

    # Signal successful pipeline completion
    requests.get(os.environ["VIGILMON_PROFILING_HEARTBEAT_URL"], timeout=5)

Step 6: Monitor whylogs Constraint Violations

whylogs Constraints define explicit data quality rules that your features must satisfy. Constraints are faster and more explainable than drift metrics for business-critical columns.

from whylogs.core.constraints import ConstraintsBuilder
from whylogs.core.constraints.factories import (
    not_null_percentage, is_non_negative, mean_between_range
)

def check_constraints(current_df: pd.DataFrame) -> None:
    VIGILMON_KEY = os.environ["VIGILMON_API_KEY"]
    CONSTRAINT_MONITOR_ID = os.environ["VIGILMON_CONSTRAINT_MONITOR_ID"]

    results = why.log(current_df)
    profile_view = results.view()

    builder = ConstraintsBuilder(dataset_profile_view=profile_view)

    # Define constraints for critical columns
    builder.add_constraint(not_null_percentage("user_id", min_perc=1.0))
    builder.add_constraint(is_non_negative("purchase_count"))
    builder.add_constraint(mean_between_range("session_duration_seconds", lower=10, upper=7200))

    constraints = builder.build()
    report = constraints.generate_constraints_report()

    failures = [r for r in report if not r.passed]
    if failures:
        failure_msgs = [
            f"{r.name}: {r.metric_value} (constraint: {r.constraint_str})"
            for r in failures
        ]
        requests.post(
            f"https://vigilmon.online/api/monitors/{CONSTRAINT_MONITOR_ID}/report",
            headers={"Authorization": f"Bearer {VIGILMON_KEY}"},
            json={
                "status": "down",
                "message": "whylogs constraint violations:\n" + "\n".join(failure_msgs),
            },
            timeout=10,
        )

Constraints are binary — pass or fail — which makes them excellent for critical invariants like "user_id must never be null" or "price must be non-negative". Drift metrics (PSI, Hellinger) are better for gradual distribution shifts. Use both.

Step 7: Configure Alert Channels and Thresholds

  1. Go to Alert Channels in Vigilmon and add Slack, email, or PagerDuty.
  2. Use a dedicated Slack channel for data quality alerts separate from infrastructure alerts — data scientists and ML engineers need to see these, not just SREs.
  3. Set per-monitor thresholds based on criticality:

| Monitor | Alert threshold | Consecutive failures | |---|---|---| | Profiling heartbeat | Missing 1 interval | 1 | | Schema violation | Any violation | 1 | | Null rate drift | >5pp from baseline | 1 | | Distribution drift | PSI >0.2 | 1 | | Data volume | <50% or >200% baseline | 1 | | Constraint violation | Any failure in critical columns | 1 |

All data quality alerts should use Consecutive failures = 1: a single batch with schema violations or extreme null rates is already a production incident, not a transient blip.

Summary

| Monitor | Target | What It Catches | |---|---|---| | Profiling heartbeat | Cron heartbeat after each job | Silent job crash, missing batch | | Null rate drift | Per-column null rate script | Upstream ETL nulling out features | | Distribution drift | PSI comparison script | Feature distribution shift pre-retraining | | Schema violation | Column type/presence check | Schema change breaking model input | | Data volume | Row count vs. expected range | Upstream join failure, deduplication bug | | Constraint violations | whylogs constraint report | Business rule violations in ML features |

whylogs gives your ML system a statistical mirror of every batch it processes — but that mirror is dark if the profiling job itself fails. With Vigilmon watching your profiling pipeline health alongside the drift and quality signals whylogs computes, you get the complete data observability picture: both "is the monitoring running?" and "what is it telling us?" covered in the same alerting system your infrastructure team already uses.

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