ETL Pipeline Monitoring: Freshness Checks, Row Reconciliation, and dbt Tests

ETL pipelines fail silently more often than they crash loudly. A broken pipeline that produces stale or wrong data is worse than one that fails fast. Here is how to build observability into your data pipelines.

What to Monitor

• Row counts: expected vs actual rows per run
• Freshness: max updated_at in destination — is data current?
• Null rates: unexpected nulls in critical columns
• Duplicate rates: duplicate keys in destination table
• Pipeline lag: time from source event to destination availability

Data Freshness Check (PostgreSQL)

-- Alert if no new rows in past 15 minutes
SELECT
  CASE
    WHEN max(created_at) < now() - INTERVAL '15 minutes'
    THEN 'STALE'
    ELSE 'FRESH'
  END AS freshness,
  max(created_at) AS latest_row,
  now() - max(created_at) AS lag
FROM orders;

Row Count Reconciliation

import psycopg2

def check_row_counts(source_conn, dest_conn, table, date):
    src = source_conn.execute(
        'SELECT count(*) FROM {} WHERE date = %s'.format(table), (date,)
    ).fetchone()[0]
    dst = dest_conn.execute(
        'SELECT count(*) FROM {} WHERE date = %s'.format(table), (date,)
    ).fetchone()[0]
    discrepancy = abs(src - dst)
    if discrepancy > 0:
        raise ValueError(f'{table}: source={src}, dest={dst}, diff={discrepancy}')
    return True

dbt Tests for Data Quality

# models/orders.yml
models:
  - name: orders
    columns:
      - name: id
        tests:
          - unique
          - not_null
      - name: status
        tests:
          - accepted_values:
              values: [pending, completed, cancelled]
      - name: amount
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0

Alerting on Pipeline Failure

import requests

def send_slack_alert(message: str, webhook_url: str):
    requests.post(webhook_url, json={'text': f':warning: ETL Alert: {message}'})

# In your pipeline runner:
try:
    run_pipeline()
except Exception as e:
    send_slack_alert(f'Pipeline failed: {e}', SLACK_WEBHOOK)
    raise

Define Freshness Before You Alert on It

Choose a timestamp that describes arrival in the platform, not only an event's business time. created_at can be old for a legitimate late-arriving record and can move forward even when most partitions are stuck. Record at least the source high-water mark, extraction time, destination load time, pipeline run ID, and the expected schedule or data interval. Express warn and error thresholds per source because a five-minute stream and a daily ledger do not share an SLA. Normalize timestamps to UTC and test how daylight-saving changes and upstream clock skew appear.

dbt source freshness compares the most recent loaded timestamp with the snapshot time. Current dbt configuration places freshness under config, with loaded_at_field moved under config in newer releases. Also note that dbt build does not itself run source freshness checks: schedule dbt source freshness explicitly, and decide whether stale input should stop downstream models or produce a separate alert.

sources:
  - name: commerce
    config:
      loaded_at_field: _etl_loaded_at
      freshness:
        warn_after: {count: 30, period: minute}
        error_after: {count: 60, period: minute}
    tables:
      - name: orders

Reconcile a Stable Boundary

Comparing two live tables at different moments creates false discrepancies. Freeze a logical interval such as source sequence 1001 through 2000, a closed event-time partition plus an allowed-lateness window, or a consistent database snapshot. Persist the lower and upper boundaries with the run. For PostgreSQL sources, a REPEATABLE READ transaction can provide one stable snapshot for multiple checks; use the equivalent snapshot or exported watermark for other systems. Do not build identifiers into SQL with string formatting from untrusted input.

Layer the Checks

Counts catch missing batches but not substitutions, duplicates, or wrong values. Reconcile in layers: total rows and distinct business keys; inserted, updated, and deleted operation counts; null and domain rates; sums over stable financial or quantity columns; and deterministic hashes over canonicalized fields within manageable partitions. Track expected late arrivals and tombstones explicitly. A source-to-destination difference should link to sampled failing keys and the exact boundary, not only a red dashboard. Compare against a documented tolerance when the source is eventually consistent rather than silently accepting any mismatch.

Make Failures Investigable

dbt data tests return failing rows and support severity, warn_if, and error_if thresholds. Use store_failures for high-value checks so responders can inspect the records, with retention and access controls because failure rows may contain sensitive data. Include test name, model version, invocation ID, compiled SQL reference, boundary, failure count, and artifact location in the alert. Deduplicate repeated alerts for the same run and page only when a human action is defined.

Recovery and Rollback

Design each load to be idempotent at a run ID or source offset. On failure, stop publishing the incomplete destination, retain the last known-good checkpoint, repair or replay into a staging table, rerun all checks, and swap or commit only after validation. If a bad model is already visible, rollback can mean restoring the prior table or view pointer while the pipeline catches up; deleting evidence and restarting from an assumed timestamp is not a rollback. Test gaps, duplicates, late updates, source deletes, schema changes, empty batches, and a retry after partial writes.

Operational Signals

Monitor freshness age, run duration, input and output rows, reconciliation difference, failed-test count, retry count, last successful boundary, and time since last successful publication. Give every alert an owner, dependency map, dashboard, and runbook. Review thresholds after schedule or volume changes. The objective is not a pipeline that is always green; it is one that fails visibly, preserves a trustworthy checkpoint, and can be replayed without creating a second incident.

Official References

• dbt source freshness execution
• dbt freshness configuration and loaded timestamps
• dbt test severity thresholds
• Persisting dbt test failures
• PostgreSQL transaction snapshots and isolation