dbt Core in Practice: Data Modeling That Doesn't Rot

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# Install dbt Core with your warehouse adapter
pip install dbt-core dbt-postgres      # PostgreSQL / Redshift
pip install dbt-core dbt-bigquery      # Google BigQuery
pip install dbt-core dbt-snowflake     # Snowflake
pip install dbt-core dbt-duckdb        # DuckDB (great for local dev)
pip install dbt-core dbt-spark         # Apache Spark
pip install dbt-core dbt-databricks    # Databricks

# Or install everything for a multi-warehouse setup
pip install dbt-core dbt-postgres dbt-bigquery

# Verify
dbt --version
# Core: 1.8.x
# Installed adapters: postgres, bigquery

1
2

dbt init analytics
cd analytics

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# ~/.dbt/profiles.yml
analytics:
  target: dev
  outputs:
    dev:
      type: postgres
      host: localhost
      port: 5432
      user: "{{ env_var('DBT_USER', 'analytics') }}"
      password: "{{ env_var('DBT_PASSWORD') }}"
      database: analytics
      schema: dbt_dev_yourname   # Each dev gets their own schema
      threads: 4

    prod:
      type: postgres
      host: prod-warehouse.company.com
      port: 5432
      user: dbt_prod
      password: "{{ env_var('DBT_PROD_PASSWORD') }}"
      database: analytics
      schema: analytics
      threads: 8

1
2
3
4
5

# Test the connection
dbt debug

# Run with the prod target
dbt run --target prod

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

# dbt_project.yml
name: analytics
version: '1.0.0'
config-version: 2

profile: analytics

# Directory structure
model-paths: ["models"]
test-paths: ["tests"]
macro-paths: ["macros"]
seed-paths: ["seeds"]
snapshot-paths: ["snapshots"]

# Target directory for compiled SQL
target-path: "target"
clean-targets: ["target", "dbt_packages"]

# Default materializations per folder
models:
  analytics:
    # staging/ models: cheap views, rebuilt each run
    staging:
      +materialized: view
      +schema: staging

    # intermediate/ models: views for complex joins
    intermediate:
      +materialized: view
      +schema: intermediate

    # marts/ models: tables that dashboards query
    marts:
      +materialized: table
      +schema: marts
      core:
        +materialized: table
      finance:
        +materialized: table

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

-- models/staging/stripe/stg_stripe__payments.sql
-- One staging model per source table.
-- Naming: stg_{source}__{table}

WITH source AS (
    -- Reference the raw source table, not a hardcoded name
    SELECT * FROM {{ source('stripe', 'payments') }}
),

renamed AS (
    SELECT
        -- Rename to our conventions
        id                          AS payment_id,
        created                     AS created_at,
        amount                      AS amount_cents,

        -- Cast and clean
        amount / 100.0              AS amount_dollars,
        LOWER(currency)             AS currency,
        LOWER(status)               AS payment_status,

        -- Extract useful derived columns
        customer_id,
        charge_id,
        invoice_id,

        -- Metadata
        _airbyte_ab_id              AS _loaded_at
    FROM source
)

SELECT * FROM renamed

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

-- models/staging/postgres/stg_postgres__orders.sql
WITH source AS (
    SELECT * FROM {{ source('postgres', 'orders') }}
),

renamed AS (
    SELECT
        order_id,
        customer_id,
        status                          AS order_status,
        CAST(ordered_at AS TIMESTAMP)   AS ordered_at,
        CAST(fulfilled_at AS TIMESTAMP) AS fulfilled_at,
        subtotal_cents / 100.0          AS subtotal,
        tax_cents / 100.0               AS tax,
        shipping_cents / 100.0          AS shipping,
        (subtotal_cents + tax_cents + shipping_cents) / 100.0 AS total_amount
    FROM source
    WHERE order_id IS NOT NULL  -- Filter obvious garbage at ingest
)

SELECT * FROM renamed

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

# models/staging/stripe/_stripe__sources.yml
version: 2

