DuckDB with dbt: Local Data Transformation

# Install dbt with DuckDB adapter
pip install dbt-duckdb
 
# Verify
dbt --version

# Create new project
dbt init my_analytics
 
# Navigate to project
cd my_analytics

my_analytics:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: /Users/yourname/data/analytics.duckdb
      threads: 4
    
    # Production profile (Snowflake example)
    prod:
      type: snowflake
      account: myaccount
      user: analytics_user
      password: "{{ env_var('SNOWFLAKE_PASSWORD') }}"
      role: analytics
      database: analytics
      warehouse: analytics_wh
      schema: public
      threads: 8

# Test connection
dbt debug

-- models/staging/stg_events.sql
{{ config(materialized='view') }}
 
SELECT
    event_id,
    user_id,
    event_name,
    -- Parse JSON properties
    json_extract_string(properties, '$.page') AS page,
    json_extract_string(properties, '$.feature') AS feature,
    CAST(json_extract(properties, '$.revenue') AS DECIMAL(10,2)) AS revenue,
    occurred_at::TIMESTAMP AS occurred_at,
    DATE_TRUNC('day', occurred_at::TIMESTAMP) AS event_date
FROM {{ source('raw', 'events') }}
WHERE occurred_at IS NOT NULL
    AND user_id IS NOT NULL

-- models/intermediate/int_user_sessions.sql
{{ config(materialized='table') }}
 
WITH session_boundaries AS (
    SELECT
        user_id,
        occurred_at,
        event_name,
        -- New session if >30 min since last event
        CASE 
            WHEN DATEDIFF('minute', 
                LAG(occurred_at) OVER (PARTITION BY user_id ORDER BY occurred_at),
                occurred_at) > 30
            OR LAG(occurred_at) OVER (PARTITION BY user_id ORDER BY occurred_at) IS NULL
            THEN 1 ELSE 0
        END AS is_new_session
    FROM {{ ref('stg_events') }}
),
sessions AS (
    SELECT
        user_id,
        occurred_at,
        SUM(is_new_session) OVER (PARTITION BY user_id ORDER BY occurred_at) AS session_num
    FROM session_boundaries
)
SELECT
    user_id,
    session_num,
    MIN(occurred_at) AS session_start,
    MAX(occurred_at) AS session_end,
    COUNT(*) AS event_count,
    DATEDIFF('minute', MIN(occurred_at), MAX(occurred_at)) AS duration_minutes
FROM sessions
GROUP BY user_id, session_num

-- models/marts/fct_revenue.sql
{{ config(
    materialized='incremental',
    unique_key='transaction_id'
) }}
 
SELECT
    t.id AS transaction_id,
    t.customer_id,
    c.segment AS customer_segment,
    c.country,
    t.type AS transaction_type,
    t.mrr_delta,
    t.arr_delta,
    t.occurred_at,
    DATE_TRUNC('month', t.occurred_at) AS revenue_month,
    DATE_TRUNC('quarter', t.occurred_at) AS revenue_quarter
FROM {{ ref('stg_transactions') }} t
LEFT JOIN {{ ref('dim_customers') }} c ON t.customer_id = c.customer_id
 
{% if is_incremental() %}
WHERE t.occurred_at >= (SELECT MAX(occurred_at) FROM {{ this }})
{% endif %}

# models/staging/sources.yml
version: 2
 
sources:
  - name: raw
    schema: main  # DuckDB schema
    tables:
      - name: events
        description: "Raw user events from the application"
      - name: transactions
        description: "Revenue transactions"
      
    freshness:
      warn_after: {count: 24, period: hour}
      error_after: {count: 48, period: hour}

# models/marts/fct_revenue.yml
version: 2
 
models:
  - name: fct_revenue
    description: "Daily revenue transactions"
    columns:
      - name: transaction_id
        tests:
          - unique
          - not_null
      - name: mrr_delta
        tests:
          - not_null
      - name: customer_segment
        tests:
          - accepted_values:
              values: ['free', 'pro', 'enterprise']

-- tests/assert_no_future_revenue.sql
-- Custom test: no revenue dated in the future
SELECT COUNT(*) AS failures
FROM {{ ref('fct_revenue') }}
WHERE occurred_at > CURRENT_TIMESTAMP
HAVING failures > 0

# Run all tests
dbt test
 
# Test specific model
dbt test --select fct_revenue

# Run all models
dbt run
 
# Run specific model and its dependencies
dbt run --select fct_revenue+
 
# Run in production profile
dbt run --target prod
 
# Full refresh (rebuild from scratch)
dbt run --full-refresh
 
# Run + test + docs in one command
dbt build

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

-- Use DuckDB's PIVOT syntax in models
{{ config(materialized='table') }}
 
PIVOT (
    SELECT stage, COUNT(*) AS deals FROM {{ ref('stg_deals') }}
)
ON stage
USING SUM(deals)

# dbt Python models (runs in DuckDB via Pandas)
# models/feature_engineering.py
def model(dbt, session):
    deals = dbt.ref("stg_deals")
    
    df = deals.df()
    df['deal_age_days'] = (pd.Timestamp.now() - pd.to_datetime(df['created_at'])).dt.days
    
    return df

# profiles.yml
denchclaw_analytics:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: /Users/yourname/.openclaw-dench/workspace/workspace.duckdb
      threads: 2

-- models/staging/stg_deals.sql
SELECT 
    id AS deal_id,
    "Company" AS company,
    "Deal Value" AS value,
    "Stage" AS stage,
    "Created At" AS created_at
FROM v_deals