Transform Data in the Warehouse with dbt

Overview

dbt (data build tool) is a SQL-first transformation framework that turns your data warehouse into a version-controlled analytics engineering platform. You write SELECT statements in .sql files, and dbt handles materialization (table, view, incremental), dependency resolution, testing, and documentation. dbt compiles your models into SQL, runs them in the warehouse (BigQuery, Snowflake, Redshift, Postgres), and tracks lineage between models.

When to Use

• Transforming raw data in a warehouse into analytics-ready tables
• When you want version-controlled, testable SQL transformations
• Building a layered model architecture (staging → intermediate → marts)
• Teams where analysts write SQL but need software engineering practices
• When you need data lineage, documentation, and freshness checks

When NOT to Use

• Real-time/streaming transformations — dbt is batch-oriented
• Small datasets in pandas — use pandas/Polars directly
• ETL where extraction and loading are the bottleneck — dbt only does the T in ETL
• When you need complex procedural logic (loops, conditionals) — use stored procedures or Python. If extraction and loading dominate and the T is trivial, a pandas ETL pipeline is a lighter fit.

Explanation

How a dbt run works

dbt run follows a fixed sequence:

1. Parse and compile — dbt reads every .sql file, resolves the Jinja ({{ ref() }}, {{ source() }}, {{ config() }}) and produces the raw SQL your warehouse executes.
2. Build the DAG — source() nodes are external tables, ref() edges are model dependencies. dbt orders execution from that DAG: a model never runs before its parents finish.
3. Materialize — each node runs its compiled SQL as CREATE VIEW, CREATE TABLE or MERGE depending on its materialization, in the target schema.

Every ref() call is an edge in that graph — which is why models reference model names, not table names. If stg_customers fails, fct_orders and dim_customers are skipped instead of being built on broken data, and dbt run reports the whole subtree as skipped.

Choosing a materialization

Materialization	Creates	Cost per run	Use for
view	A view	~0 (no data written)	Staging; models consumed through other models
table	A physical table	Full rebuild every run	Marts queried directly and often
incremental	Table + merge/append of new rows	Only new or changed data	Large facts with a reliable unique_key
ephemeral	Nothing — inlined as a CTE downstream	Compile-time only	Intermediate steps used by one or two models

The layout used in this recipe — staging as views, intermediate as ephemeral, marts as tables with the big fact incremental — maps each layer to the cheapest materialization that still does its job.

What dbt test actually runs

Schema tests compile to SQL assertions: unique becomes a GROUP BY ... HAVING COUNT(*) > 1 query, not_null a WHERE col IS NULL scan, relationships an anti-join against the referenced model. A test that returns rows fails. Since dbt 1.2, dbt build runs models and their tests together in DAG order — a model that fails its tests blocks its downstream children, which is exactly the behavior you want in CI. For checks beyond schema tests (distributions, ranges, business rules), a Python-side validator like the Pandera recipe complements dbt tests.

Solution

Project setup

pip install dbt-core dbt-postgres   # adapter per warehouse: dbt-snowflake, dbt-bigquery...
dbt init my_project                  # creates dbt_project.yml and the folder skeleton

profiles.yml (at ~/.dbt/ by default) holds the connection — it’s credentials, so it stays out of version control:

# ~/.dbt/profiles.yml
my_project:
  target: dev            # default environment
  outputs:
    dev:
      type: postgres
      host: localhost
      port: 5432
      user: analytics
      pass: "{{ env_var('DB_PASSWORD') }}"
      dbname: warehouse
      schema: dbt_dev
      threads: 4

Project structure

dbt_project/
├── dbt_project.yml
├── profiles.yml
├── models/
│   ├── staging/
│   │   ├── stg_orders.sql
│   │   ├── stg_customers.sql
│   │   └── schema.yml
│   ├── intermediate/
│   │   └── int_orders_enriched.sql
│   └── marts/
│       ├── fct_orders.sql
│       ├── dim_customers.sql
│       └── schema.yml
├── macros/
│   └── cents_to_dollars.sql
├── tests/
│   └── assert_order_status.sql
└── snapshots/
    └── snap_customers.sql

Basic model

-- models/staging/stg_orders.sql
SELECT
    order_id::integer AS order_id,
    customer_id::integer AS customer_id,
    order_date::date AS order_date,
    amount::numeric(10,2) AS amount,
    status::varchar AS status
FROM {{ source('raw', 'orders') }}

Source definitions

# models/staging/schema.yml
version: 2

sources:
  - name: raw
    database: warehouse
    schema: raw_data
    freshness:
      warn_after: {count: 12, period: hour}
      error_after: {count: 24, period: hour}
    loaded_at_field: _loaded_at
    tables:
      - name: orders
        columns:
          - name: order_id
            tests:
              - unique
              - not_null
          - name: amount
            tests:
              - not_null
      - name: customers
        columns:
          - name: customer_id
            tests:
              - unique
              - not_null

Model with references

-- models/marts/fct_orders.sql
SELECT
    o.order_id,
    o.customer_id,
    o.order_date,
    o.amount,
    o.status,
    c.customer_name,
    c.customer_tier,
    c.city
FROM {{ ref('stg_orders') }} o
LEFT JOIN {{ ref('stg_customers') }} c
    ON o.customer_id = c.customer_id
WHERE o.status = 'completed'