sources:
  - name: stripe
    description: "Stripe payment data loaded by Fivetran"
    database: raw_data
    schema: stripe
    loader: fivetran

    # Check that raw data is arriving on time
    freshness:
      warn_after: {count: 6, period: hour}
      error_after: {count: 24, period: hour}
    loaded_at_field: _fivetran_synced

    tables:
      - name: payments
        description: "All payment attempts, successful and failed"
        columns:
          - name: id
            description: "Stripe payment intent ID"
            tests:
              - unique
              - not_null
          - name: status
            tests:
              - accepted_values:
                  values: ['succeeded', 'pending', 'failed', 'canceled']
          - name: amount
            description: "Amount in smallest currency unit (cents for USD)"
            tests:
              - not_null
              - dbt_utils.accepted_range:
                  min_value: 0

      - name: customers
        description: "Stripe customer records"
        freshness:
          warn_after: {count: 12, period: hour}
        columns:
          - name: id
            tests: [unique, not_null]

1
2

# Check source freshness (runs before dbt run in production)
dbt source freshness

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

-- models/intermediate/int_orders__with_payments.sql
-- Joins orders with their payment records.
-- Used by multiple mart models — defined once here.

WITH orders AS (
    SELECT * FROM {{ ref('stg_postgres__orders') }}
),

payments AS (
    SELECT * FROM {{ ref('stg_stripe__payments') }}
    WHERE payment_status = 'succeeded'
),

order_payments AS (
    SELECT
        o.order_id,
        o.customer_id,
        o.order_status,
        o.ordered_at,
        o.fulfilled_at,
        o.subtotal,
        o.tax,
        o.shipping,
        o.total_amount,

        -- Payment details
        p.payment_id,
        p.amount_dollars       AS amount_paid,
        p.currency,
        p.created_at           AS paid_at,

        -- Derived
        CASE
            WHEN p.payment_id IS NULL THEN FALSE
            ELSE TRUE
        END                    AS is_paid,

        DATEDIFF('day', o.ordered_at, o.fulfilled_at) AS days_to_fulfill

    FROM orders o
    LEFT JOIN payments p ON o.order_id = p.invoice_id
)

SELECT * FROM order_payments

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

-- models/marts/core/dim_customers.sql
-- Customer dimension table: one row per customer, current state.

WITH customers AS (
    SELECT * FROM {{ ref('stg_postgres__customers') }}
),

orders AS (
    SELECT * FROM {{ ref('int_orders__with_payments') }}
),

customer_orders AS (
    SELECT
        customer_id,
        COUNT(*)                                AS total_orders,
        SUM(total_amount)                       AS lifetime_value,
        MIN(ordered_at)                         AS first_order_at,
        MAX(ordered_at)                         AS most_recent_order_at,
        COUNT(CASE WHEN is_paid THEN 1 END)     AS paid_orders,
        AVG(total_amount)                       AS avg_order_value
    FROM orders
    GROUP BY customer_id
),

final AS (
    SELECT
        c.customer_id,
        c.email,
        c.first_name,
        c.last_name,
        c.created_at                            AS customer_since,
        c.country,
        c.plan_type,

        -- Aggregated order behavior
        COALESCE(co.total_orders, 0)            AS total_orders,
        COALESCE(co.lifetime_value, 0)          AS lifetime_value,
        COALESCE(co.paid_orders, 0)             AS paid_orders,
        co.first_order_at,
        co.most_recent_order_at,
        co.avg_order_value,

        -- Segmentation
        CASE
            WHEN co.total_orders IS NULL THEN 'prospect'
            WHEN co.total_orders = 1 THEN 'new'
            WHEN co.most_recent_order_at >= CURRENT_DATE - INTERVAL '90 days'
                THEN 'active'
            ELSE 'lapsed'
        END                                     AS customer_segment,

        -- Metadata
        CURRENT_TIMESTAMP                       AS dbt_updated_at
    FROM customers c
    LEFT JOIN customer_orders co USING (customer_id)
)

SELECT * FROM final

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

-- models/marts/finance/fct_revenue.sql
-- Revenue fact table: one row per transaction for finance reporting.

WITH order_payments AS (
    SELECT * FROM {{ ref('int_orders__with_payments') }}
),

final AS (
    SELECT
        payment_id,
        order_id,
        customer_id,

        -- Time dimensions
        paid_at,
        DATE_TRUNC('month', paid_at)        AS revenue_month,
        DATE_TRUNC('week', paid_at)         AS revenue_week,
        EXTRACT(YEAR FROM paid_at)          AS revenue_year,

        -- Revenue breakdown
        subtotal                            AS gross_revenue,
        tax,
        shipping,
        total_amount,
        amount_paid,

        -- Flags
        currency,
        order_status,
        days_to_fulfill
    FROM order_payments
    WHERE is_paid
)

SELECT * FROM final

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

# models/marts/core/_core__models.yml
version: 2

