Data Observability Metrics Pack

Replace {{PLACEHOLDERS}} with your environment values, then deploy to your stack.

-- =============================================================================
-- DATA OBSERVABILITY METRICS PACK (SQL)
-- Run against your warehouse to score freshness, volume, schema, and quality.
-- Replace {{PLACEHOLDERS}} and schedule as a daily observability job.
-- =============================================================================

-- ── CONFIGURATION ────────────────────────────────────────────────────────────
-- Target table: {{DATABASE}}.{{SCHEMA}}.{{TABLE}}
-- PK column:     {{PK_COLUMN}}
-- Date column:   {{DATE_COLUMN}}
-- Loaded at col: {{LOADED_AT_COLUMN}}

WITH config AS (
    SELECT
        '{{DATABASE}}.{{SCHEMA}}.{{TABLE}}' AS full_table_name,
        '{{PK_COLUMN}}'                     AS pk_column,
        '{{DATE_COLUMN}}'                   AS date_column,
        '{{LOADED_AT_COLUMN}}'              AS loaded_at_column,
        24                                  AS freshness_sla_hours,
        0.20                                AS volume_anomaly_pct,   -- 20% swing triggers alert
        7                                   AS lookback_days
),

-- ── 1. FRESHNESS METRIC ─────────────────────────────────────────────────────
freshness AS (
    SELECT
        'freshness' AS metric_name,
        DATEDIFF('hour', MAX({{LOADED_AT_COLUMN}}), CURRENT_TIMESTAMP()) AS observed_value,
        c.freshness_sla_hours AS threshold,
        CASE
            WHEN DATEDIFF('hour', MAX({{LOADED_AT_COLUMN}}), CURRENT_TIMESTAMP()) <= c.freshness_sla_hours
            THEN 1 ELSE 0
        END AS score
    FROM {{DATABASE}}.{{SCHEMA}}.{{TABLE}}, config c
),

-- ── 2. VOLUME METRIC (today vs 7-day avg) ─────────────────────────────────────
daily_volume AS (
    SELECT
        DATE({{DATE_COLUMN}}) AS dt,
        COUNT(*) AS row_count
    FROM {{DATABASE}}.{{SCHEMA}}.{{TABLE}}
    WHERE {{DATE_COLUMN}} >= DATEADD('day', -7, CURRENT_DATE())
    GROUP BY 1
),

volume AS (
    SELECT
        'volume_anomaly' AS metric_name,
        ABS(today.row_count - avg_hist.avg_rows) / NULLIF(avg_hist.avg_rows, 0) AS observed_value,
        c.volume_anomaly_pct AS threshold,
        CASE
            WHEN ABS(today.row_count - avg_hist.avg_rows) / NULLIF(avg_hist.avg_rows, 0) <= c.volume_anomaly_pct
            THEN 1 ELSE 0
        END AS score
    FROM config c
    CROSS JOIN (SELECT row_count FROM daily_volume WHERE dt = CURRENT_DATE()) today
    CROSS JOIN (SELECT AVG(row_count) AS avg_rows FROM daily_volume WHERE dt < CURRENT_DATE()) avg_hist
),

-- ── 3. QUALITY: PK NULL RATE ──────────────────────────────────────────────────
quality AS (
    SELECT
        'pk_null_rate' AS metric_name,
        SUM(CASE WHEN {{PK_COLUMN}} IS NULL THEN 1 ELSE 0 END)::FLOAT / NULLIF(COUNT(*), 0) AS observed_value,
        0.001 AS threshold,
        CASE
            WHEN SUM(CASE WHEN {{PK_COLUMN}} IS NULL THEN 1 ELSE 0 END)::FLOAT / NULLIF(COUNT(*), 0) <= 0.001
            THEN 1 ELSE 0
        END AS score
    FROM {{DATABASE}}.{{SCHEMA}}.{{TABLE}}
    WHERE {{DATE_COLUMN}} >= DATEADD('day', -1, CURRENT_DATE())
),

-- ── 4. COMPOSITE OBSERVABILITY SCORECARD ─────────────────────────────────────
scorecard AS (
    SELECT * FROM freshness
    UNION ALL SELECT * FROM volume
    UNION ALL SELECT * FROM quality
)

// ... download full template for remaining code