ref() creates a dependency — dbt runs stg_orders and stg_customers before fct_orders. The freshness block on the source pairs with loaded_at_field so dbt source freshness warns when the upstream load is late.

Materializations

# dbt_project.yml
models:
  my_project:
    staging:
      +materialized: view
    intermediate:
      +materialized: ephemeral
    marts:
      +materialized: table
      fct_orders:
        +materialized: incremental

-- models/marts/fct_orders.sql — incremental model
{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge'
) }}

SELECT
    o.order_id,
    o.customer_id,
    o.order_date,
    o.amount
FROM {{ ref('stg_orders') }} o
WHERE o.status = 'completed'

{% if is_incremental() %}
    AND o.order_date > (SELECT MAX(order_date) FROM {{ this }})
{% endif %}

Schema tests

# models/marts/schema.yml
version: 2

models:
  - name: fct_orders
    description: "Completed orders with customer details"
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: customer_id
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: amount
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: "amount >= 0"
      - name: status
        tests:
          - accepted_values:
              values: ['completed', 'pending', 'cancelled']

Custom singular test

-- tests/assert_no_negative_amounts.sql
SELECT *
FROM {{ ref('fct_orders') }}
WHERE amount < 0

Macros

-- macros/cents_to_dollars.sql
{% macro cents_to_dollars(column_name) %}
    ({{ column_name }} / 100.0)
{% endmacro %}

-- Using the macro in a model
SELECT
    order_id,
    {{ cents_to_dollars('amount_cents') }} AS amount_dollars
FROM {{ ref('stg_orders') }}

Snapshots for SCD Type 2

-- snapshots/snap_customers.sql
{% snapshot snap_customers %}
{{
    config(
      target_schema='snapshots',
      unique_key='customer_id',
      strategy='timestamp',
      updated_at='updated_at',
    )
}}
SELECT * FROM {{ source('raw', 'customers') }}
{% endsnapshot %}

Snapshots track changes to source data over time, creating a history table with valid_from, valid_to, and dbt_scd_id columns.

Using variables

-- models/marts/fct_orders.sql
SELECT *
FROM {{ ref('stg_orders') }}
WHERE order_date >= '{{ var('start_date', '2025-01-01') }}'

dbt run --vars '{"start_date": "2025-06-01"}'

Hooks

# dbt_project.yml
on-run-start:
  - "CREATE SCHEMA IF NOT EXISTS {{ target.schema }}"
on-run-end:
  - "{{ log('Model run complete', info=True) }}"

Variants

The full dbt project — all models, tests, snapshot, macro, profiles.yml template, and the Airflow DAG — is in the companion repository.

Using dbt with Airflow

A single dbt build task covers run + test in DAG order (dbt ≥1.2). If you need separate run/test tasks for alerting, keep them as two BashOperators:

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

dag = DAG("dbt_pipeline", schedule="@daily", start_date=datetime(2025, 1, 1))

dbt_build = BashOperator(
    task_id="dbt_build",
    bash_command="cd /opt/dbt && dbt build --select marts.*",
    dag=dag,
)

For richer integration — per-model tasks, retries per model, partial parsing — look at astronomer-cosmos, the Airflow provider that renders the dbt DAG inside Airflow.

Custom materialization with post-hook

{{ config(
    materialized='table',
    post_hook="CREATE INDEX IF NOT EXISTS idx_orders_date ON {{ this }} (order_date)"
) }}

SELECT * FROM {{ ref('stg_orders') }}

Packages

# packages.yml
packages:
  - package: dbt-labs/dbt_utils
    version: 1.1.1
  - package: calogica/dbt_expectations
    version: 0.10.3

dbt deps

-- Using dbt_utils
{{ dbt_utils.date_spine(
    datepart="day",
    start_date="'2025-01-01'",
    end_date="'2025-12-31'"
) }}

Best Practices

• Use a layered architecture: staging (cleaning) → intermediate (joins) → marts (business logic)
• Always test primary keys with unique and not_null — catches data quality issues early
• Use ephemeral materialization for intermediate models used only once — avoids storage
• Use incremental for large fact tables — only process new rows, not the full history
• Document every model and column — dbt docs generate creates a documentation site
• Use ref() instead of hardcoding table names — dbt resolves dependencies and order
• Use sources for raw data and ref() for dbt models — separates external from internal
• Run dbt test (or dbt build) in CI — fail the pipeline on data quality issues
• When the pipeline grows beyond dbt run, schedule and monitor the DAG with Airflow

Common Mistakes

• Not testing models: without tests, bad data propagates silently. Always test primary keys and critical fields.
• Using table materialization for everything: views are cheaper for models queried infrequently. Use table only for models queried often.
• Not using incremental for large tables: full refresh on a 100M row table is slow. Use incremental with a unique_key.
• Hardcoding schema names: use {{ target.schema }} or {{ this }} — that lets you deploy to dev/staging/prod without touching the models.
• Not using packages: dbt_utils and dbt_expectations provide tested macros for common patterns.

See Also

• dbt Core documentation — the official docs for dbt Core 1.x: models, Jinja context, and the CLI.
• Materializations — how view, table, incremental, and ephemeral behave, and custom materializations.
• Sources and ref — declaring sources, freshness checks, and how ref() builds the DAG.
• dbt_utils and dbt_expectations — macro packages for common tests and transformations.
• dbt build — run models, tests, snapshots, and seeds in DAG order.