models:
  - name: dim_customers
    description: "One row per customer with aggregated order behavior"
    columns:
      - name: customer_id
        description: "Primary key — Postgres customer ID"
        tests:
          - unique
          - not_null

      - name: email
        tests:
          - unique
          - not_null
          - dbt_utils.expression_is_true:
              expression: "email LIKE '%@%'"

      - name: customer_segment
        tests:
          - not_null
          - accepted_values:
              values: ['prospect', 'new', 'active', 'lapsed']

      - name: lifetime_value
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0

  - name: fct_revenue
    description: "One row per successful payment transaction"
    columns:
      - name: payment_id
        tests:
          - unique
          - not_null

      - name: total_amount
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0

    # Model-level tests
    tests:
      # No revenue rows without a matching customer
      - dbt_utils.expression_is_true:
          expression: "total_orders >= paid_orders"
          model: ref('dim_customers')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

-- tests/assert_revenue_matches_orders.sql
-- Revenue fact table total should match order totals within 1% tolerance.
-- Fails if the query returns any rows.

WITH revenue_total AS (
    SELECT SUM(total_amount) AS total FROM {{ ref('fct_revenue') }}
),

order_total AS (
    SELECT SUM(total_amount) AS total
    FROM {{ ref('int_orders__with_payments') }}
    WHERE is_paid
),

comparison AS (
    SELECT
        r.total AS revenue_total,
        o.total AS order_total,
        ABS(r.total - o.total) / NULLIF(o.total, 0) AS discrepancy_pct
    FROM revenue_total r
    CROSS JOIN order_total o
)

-- Return rows when the discrepancy exceeds 1%
SELECT *
FROM comparison
WHERE discrepancy_pct > 0.01

1
2
3
4
5
6

-- tests/assert_no_future_orders.sql
-- Orders should not have ordered_at in the future.

SELECT order_id, ordered_at
FROM {{ ref('stg_postgres__orders') }}
WHERE ordered_at > CURRENT_TIMESTAMP

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# Run all tests
dbt test

# Test a specific model
dbt test --select dim_customers

# Test a model and all its upstream dependencies
dbt test --select +dim_customers

# Run only schema tests (not singular tests)
dbt test --select "test_type:generic"

# Check source freshness + run tests in one command
dbt source freshness && dbt test

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

-- models/marts/core/fct_events.sql
-- Event tracking table: potentially hundreds of millions of rows.
-- Only process new events on each run.

{{
    config(
        materialized='incremental',
        unique_key='event_id',
        incremental_strategy='merge',   -- 'append', 'merge', 'delete+insert'
        on_schema_change='append_new_columns'
    )
}}

WITH events AS (
    SELECT * FROM {{ source('segment', 'events') }}
    {% if is_incremental() %}
    -- On incremental runs, only process new events
    -- The filter uses the max value already in the table
    WHERE received_at > (SELECT MAX(received_at) FROM {{ this }})
    {% endif %}
),

final AS (
    SELECT
        id              AS event_id,
        anonymous_id,
        user_id,
        event           AS event_name,
        received_at,
        sent_at,
        properties,
        context_page_url,
        context_page_title
    FROM events
    WHERE id IS NOT NULL
)

SELECT * FROM final

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

-- append: only add new rows (fastest, no deduplication)
{{ config(materialized='incremental', incremental_strategy='append') }}

-- merge: upsert based on unique_key (handles late-arriving updates)
{{ config(
    materialized='incremental',
    unique_key='event_id',
    incremental_strategy='merge'
) }}

-- delete+insert: delete matching partition then insert (good for date-partitioned tables)
{{ config(
    materialized='incremental',
    unique_key=['date_day', 'user_id'],
    incremental_strategy='delete+insert',
    partition_by={'field': 'date_day', 'data_type': 'date'}
) }}

1
2
3
4
5

# Rebuild a single incremental model from scratch
dbt run --select fct_events --full-refresh

# Rebuild all incremental models
dbt run --full-refresh

1
2
3
4

-- macros/cents_to_dollars.sql
{% macro cents_to_dollars(column_name, precision=2) %}
    ROUND({{ column_name }} / 100.0, {{ precision }})
{% endmacro %}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

