Data Observability — Monitoring Your Data, Not Just Pipes

Your Airflow DAGs are green. Your dbt models compiled and ran without errors. CPU utilization is normal. Memory is fine. And yet — the revenue dashboard is showing numbers that are 40% lower than yesterday. Nothing in your infrastructure monitoring caught it because the infrastructure is working perfectly. The data is broken.

This is the gap that data observability fills. Traditional monitoring watches whether your pipelines run. Data observability watches whether the data itself is correct, fresh, and complete.

What data observability is

Think of it this way: application observability (Datadog, New Relic, Grafana) monitors the health of your systems. Data observability monitors the health of what flows through those systems.

A pipeline can succeed technically (DAG completes, model builds, no errors) while producing completely wrong results. An API might start returning null values for a field that used to be populated. A source table might stop receiving rows silently. A schema migration might rename a column upstream without telling anyone.

Data observability is the practice of continuously monitoring your data for these invisible failures — catching them before a stakeholder does.

The five pillars

1. Freshness

Is the data up-to-date? If your daily pipeline was supposed to run at 6 AM and it is now noon with no new data, something is wrong.

-- Freshness check: when was the latest record loaded?
SELECT
    MAX(_loaded_at) AS latest_load,
    DATEDIFF(hour, MAX(_loaded_at), CURRENT_TIMESTAMP) AS hours_since_load,
    CASE
        WHEN DATEDIFF(hour, MAX(_loaded_at), CURRENT_TIMESTAMP) > 24
        THEN 'CRITICAL'
        WHEN DATEDIFF(hour, MAX(_loaded_at), CURRENT_TIMESTAMP) > 12
        THEN 'WARNING'
        ELSE 'OK'
    END AS freshness_status
FROM raw.orders;

In dbt, freshness is a first-class concept:

sources:
  - name: raw
    tables:
      - name: orders
        loaded_at_field: _etl_loaded_at
        freshness:
          warn_after: { count: 12, period: hour }
          error_after: { count: 24, period: hour }

Run dbt source freshness to check all sources at once.

2. Volume

Is the expected amount of data present? A sudden drop or spike in row counts signals a problem.

def check_volume_anomaly(table, partition_col='created_date'):
    """Compare today's volume against historical baseline."""
    query = f"""
    WITH daily_counts AS (
        SELECT
            {partition_col} AS dt,
            COUNT(*) AS row_count
        FROM {table}
        WHERE {partition_col} >= CURRENT_DATE - 30
        GROUP BY {partition_col}
    ),
    stats AS (
        SELECT
            AVG(row_count) AS avg_count,
            STDDEV(row_count) AS stddev_count
        FROM daily_counts
        WHERE dt < CURRENT_DATE  -- Exclude today from baseline
    )
    SELECT
        dc.row_count AS today_count,
        s.avg_count,
        s.stddev_count,
        (dc.row_count - s.avg_count) / NULLIF(s.stddev_count, 0) AS z_score
    FROM daily_counts dc
    CROSS JOIN stats s
    WHERE dc.dt = CURRENT_DATE
    """
    result = execute_query(query)

    if abs(result['z_score']) > 3:
        alert(
            f"Volume anomaly in {table}: "
            f"{result['today_count']} rows today vs "
            f"{result['avg_count']:.0f} avg (z-score: {result['z_score']:.2f})"
        )

A z-score greater than 3 means today’s volume is more than 3 standard deviations from the mean — a strong signal that something unusual happened.

3. Schema

Did the structure of the data change? Column additions are usually fine. Column deletions, renames, or type changes break downstream models.

def check_schema_changes(table, expected_schema):
    """Detect schema drift by comparing current schema to baseline."""
    current_schema = get_table_schema(table)  # Returns {col: type}

    added = set(current_schema.keys()) - set(expected_schema.keys())
    removed = set(expected_schema.keys()) - set(current_schema.keys())
    type_changes = {
        col: (expected_schema[col], current_schema[col])
        for col in set(expected_schema.keys()) & set(current_schema.keys())
        if expected_schema[col] != current_schema[col]
    }

    issues = []
    if removed:
        issues.append(f"Columns removed: {removed}")  # CRITICAL
    if type_changes:
        issues.append(f"Type changes: {type_changes}")  # WARNING
    if added:
        issues.append(f"Columns added: {added}")  # INFO

    if removed or type_changes:
        alert(f"Schema change in {table}: {'; '.join(issues)}")

4. Distribution

Are the values within expected ranges? This catches the subtle bugs that other checks miss.

-- Distribution check: detect anomalies in order amounts
WITH daily_stats AS (
    SELECT
        DATE(ordered_at) AS order_date,
        AVG(total_amount) AS avg_amount,
        PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY total_amount) AS median_amount,
        PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY total_amount) AS p99_amount,
        COUNT(CASE WHEN total_amount = 0 THEN 1 END)::FLOAT / COUNT(*) AS zero_pct,
        COUNT(CASE WHEN total_amount IS NULL THEN 1 END)::FLOAT / COUNT(*) AS null_pct
    FROM fct_orders
    WHERE DATE(ordered_at) >= CURRENT_DATE - 7
    GROUP BY DATE(ordered_at)
)
SELECT *
FROM daily_stats
WHERE avg_amount < (SELECT AVG(avg_amount) * 0.5 FROM daily_stats WHERE order_date < CURRENT_DATE)
   OR null_pct > 0.05
   OR zero_pct > 0.10;

A common real-world scenario: a currency conversion API goes down, and all international orders are stored with amount = 0. Row counts look normal. Freshness is fine. But the distribution of amounts has shifted dramatically. Distribution checks catch this.

5. Lineage

Where did this data come from, and what does it feed? Lineage is the map that connects sources to transformations to outputs.

Source: Stripe API → raw.payments
    ↓  Airflow DAG: ingest_stripe
Staging: stg_payments (cleaned, typed)
    ↓  dbt model
Intermediate: int_payments_enriched (joined with customers)
    ↓  dbt model
Mart: fct_revenue (aggregated, final)
    ↓  BI tool
Dashboard: "Revenue by Region" (Looker)
    ↓  Alert
Slack: #revenue-alerts channel

When the revenue dashboard shows wrong numbers, lineage tells you: check fct_revenue, which comes from int_payments_enriched, which comes from stg_payments, which comes from the Stripe API ingest. You trace upstream until you find the break.

Building observability with dbt + Elementary

Elementary is an open-source dbt package that adds automated observability to your dbt project. It runs anomaly detection on volume, freshness, and distributions — no separate infrastructure required.

Setup

# Install the dbt package
# packages.yml
packages:
  - package: elementary-data/elementary
    version: [">=0.13.0", "<0.14.0"]

dbt deps
dbt run --select elementary

Configure monitors

# models/marts/_marts.yml
models:
  - name: fct_orders
    meta:
      owner: "data-team@company.com"
    tests:
      # Volume anomaly detection
      - elementary.volume_anomalies:
          timestamp_column: ordered_at
          where: "status != 'cancelled'"

      # Freshness monitoring
      - elementary.freshness_anomalies:
          timestamp_column: ordered_at

      # Column-level distribution monitoring
    columns:
      - name: total_amount
        tests:
          - elementary.column_anomalies:
              timestamp_column: ordered_at
              column_anomalies:
                - zero_count
                - null_count
                - average
                - standard_deviation

Elementary stores historical metrics in your warehouse and uses statistical methods to detect when current values deviate from the baseline. No manual threshold configuration needed.

Alerts

# Send alerts to Slack
pip install elementary-data[slack]
edr monitor --slack-webhook $SLACK_WEBHOOK --slack-channel data-alerts

Anomaly detection methods

Z-score method

The simplest approach. Calculate the z-score (how many standard deviations from the mean) and alert when it exceeds a threshold.

import numpy as np

def z_score_anomaly(values, current_value, threshold=3):
    """Flag if current value is more than threshold std devs from mean."""
    mean = np.mean(values)
    std = np.std(values)

    if std == 0:
        return current_value != mean

    z = abs(current_value - mean) / std
    return z > threshold

Limitation: Assumes normal distribution. Fails with seasonal data or trends.

Moving average with bounds

Better for data with trends:

def moving_average_anomaly(values, current_value, window=14, multiplier=2.5):
    """Detect anomalies using rolling average and standard deviation."""
    recent = values[-window:]
    rolling_mean = np.mean(recent)
    rolling_std = np.std(recent)

    lower = rolling_mean - (multiplier * rolling_std)
    upper = rolling_mean + (multiplier * rolling_std)

    return current_value < lower or current_value > upper

Seasonal decomposition

For data with weekly or monthly patterns (e-commerce with weekend spikes):

from statsmodels.tsa.seasonal import seasonal_decompose

def seasonal_anomaly(time_series, current_value, period=7):
    """Account for seasonality when detecting anomalies."""
    decomposition = seasonal_decompose(
        time_series, model='multiplicative', period=period
    )

    # Get the residual (what's left after trend and season)
    residuals = decomposition.resid.dropna()
    residual_mean = residuals.mean()
    residual_std = residuals.std()

    # Check if current residual is anomalous
    expected = decomposition.trend.iloc[-1] * decomposition.seasonal.iloc[-1]
    actual_residual = current_value / expected

    z = abs(actual_residual - residual_mean) / residual_std
    return z > 3

Incident response for data quality issues

When a data issue is detected, follow a structured response:

1. Detect and classify

SEVERITY_LEVELS = {
    'P0': 'Revenue/financial data affected, executive dashboards wrong',
    'P1': 'Major business metrics affected, multiple teams impacted',
    'P2': 'Single team affected, non-critical metrics',
    'P3': 'Minor data quality degradation, cosmetic issues',
}

2. Communicate immediately

Do not wait until you have a root cause. Stakeholders need to know their data might be unreliable:

TEMPLATE: Data Incident Notification
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Severity: P1
Affected: Revenue dashboard, fct_orders table
Detected: 2024-01-15 08:30 UTC
Status: Investigating

What happened: Revenue figures for Jan 14 are ~40% lower
than expected. Volume checks flagged abnormal row count in
stg_payments.

Impact: Revenue reporting for Jan 14 may be inaccurate.
Do NOT use Jan 14 figures for decisions.

Next update: 09:00 UTC
Owner: @data-team-oncall

3. Trace lineage to root cause

Follow the lineage upstream from the affected output:

Dashboard (wrong) → fct_revenue (wrong) → stg_payments (missing rows)
→ Stripe API ingest (API timeout at 02:00 UTC, partial load)

4. Fix and validate

# Re-run the failed ingest for the affected period
rerun_ingest(source='stripe', start='2024-01-14', end='2024-01-15')

# Rebuild affected models
# dbt run --select stg_payments+ (stg_payments and all downstream)

# Validate the fix
assert_revenue_within_range('2024-01-14', expected_min=900000, expected_max=1200000)

5. Post-incident review

Document what happened, why monitoring did not catch it sooner, and what to improve. Add new monitors for the failure mode.

Tools comparison

Tool	Type	Pricing	Best for
Elementary	Open-source dbt package	Free	dbt-centric teams, budget-conscious
Monte Carlo	SaaS platform	$$$	Enterprise, multi-tool environments
Great Expectations	Open-source Python	Free	Python-heavy pipelines, custom checks
Soda	YAML-based checks	Free + paid	Multi-platform, simple setup
dbt tests	Built-in dbt	Free	Basic quality gates, schema validation
Datafold	SaaS + open-source	$	Diff testing, CI/CD integration

Decision framework

Start with dbt tests if you use dbt. They are free, built-in, and cover the basics (unique, not_null, relationships, accepted_values).

Add Elementary when you need anomaly detection without managing infrastructure. It runs inside your existing dbt project and warehouse.

Consider Monte Carlo when you have 100+ data sources, multiple transformation tools, and need automated lineage across the entire stack. It is expensive but comprehensive.

Use Great Expectations if your pipelines are Python-based (not dbt) and you need custom validation logic.

Try Soda if you want a simple YAML-based approach that works across different databases and tools.

Next steps

• Data Quality & Governance — the quality framework that observability monitors.
• dbt Fundamentals — the tool that integrates most tightly with observability.
• Pipeline Testing — proactive testing vs reactive observability.
• Pipeline CI/CD — integrate observability checks into your deployment pipeline.