03 - dbt: Observability

10 min read

dbt: Observability

Quote

“Monitoring is TDD for production. Observability is debugging for production — give Future You the power to answer any question.”

Source: Charity Majors | charity.wtf (2018)

dbt Artifacts Overview

FileGenerated byKey Contents
manifest.jsondbt compile, dbt run, dbt docs generateFull DAG: nodes, sources, exposures, SQL, metadata
run_results.jsondbt run, dbt test, dbt buildPer-node execution time, status, adapter response
catalog.jsondbt docs generateColumn types and row counts from warehouse info schema
sources.jsondbt source freshnessSource freshness check results

All files land in the target/ directory. In CI/CD they are uploaded to GCS for retention and cross-job sharing.

dbt Artifacts — Key Fields in run_results.json

{
  "metadata": {"dbt_version": "1.8.3", "generated_at": "2026-03-23T05:12:44Z"},
  "results": [
    {
      "unique_id": "model.financial_indices.fct_esg_scores",
      "status": "success",
      "execution_time": 47.3,
      "adapter_response": {"rows_affected": 182400},
      "failures": null,
      "message": "CREATE TABLE (182400 rows, 2.1 GB processed)"
    },
    {
      "unique_id": "test.financial_indices.not_null_fct_esg_scores_issuer_id",
      "status": "fail",
      "execution_time": 3.1,
      "failures": 12,
      "message": "Got 12 results, configured to fail if != 0"
    }
  ]
}

Python Script: Push dbt Results to DogStatsD

Parse run_results.json after each dbt run and emit custom metrics to the Datadog Agent’s DogStatsD UDP endpoint.

#!/usr/bin/env python3
"""post_run_metrics.py — emit dbt run results as Datadog custom metrics."""
 
import json
import sys
import time
from datadog import initialize, statsd
 
RUN_RESULTS_PATH = sys.argv[1] if len(sys.argv) > 1 else "target/run_results.json"
DD_HOST = "localhost"
DD_PORT = 8125
 
initialize(statsd_host=DD_HOST, statsd_port=DD_PORT)
 
with open(RUN_RESULTS_PATH) as f:
    run_results = json.load(f)
 
generated_at = run_results["metadata"]["generated_at"]
dbt_version = run_results["metadata"]["dbt_version"]
 
for result in run_results["results"]:
    unique_id = result["unique_id"]
    node_type = unique_id.split(".")[0]   # "model", "test", "snapshot"
    node_name = unique_id.split(".")[-1]
    status = result["status"]             # success | error | fail | warn | skipped
    execution_time = result.get("execution_time", 0)
    failures = result.get("failures") or 0
 
    base_tags = [
        f"node_type:{node_type}",
        f"node_name:{node_name}",
        f"status:{status}",
        f"dbt_version:{dbt_version}",
        "pipeline:esg_transformer",
    ]
 
    # Execution duration per model
    statsd.gauge(
        "dbt.model.execution_time_seconds",
        execution_time,
        tags=base_tags,
    )
 
    # Test failure count (0 = passing)
    if node_type == "test":
        statsd.gauge(
            "dbt.test.failure_count",
            failures,
            tags=base_tags,
        )
 
    # Increment success/failure counters
    if status == "success":
        statsd.increment("dbt.node.success", tags=base_tags)
    elif status in ("error", "fail"):
        statsd.increment("dbt.node.failure", tags=base_tags)
 
# Emit overall pipeline status
all_statuses = {r["status"] for r in run_results["results"]}
pipeline_success = 1 if all_statuses <= {"success", "warn", "skipped"} else 0
statsd.gauge("dbt.pipeline.success", pipeline_success, tags=["pipeline:esg_transformer"])
 
print(f"Emitted metrics for {len(run_results['results'])} nodes.")

Invoke in the Airflow DAG after dbt completes:

from airflow.operators.bash import BashOperator
 
emit_metrics = BashOperator(
    task_id="emit_dbt_metrics",
    bash_command="python /opt/scripts/post_run_metrics.py /opt/dbt/financial_indices/target/run_results.json",
    trigger_rule="all_done",   # run even if dbt failed
)

elementary Package

elementary-data is an open-source dbt package that adds:

  • Anomaly detection on metric values, row counts, null rates, and schema.
  • A metadata schema inside your warehouse for storing test results.
  • A CLI report (edr) for browsing historical test results.

Installation

# packages.yml
packages:
  - package: elementary-data/elementary
    version: [">=0.14.0", "<0.15.0"]
# dbt_project.yml — elementary schema in its own dataset
models:
  elementary:
    +schema: elementary
    +materialized: table
dbt deps && dbt run --select elementary

Anomaly Detection Tests

# models/marts/esg/schema.yml
models:
  - name: fct_esg_scores
    columns:
      - name: environmental_score
        tests:
          - elementary.column_anomalies:
              column_anomalies:
                - null_count
                - null_percent
                - average
                - standard_deviation
              timestamp_column: score_date
              days_back: 30
              time_bucket:
                period: day
                count: 1
 
  - name: fct_index_composition
    tests:
      - elementary.table_anomalies:
          table_anomalies:
            - row_count
            - freshness
          timestamp_column: composition_date
          days_back: 14

