dbt Cheat Sheet

-- models/staging/stg_orders.sql
select
  id as order_id,
  customer_id,
  amount,
  ordered_at
from {{ source('app', 'orders') }}

models/staging/        stg_<source>__<entity>   one source, light renames
models/intermediate/  int_<entity>             joins and business rules
models/marts/         fct_ and dim_            one row per business grain
# staging stays close to the source; marts are what people query

dbt run --select stg_orders+
dbt compile
# + is this model and everything downstream
# compiled SQL lands in target/; the warehouse only sees that SQL
# dbt Core runs locally; dbt Cloud runs the same project in a job

version: 2

sources:
  - name: app
    database: raw
    schema: public
    tables:
      - name: orders
      - name: customers

-- models/marts/fct_orders.sql
select order_id, customer_id, amount
from {{ ref('stg_orders') }}
where amount > 0
# source() and ref() are the edges; parents run first
# do not hard-code schema.table

sources:
  - name: app
    freshness:
      warn_after: {count: 12, period: hour}
      error_after: {count: 24, period: hour}
    loaded_at_field: updated_at
    tables:
      - name: orders
# dbt source freshness

view         the query runs on every read
table        stored rows; full rebuild each run
incremental  stored rows; only new or changed keys
ephemeral    no relation; inlined as a CTE

{{ config(materialized='view') }}

select * from {{ ref('stg_orders') }}
# a staging model is usually a view; a mart is usually a table

# dbt_project.yml
models:
  analytics:
    staging:
      +materialized: view
    marts:
      +materialized: table

{{ config(materialized='ephemeral') }}

select * from {{ ref('stg_orders') }} where amount > 0
# downstream ref() inlines this SELECT; nothing is created

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge'
) }}

select * from {{ ref('stg_orders') }}
{% if is_incremental() %}
where updated_at > (select coalesce(max(updated_at), '1970-01-01') from {{ this }})
{% endif %}

merge          update on unique_key, insert the rest
delete+insert  delete matching keys, then insert
append         insert only; duplicates are your problem
# the first run builds the full table; later runs enter is_incremental()

dbt run --full-refresh --select fct_orders
# use this when the logic changes, not on every schedule
# a row older than max(updated_at) is missed until the filter is wider

version: 2

models:
  - name: stg_orders
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: status
        tests:
          - accepted_values:
              values: ['placed', 'paid', 'cancelled']

- name: customer_id
  tests:
    - relationships:
        to: ref('stg_customers')
        field: customer_id
# zero rows means pass; dbt build --select stg_orders runs the model and the tests

-- tests/assert_positive_amounts.sql
select order_id
from {{ ref('stg_orders') }}
where amount < 0
# any returned row fails the test

tests:
  - not_null:
      config:
        severity: error
  - unique:
      config:
        severity: warn
# warn lets the job continue; error stops it
# test keys in staging and business rules closer to marts

version: 2

models:
  - name: fct_orders
    description: One row per order.
    columns:
      - name: amount
        description: "{{ doc('order_amount') }}"

-- models/docs.md
{% docs order_amount %}
Order total in major currency units, after discounts.
{% enddocs %}

dbt docs generate
dbt docs serve
# the site shows descriptions and the DAG
# write the description when you add the column

select
  order_id,
  {% if var('include_tax', false) %}
  tax_amount,
  {% endif %}
  amount
from {{ ref('stg_orders') }}
where ordered_at >= '{{ var("start_date") }}'

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

select {{ cents_to_dollars('amount_cents') }} as amount
from {{ ref('stg_payments') }}

{% set statuses = ['placed', 'paid'] %}
select * from {{ ref('stg_orders') }}
where status in (
  {% for status in statuses %}
    '{{ status }}'{% if not loop.last %},{% endif %}
  {% endfor %}
)
# stop when the SQL is harder to read than a plain query

# packages.yml
packages:
  - package: dbt-labs/dbt_utils
    version: [">=1.1.0", "<2.0.0"]
# dbt deps

{{ dbt_utils.generate_surrogate_key(['order_id', 'line_id']) }} as order_line_sk

# schema.yml
tests:
  - dbt_utils.unique_combination_of_columns:
      combination_of_columns: [order_id, line_id]

# seeds/country_codes.csv
country_code,country_name
IN,India
US,United States
# dbt seed loads a table you can ref('country_codes')
# seeds are for small reference data, not a fact table

{% snapshot customers_snapshot %}
{{ config(
    target_schema='snapshots',
    unique_key='customer_id',
    strategy='timestamp',
    updated_at='updated_at'
) }}
select * from {{ source('app', 'customers') }}
{% endsnapshot %}
# dbt snapshot

{{ config(
    strategy='check',
    unique_key='customer_id',
    check_cols=['email', 'status']
) }}
# use check when there is no reliable updated_at
# dbt closes the old row and inserts a new one when a checked column changes

select *
from {{ ref('customers_snapshot') }}
where dbt_valid_to is null
# dbt_valid_from and dbt_valid_to bound each version
# dbt_scd_id identifies the version; this is SCD type 2

# profiles.yml — dbt Core reads this; dbt Cloud stores the connection on the job
analytics:
  target: dev
  outputs:
    dev:
      type: snowflake
      account: "{{ env_var('SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('SNOWFLAKE_USER') }}"
      private_key_path: "{{ env_var('SNOWFLAKE_KEY_PATH') }}"
      role: transformer
      database: analytics_dev
      warehouse: dev_wh
      schema: "dbt_{{ env_var('DBT_USER') }}"
      threads: 4
    prod:
      type: snowflake
      account: "{{ env_var('SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('SNOWFLAKE_USER') }}"
      private_key_path: "{{ env_var('SNOWFLAKE_KEY_PATH') }}"
      role: transformer
      database: analytics
      warehouse: transform_wh
      schema: marts
      threads: 8

# dbt_project.yml
vars:
  start_date: "2024-01-01"
# one run
dbt run --vars '{start_date: 2024-06-01}'

{% if target.name == 'dev' %}
  {{ config(materialized='view') }}
{% else %}
  {{ config(materialized='table') }}
{% endif %}
# dbt run --target prod

{{ config(post_hook="grant select on {{ this }} to role analyst") }}

# dbt_project.yml
on-run-end:
  - "grant usage on schema analytics.marts to role analyst"

dbt build --select state:modified+ --defer --state ./prod-artifacts
# --defer reuses unchanged prod relations instead of rebuilding them in CI