-- macros/date_spine.sql
-- Generate a date range table for calendar joins
{% macro generate_date_spine(start_date, end_date) %}
    SELECT
        CAST(date_day AS DATE) AS date_day,
        EXTRACT(YEAR FROM date_day) AS year,
        EXTRACT(MONTH FROM date_day) AS month,
        EXTRACT(DOW FROM date_day) AS day_of_week,
        date_day >= CURRENT_DATE AS is_future
    FROM (
        SELECT
            CAST('{{ start_date }}' AS DATE) + CAST(generate_series AS INTEGER) AS date_day
        FROM generate_series(
            0,
            DATEDIFF('day',
                CAST('{{ start_date }}' AS DATE),
                CAST('{{ end_date }}' AS DATE)
            )
        )
    ) dates
{% endmacro %}

1
2
3
4
5
6

-- Using macros in models
SELECT
    order_id,
    {{ cents_to_dollars('subtotal_cents') }} AS subtotal,
    {{ cents_to_dollars('tax_cents') }} AS tax
FROM {{ ref('stg_postgres__orders') }}

1
2
3
4
5
6

-- macros/test_not_negative.sql
{% test not_negative(model, column_name) %}
    SELECT {{ column_name }}
    FROM {{ model }}
    WHERE {{ column_name }} < 0
{% endtest %}

1
2
3
4
5

# Use it anywhere
columns:
  - name: total_amount
    tests:
      - not_negative

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# packages.yml
packages:
  - package: dbt-labs/dbt_utils
    version: 1.2.0
  - package: calogica/dbt_expectations
    version: 0.10.3
  - package: dbt-labs/audit_helper
    version: 0.9.0
  - package: brooklyn-data/dbt_artifacts
    version: 2.9.0

1

dbt deps  # Install packages

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# models/marts/core/_core__models.yml
version: 2

models:
  - name: dim_customers
    description: >
      Customer dimension table. One row per customer.
      Includes aggregated lifetime order behavior.
      Used by: Revenue dashboards, Customer success Looker explores.

    meta:
      owner: "@data-team"
      contains_pii: true
      sla: "refreshed every 6 hours"

    columns:
      - name: customer_id
        description: "Primary key. Integer ID from the Postgres customers table."

      - name: customer_segment
        description: >
          Behavioral segment based on order recency.
          - prospect: no orders yet
          - new: exactly 1 order
          - active: ordered in the last 90 days
          - lapsed: ordered before, not in 90 days

      - name: lifetime_value
        description: "Sum of all successful order totals in USD."

1
2
3
4
5
6
7
8
9

# Generate and serve docs locally
dbt docs generate
dbt docs serve  # Opens http://localhost:8080

# The docs site includes:
# - Lineage DAG (visual dependency graph)
# - Model descriptions and column docs
# - Source freshness status
# - Test results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

-- snapshots/snap_customers.sql
{% snapshot snap_customers %}

{{
    config(
        target_schema='snapshots',
        unique_key='customer_id',
        strategy='timestamp',       -- or 'check' for tables without updated_at
        updated_at='updated_at',
        invalidate_hard_deletes=True
    )
}}

SELECT * FROM {{ source('postgres', 'customers') }}

{% endsnapshot %}

1
2

# Run snapshots (separate from dbt run)
dbt snapshot

1
2
3
4
5
6

-- seeds/country_codes.csv
country_code,country_name,region
US,United States,North America
GB,United Kingdom,Europe
DE,Germany,Europe
JP,Japan,Asia Pacific

1

dbt seed  # Loads CSV files as tables in the warehouse

1
2
3

SELECT o.*, cc.region
FROM {{ ref('stg_postgres__orders') }} o
LEFT JOIN {{ ref('country_codes') }} cc ON o.country_code = cc.country_code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

# .github/workflows/dbt-ci.yml
name: dbt CI

on:
  pull_request:
    paths:
      - 'models/**'
      - 'tests/**'
      - 'macros/**'

jobs:
  dbt-ci:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
          cache: pip

      - name: Install dbt
        run: pip install dbt-core dbt-postgres

      - name: Install packages
        run: dbt deps

      - name: Check source freshness
        run: dbt source freshness
        env:
          DBT_PASSWORD: ${{ secrets.DBT_CI_PASSWORD }}

      - name: Run modified models + downstream
        run: |
          # Run only models changed in this PR plus their dependencies
          dbt run --select state:modified+ \
            --defer \
            --state ./target
        env:
          DBT_PASSWORD: ${{ secrets.DBT_CI_PASSWORD }}

      - name: Test modified models + downstream
        run: |
          dbt test --select state:modified+
        env:
          DBT_PASSWORD: ${{ secrets.DBT_CI_PASSWORD }}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