Schema Change Tracking

models:
  - name: stg_esg_provider_raw
    tests:
      - elementary.schema_changes
      - elementary.schema_changes_from_baseline:
          fail_on_added: false
          fail_on_removed: true

schema_changes_from_baseline is critical for ESG provider feeds: if a provider silently drops a column (e.g., governance_controversy_score), the test fails immediately rather than silently propagating nulls into benchmark calculations.

elementary CLI Report

pip install elementary-data[bigquery]
edr report --project-dir /opt/dbt/financial_indices --days-back 7

Generates an HTML report with test result history, anomaly trend charts, and model run durations.

Monitoring dbt in Datadog

dbt Datadog Monitoring — Custom Metric Naming Convention

dbt.model.execution_time_seconds   (gauge)   — tagged: node_name, status
dbt.test.failure_count             (gauge)   — tagged: node_name, test_type
dbt.node.success                   (count)   — tagged: node_type
dbt.node.failure                   (count)   — tagged: node_type
dbt.pipeline.success               (gauge)   — 1 = success, 0 = failure
dbt.source.freshness_minutes_lag   (gauge)   — from sources.json

dbt Datadog Monitoring — Source Freshness Metric

import json
from datadog import statsd
 
with open("target/sources.json") as f:
    sources = json.load(f)
 
for source_name, result in sources.get("sources", {}).items():
    max_loaded = result.get("max_loaded_at")
    if max_loaded:
        import datetime, pytz
        loaded_dt = datetime.datetime.fromisoformat(max_loaded.replace("Z", "+00:00"))
        lag_minutes = (datetime.datetime.now(pytz.utc) - loaded_dt).total_seconds() / 60
        statsd.gauge(
            "dbt.source.freshness_minutes_lag",
            lag_minutes,
            tags=[f"source:{source_name}", "pipeline:esg_transformer"],
        )

Alerting: Airflow Callback → Datadog → Slack

dbt Alerting — Airflow Failure Callback to Datadog

import os
import requests
 
def notify_datadog_and_slack(context: dict):
    task_id = context["task_instance"].task_id
    dag_id = context["dag"].dag_id
    run_id = context["run_id"]
    log_url = context["task_instance"].log_url
 
    # Post event to Datadog
    requests.post(
        "https://api.datadoghq.eu/api/v1/events",
        headers={"DD-API-KEY": os.environ["DD_API_KEY"]},
        json={
            "title": f"dbt failure: {dag_id}.{task_id}",
            "text": f"Run ID: {run_id}\nLogs: {log_url}",
            "alert_type": "error",
            "tags": ["pipeline:esg_transformer", "source:airflow"],
        },
    )
 
    # Post to Slack via webhook
    requests.post(
        os.environ["SLACK_WEBHOOK_URL"],
        json={
            "text": (
                f":x: *dbt failure* in `{dag_id}.{task_id}`\n"
                f"Run: `{run_id}`\n"
                f"<{log_url}|View logs>"
            )
        },
    )
 
default_args = {"on_failure_callback": notify_datadog_and_slack}

Datadog Monitor for Pipeline Failure

Create a Datadog monitor on dbt.pipeline.success that pages the on-call engineer if the metric drops to 0 for two consecutive checks:

{
  "name": "ESG dbt pipeline failed",
  "type": "metric alert",
  "query": "min(last_2):avg:dbt.pipeline.success{pipeline:esg_transformer} < 1",
  "message": "@pagerduty-data-oncall Pipeline failure detected. Check Airflow for details.",
  "thresholds": {"critical": 1}
}

Dashboard Template: Model Durations, Test Failures, Freshness

Widgets to Include

WidgetMetric / SourceVisualisation
Pipeline success ratedbt.pipeline.successSLO widget (28-day window)
Top 10 slowest models (today)dbt.model.execution_time_secondsHorizontal bar, grouped by node_name
Test failure count by modeldbt.test.failure_countHeat map — model × day
Source freshness lagdbt.source.freshness_minutes_lagTime series — threshold line at SLA
Daily node success/failuredbt.node.success, dbt.node.failureStacked bar chart
Elementary anomalieselementary schema in BQBigQuery widget / iframe
Recent pipeline eventsDatadog EventsEvent stream widget

Terraform Snippet (Datadog Dashboard)

resource "datadog_dashboard" "dbt_observability" {
  title       = "dbt — ESG Pipeline Observability"
  layout_type = "ordered"
 
  widget {
    timeseries_definition {
      title = "Model Execution Time (top 15)"
      request {
        q            = "top(avg:dbt.model.execution_time_seconds{pipeline:esg_transformer} by {node_name}, 15, 'mean', 'desc')"
        display_type = "bars"
      }
    }
  }
 
  widget {
    query_value_definition {
      title = "Pipeline Success (today)"
      request {
        q          = "min:dbt.pipeline.success{pipeline:esg_transformer}"
        aggregator = "min"
      }
      precision = 0
    }
  }
}

Graph View

Backlinks