Data Observability Audit

Most teams should target Level 2-3.

Audit Questions

Answer these to assess your current state:

Coverage

1. What % of dbt models have at least one test?
2. What % of mart models have primary key tests?
3. What % of source tables have freshness checks?
4. Do you monitor row counts over time?
5. Do you monitor null rates per column?
6. Do you detect schema drift (source column changes)?

Detection Time

7. How long after a data failure do you typically know about it?
8. Do failures come from automated alerts or user complaints?
9. Do you have SLAs defined and tracked?

Lineage & Root Cause

10. Can you quickly identify which dashboards are affected by a model failure?
11. Can you trace an anomaly back to its source system within 15 minutes?

Audit Your Current dbt Tests

# Count test coverage by model
dbt ls --select "resource_type:test" --output json \
| python3 -c "
import json, sys, collections
counts = collections.defaultdict(int)
for line in sys.stdin:
    try:
        obj = json.loads(line)
        parent = obj.get('depends_on', {}).get('nodes', [''])[0]
        model = parent.replace('model.my_project.', '')
        counts[model] += 1
    except:
        pass
for model, count in sorted(counts.items(), key=lambda x: x[1]):
    print(f'{count:3d} tests: {model}')
"

# Models with zero tests (critical gap)
dbt ls --select "resource_type:model" --output json \
| python3 -c "
import json, sys
for line in sys.stdin:
    try:
        obj = json.loads(line)
        if not obj.get('config', {}).get('contract', {}).get('enforced') \
           and 'marts' in obj.get('fqn', []):
            print(obj['name'], '— no contract enforced')
    except:
        pass
"

Observability Gap Analysis

Layer 1: Schema Checks (Foundation)

What to check:

• Every mart model has unique + not_null on primary key
• Every FK column has a relationships test
• Enum/categorical columns have accepted_values tests
• Critical amount columns have accepted_range tests

Quick audit SQL (Snowflake):

-- Models missing primary key tests
-- (Run in your warehouse against information_schema)
select
    table_name,
    column_name
from information_schema.columns
where table_schema = 'MARTS'
  and column_name like '%_id'
  and column_name not in (
      -- List your tested PK columns here
      select column_name from your_dbt_test_results
      where test_type = 'unique'
  )

Layer 2: Freshness Checks

What to check:

• All sources in sources.yml have freshness + loaded_at_field
• dbt source freshness runs in CI/CD
• Freshness alerts route to on-call channel

Fix: