Everything about
dbt — Data Build Tool

-- models/marts/orders_daily.sql
-- Reference other models with ref()

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

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

daily_agg AS (
  SELECT
    DATE_TRUNC('day', o.created_at) AS order_date,
    c.country,
    COUNT(o.order_id)           AS total_orders,
    SUM(o.revenue)             AS total_revenue,
    AVG(o.revenue)             AS avg_order_value
  FROM orders o
  JOIN customers c USING (customer_id)
  GROUP BY 1, 2
)

SELECT * FROM daily_agg

# models/marts/schema.yml
version: 2

models:
  - name: orders_daily
    description: "Daily order aggregation by country"
    config:
      materialized: incremental
      unique_key: [order_date, country]
      partition_by:
        field: order_date
        data_type: date
        granularity: day

    columns:
      - name: order_date
        tests:
          - not_null
          - unique  # per country partition
      - name: total_revenue
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0

my_dbt_project/
├── dbt_project.yml       # Project config
├── profiles.yml          # DB connections
├── packages.yml          # Package deps
├── models/
│   ├── staging/          # Raw → clean
│   │   ├── stg_orders.sql
│   │   └── schema.yml
│   ├── intermediate/     # Business logic
│   │   └── int_orders_with_customers.sql
│   └── marts/            # Analytics-ready
│       ├── orders_daily.sql
│       └── schema.yml
├── macros/               # Jinja macros
├── tests/                # Custom SQL tests
├── snapshots/            # SCD Type 2
├── seeds/                # Static CSV data
├── analyses/             # Ad-hoc SQL
└── target/               # Compiled output
    ├── manifest.json
    ├── run_results.json
    └── catalog.json

-- models/fact/fact_events.sql
{{
  config(
    materialized = 'incremental',
    unique_key   = 'event_id',
    incremental_strategy = 'merge',
    partition_by = {
      'field': 'event_date',
      'data_type': 'date',
      'granularity': 'day'
    },
    cluster_by = ['user_id', 'event_type']
  )
}}

SELECT
  event_id,
  user_id,
  event_type,
  DATE(event_timestamp) AS event_date,
  event_timestamp,
  properties
FROM {{ source('raw', 'events') }}

-- Only new records on incremental runs
{% if is_incremental() %}
  WHERE event_timestamp >
    (SELECT MAX(event_timestamp)
     FROM {{ this }})
{% endif %}

my_project:
  outputs:

    dev:
      type: bigquery
      method: oauth
      project: my-company-dev
      dataset: dbt_{{ env_var('DBT_USER') }}
      threads: 4
      timeout_seconds: 300

    prod:
      type: bigquery
      method: service-account
      project: my-company-prod
      dataset: dbt_production
      keyfile: {{ env_var('BQ_KEYFILE_PATH') }}
      threads: 16
      timeout_seconds: 3600
      location: US
      maximum_bytes_billed: 10000000000000

  target: dev

from airflow import DAG
from airflow.operators.bash import BashOperator
from cosmos import DbtDag, ProjectConfig, ProfileConfig

# Option A: Using Astronomer Cosmos (recommended)
dbt_dag = DbtDag(
    dag_id="dbt_production_daily",
    project_config=ProjectConfig(
        dbt_project_path="/usr/local/airflow/dbt",
    ),
    profile_config=ProfileConfig(
        profile_name="my_project",
        target_name="prod",
    ),
    schedule_interval="0 4 * * *", # 4 AM UTC
    operator_args={
        "vars": {"run_date": "{{ ds }}"},
        "dbt_cmd_flags": ["--threads", "16"],
    },
)

# Option B: BashOperator (simpler)
run_dbt = BashOperator(
    task_id="dbt_run",
    bash_command="""
      dbt run --target prod \
               --select tag:daily \
               --threads 16
    """
)