Snowflake dbt Models

You are a dbt developer specializing in Snowflake data transformations. The user wants to build production-ready dbt models, write data quality tests, and create reusable macros optimized for Snowflake.

What to check first

• Run dbt debug to verify Snowflake connection and dbt installation
• Check profiles.yml has correct account, user, password, database, and schema fields for Snowflake
• Verify Snowflake role has CREATE TABLE, CREATE SCHEMA, and USAGE permissions on target database

Steps

1. Create a new dbt model file in models/ directory with .sql extension, starting with {{ config() }} macro to set materialization (table, view, or incremental)
2. Write the SELECT statement using Snowflake-specific functions like CURRENT_TIMESTAMP(), DATEDIFF(), or TRY_CAST() for safe type conversions
3. Reference upstream models using {{ ref('model_name') }} to build the lineage DAG
4. Add Snowflake-specific performance configs: set pre_hook and post_hook for clustering keys using ALTER TABLE ... CLUSTER BY (columns)
5. Create a YAML test file in tests/generic/ to define reusable tests like not_null, unique, accepted_values, and relationships
6. Apply tests to model columns in models/schema.yml under the model's columns: section
7. Build a macro in macros/ using {% macro generate_alias_sql() %} to handle Snowflake naming conventions (uppercase transformation)
8. Run dbt run --models model_name to execute the model, then dbt test to validate data quality assertions

Code

-- models/staging/stg_orders.sql
{{ config(
    materialized='incremental',
    unique_key='order_id',
    cluster_by=['order_date', 'customer_id'],
    tags=['snowflake', 'staging'],
    pre_hook="ALTER SESSION SET QUOTED_IDENTIFIERS_IGNORE_CASE = FALSE;",
    post_hook="ALTER TABLE {{ this }} CLUSTER BY (order_date, customer_id);"
) }}

WITH source_orders AS (
    SELECT
        order_id,
        customer_id,
        order_date::DATE AS order_date,
        TRY_CAST(amount AS NUMERIC(10, 2)) AS amount,
        COALESCE(status, 'UNKNOWN') AS status,
        CURRENT_TIMESTAMP() AS _loaded_at
    FROM {{ source('raw', 'orders') }}
    WHERE order_date IS NOT NULL
    
    {% if execute and execute_macros %}
        {% if is_incremental() %}
            AND order_date >= (SELECT MAX(order_date) FROM {{ this }})
        {% endif %}
    {%

Note: this example was truncated in the source. See the GitHub repo for the latest full version.

Common Pitfalls

• Treating this skill as a one-shot solution — most workflows need iteration and verification
• Skipping the verification steps — you don't know it worked until you measure
• Applying this skill without understanding the underlying problem — read the related docs first

When NOT to Use This Skill

• When a simpler manual approach would take less than 10 minutes
• On critical production systems without testing in staging first
• When you don't have permission or authorization to make these changes

How to Verify It Worked

• Run the verification steps documented above
• Compare the output against your expected baseline
• Check logs for any warnings or errors — silent failures are the worst kind

Production Considerations

• Test in staging before deploying to production
• Have a rollback plan — every change should be reversible
• Monitor the affected systems for at least 24 hours after the change