# dags/dbt_production.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.external_task import ExternalTaskSensor

DBT_DIR = "/opt/airflow/dbt/analytics"
DBT_PROFILE = "analytics"
DBT_TARGET = "prod"

default_args = {
    "owner": "data-engineering",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "on_failure_callback": slack_alert,   # Your alert function
}

with DAG(
    dag_id="dbt_production",
    default_args=default_args,
    schedule_interval="0 */6 * * *",   # Every 6 hours
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["dbt", "production"],
) as dag:

    # Wait for data loaders to finish
    wait_for_fivetran = ExternalTaskSensor(
        task_id="wait_for_fivetran",
        external_dag_id="fivetran_sync",
        external_task_id="fivetran_complete",
        timeout=3600,
        poke_interval=60,
    )

    check_sources = BashOperator(
        task_id="check_source_freshness",
        bash_command=f"""
            cd {DBT_DIR}
            dbt source freshness \
              --profiles-dir ~/.dbt \
              --profile {DBT_PROFILE} \
              --target {DBT_TARGET}
        """,
    )

    run_staging = BashOperator(
        task_id="run_staging_models",
        bash_command=f"""
            cd {DBT_DIR}
            dbt run \
              --select staging \
              --profiles-dir ~/.dbt \
              --profile {DBT_PROFILE} \
              --target {DBT_TARGET} \
              --threads 8
        """,
    )

    test_staging = BashOperator(
        task_id="test_staging_models",
        bash_command=f"""
            cd {DBT_DIR}
            dbt test \
              --select staging \
              --profiles-dir ~/.dbt \
              --profile {DBT_PROFILE} \
              --target {DBT_TARGET}
        """,
    )

    run_marts = BashOperator(
        task_id="run_mart_models",
        bash_command=f"""
            cd {DBT_DIR}
            dbt run \
              --select marts \
              --profiles-dir ~/.dbt \
              --profile {DBT_PROFILE} \
              --target {DBT_TARGET} \
              --threads 8
        """,
    )

    test_marts = BashOperator(
        task_id="test_mart_models",
        bash_command=f"""
            cd {DBT_DIR}
            dbt test \
              --select marts \
              --threads 4
        """,
    )

    generate_docs = BashOperator(
        task_id="generate_docs",
        bash_command=f"""
            cd {DBT_DIR}
            dbt docs generate
            aws s3 sync ./target s3://company-dbt-docs/latest/
        """,
    )

    # DAG order
    (
        wait_for_fivetran
        >> check_sources
        >> run_staging
        >> test_staging
        >> run_marts
        >> test_marts
        >> generate_docs
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

# ── Running Models ──────────────────────────────────────────────────────────

# Run all models
dbt run

# Run a specific model
dbt run --select dim_customers

# Run a model and all its dependencies (upstream)
dbt run --select +dim_customers

# Run a model and all models that depend on it (downstream)
dbt run --select dim_customers+

# Run a model and everything: upstream and downstream
dbt run --select +dim_customers+

# Run only staging models
dbt run --select staging

# Run everything in a directory
dbt run --select models/marts/finance

# Run models changed since last run (great for CI)
dbt run --select state:modified+

# Full refresh of incremental models
dbt run --select fct_events --full-refresh

# ── Testing ─────────────────────────────────────────────────────────────────

dbt test                          # All tests
dbt test --select dim_customers   # Tests for one model
dbt test --select +dim_customers  # Tests for model + upstream
dbt test --select "tag:critical"  # Tests tagged as critical

# ── Other Commands ───────────────────────────────────────────────────────────

dbt compile                       # Compile SQL without running
dbt source freshness              # Check source table age
dbt docs generate && dbt docs serve  # Build + open documentation
dbt snapshot                      # Run snapshot models
dbt seed                          # Load CSV seeds
dbt deps                          # Install packages
dbt clean                         # Delete target/ and dbt_packages/
dbt debug                         # Test connection and configuration
dbt list                          # List all models in project
dbt list --select +dim_customers  # List models in the DAG subtree

1
2
3
4

models:
  - name: fct_revenue
    config:
      tags: ['critical', 'finance']

1

dbt test --select "tag:critical"  # Run critical tests first

1
2
3
4
5
6

SELECT
    {{ dbt_utils.generate_surrogate_key(['order_id', 'product_id']) }} AS order_line_key,
    order_id,
    product_id,
    quantity
FROM {{ ref('stg_postgres__order_lines') }}