Data Observability: Monitoring Freshness, Volume, and Distribution Before the User Does

Data observability is the discipline of knowing, at any moment, whether the data flowing through a platform is fresh, complete, and consistent with expectations, without waiting for a stakeholder to discover that it is not. The gap between "a problem exists in the data" and "someone notices the problem" is the observability gap. Shrinking that gap is the function of a monitoring layer built on the same principles as application monitoring: measure what matters, alert on thresholds, and provide enough context in the alert to take action without further investigation.

Freshness: has the data arrived?

Freshness monitoring checks whether the most recent data in a table or partition was loaded within the expected time window. The check runs on a schedule (every 15 or 30 minutes for hourly SLAs, every hour for daily ones) and fires an alert if the latest load timestamp is outside the SLA window.

import pandas as pd
from datetime import datetime, timedelta
from sqlalchemy import create_engine, text

FRESHNESS_SLA = {
    "fct_orders":       timedelta(hours=2),
    "fct_sessions":     timedelta(hours=1),
    "dim_products":     timedelta(hours=25),   # daily, with buffer
    "fct_payment_events": timedelta(minutes=30),
}

def check_freshness(engine) -> list[dict]:
    issues = []
    now = datetime.utcnow()

    for table, sla in FRESHNESS_SLA.items():
        result = engine.execute(text(f"SELECT MAX(loaded_at) AS latest FROM {table}")).fetchone()
        latest = result["latest"]

        if latest is None:
            issues.append({"table": table, "issue": "NO_DATA", "latest": None, "sla_hours": sla.total_seconds()/3600})
            continue

        age = now - latest
        if age > sla:
            issues.append({
                "table": table,
                "issue": "STALE",
                "latest": latest.isoformat(),
                "age_hours": round(age.total_seconds() / 3600, 2),
                "sla_hours": sla.total_seconds() / 3600,
            })

    return issues

Freshness checks are the highest-value observability investment for most platforms because freshness failures are the most common and most impactful data quality issue. A dashboard showing last week's data with no indication of staleness is not a data quality issue that requires statistical detection: it is a timestamp comparison that can be automated in a dozen lines of SQL.

Volume: is all the data there?

Volume monitoring checks whether the amount of data in a table or partition is consistent with historical expectations. A fresh table with half the expected rows is worse than a stale table: the stale table at least has the right data up to its last load. The half-full table has recent data that is incorrect in ways that may not be visible without knowing what the correct count should be.

def check_volume_anomalies(engine, table: str, date_col: str, lookback_days: int = 28) -> dict:
    query = text(f"""
        WITH daily_counts AS (
            SELECT
                DATE({date_col})     AS dt,
                COUNT(*)             AS row_count
            FROM {table}
            WHERE DATE({date_col}) >= CURRENT_DATE - {lookback_days + 1}
            GROUP BY 1
        ),
        baseline AS (
            SELECT
                dt,
                row_count,
                AVG(row_count) OVER (
                    ORDER BY dt
                    ROWS BETWEEN {lookback_days} PRECEDING AND 1 PRECEDING
                ) AS rolling_avg,
                STDDEV(row_count) OVER (
                    ORDER BY dt
                    ROWS BETWEEN {lookback_days} PRECEDING AND 1 PRECEDING
                ) AS rolling_std
            FROM daily_counts
        )
        SELECT *,
            CASE
                WHEN rolling_avg IS NULL THEN 'INSUFFICIENT_HISTORY'
                WHEN row_count < rolling_avg - 3 * rolling_std THEN 'ANOMALY_LOW'
                WHEN row_count > rolling_avg + 3 * rolling_std THEN 'ANOMALY_HIGH'
                ELSE 'OK'
            END AS status
        FROM baseline
        WHERE dt = CURRENT_DATE - 1
    """)
    return dict(engine.execute(query).fetchone())

The 3-sigma threshold (3 standard deviations from the rolling mean) produces a false positive rate of about 0.3% under a normal distribution: one spurious alert per 330 days for a daily check. In practice, volume distributions are right-skewed and seasonal, so the threshold should be calibrated against 90 days of historical data before deployment, and known exception dates (holidays, maintenance windows) should be excluded from the baseline.

Distribution monitoring: have the values changed?

Freshness and volume tell you whether the right amount of data arrived on time. Distribution monitoring tells you whether the data looks like the same kind of data it was before: the same column cardinalities, the same null rates, the same value ranges, the same category frequencies. A schema change from an upstream source will typically appear as a distribution change before it appears as a pipeline error.

def compute_column_profile(df: pd.DataFrame) -> dict:
    profile = {}
    for col in df.columns:
        null_rate = df[col].isnull().mean()
        col_profile = {"null_rate": round(null_rate, 4)}

        if pd.api.types.is_numeric_dtype(df[col]):
            col_profile.update({
                "mean":   round(df[col].mean(), 4),
                "std":    round(df[col].std(),  4),
                "p25":    df[col].quantile(0.25),
                "p75":    df[col].quantile(0.75),
                "p99":    df[col].quantile(0.99),
            })
        elif pd.api.types.is_object_dtype(df[col]):
            col_profile.update({
                "cardinality":    df[col].nunique(),
                "top_value":      df[col].value_counts().index[0] if null_rate < 1 else None,
                "top_value_freq": df[col].value_counts().iloc[0] / len(df) if null_rate < 1 else None,
            })

        profile[col] = col_profile
    return profile

def compare_profiles(baseline: dict, current: dict, null_rate_threshold: float = 0.05) -> list[str]:
    alerts = []
    for col in baseline:
        if col not in current:
            alerts.append(f"COLUMN_MISSING: {col} present in baseline, absent in current load")
            continue

        b_null = baseline[col]["null_rate"]
        c_null = current[col]["null_rate"]
        if abs(c_null - b_null) > null_rate_threshold:
            alerts.append(
                f"NULL_RATE_CHANGE: {col} null rate changed from {b_null:.1%} to {c_null:.1%}"
            )

    for col in current:
        if col not in baseline:
            alerts.append(f"NEW_COLUMN: {col} present in current load, absent in baseline")

    return alerts

The observability stack in practice

The three check types (freshness, volume, distribution) run on different schedules because they answer different questions. Freshness checks run frequently (every 15 to 30 minutes) because freshness failures need to be caught before the next scheduled report. Volume checks run once per load cycle because they compare today to the rolling baseline. Distribution checks run on a sample of each day's data as a post-load validation step.

Alerts from all three checks route to a single channel with consistent formatting: the table name, the check that failed, the measured value, the expected value, and the runbook link. An alert that requires opening a second tool to understand what it means is an alert that takes longer to resolve. Actionable alerts reduce the mean time to resolution, which is the metric that matters.