How I Wired Snowflakeβs Native dbt Projects to Airflow β And Finally Got True End-to-End Orchestration
Iβll be honest with you β for a long time I was running dbt the way most people run it. dbt Core installed on a server, profiles.yml file that I kept updating manually, a cron job (yes, a cron job) doing the scheduling, and Airflow somewhere nearby doing the βrealβ orchestration while dbt lived in its own separate corner of the infrastructure.
It worked. It was fine. It was also quietly annoying in ways that Iβd gotten so used to I stopped noticing them. Managing the dbt server separately. Keeping the Snowflake credentials synced in two places. Debugging failures by jumping between the Airflow UI, SSH logs on the dbt server, and Snowsight β all at once.
Then Snowflake went GA with dbt Projects in November 2025, and I spent a weekend rebuilding the whole thing. This article is what I learned.
What weβre building here is a genuine end-to-end pipeline: raw data lands in Snowflake, Airflow orchestrates the entire flow, and the dbt transformations run as a native DBT PROJECT object inside Snowflake β not on an external box, not in a container, inside Snowflake itself. The monitoring, the scheduling trigger, the execution logs β all in one place.
Letβs build it from the ground up.
First β What Exactly Is a dbt Project on Snowflake?
This is important because the terminology can trip you up, and I donβt want you 45 minutes into setup before the confusion hits.
dbt Projects on Snowflake let you use familiar Snowflake features to create, edit, test, run, and manage dbt Core projects. You can use Workspaces in Snowsight to work with dbt project files and directories and deploy a dbt project as a schema-level DBT PROJECT object.
The key word there is object. Snowflake introduces a first-class schema-level object called DBT PROJECT. The DBT PROJECT object in Snowflake is essentially a file container that can contain one or more dbt Core projects. Furthermore, the DBT PROJECT object is versioned so that each change made to the object via ALTER will add a new version.
This means your dbt project β the models, the sources YAML, the dbt_project.yml β lives inside Snowflake as a versioned, native object. Not on a VM. Not in an S3 bucket somewhere. In Snowflake itself.
dbt Projects on Snowflake streamline workflows for data engineers to standardize and automate transformation pipelines by allowing for: development and testing in Workspaces using a file-based IDE that integrates with Git; visualization and debugging of DAGs to inspect lineage and dependencies directly in the UI; deployment and scheduling using native Snowflake Tasks; and selection of dbt commands such as COMPILE, TEST, RUN and more, right from the native Workspaces IDE.
So yes β you can schedule and run it purely with Snowflake Tasks and never touch Airflow. But if your organization already runs Airflow, or if your dbt pipeline is one piece of a larger orchestration that includes data ingestion, validation, downstream alerts, and reporting β you want Airflow in charge, calling into Snowflake to execute the DBT PROJECT object. That hybrid approach is exactly what this article covers.
The Architecture Weβre Building
Before I show you a single line of code, let me draw the full picture because I think this is where most blog posts let you down β they show you a piece without the whole.
[Source System / S3 / API]
β
[Airflow DAG starts]
β
Task 1: Load raw data β Snowflake staging table (via COPY INTO or S3 stage)
β
Task 2: Run data quality checks on raw data (SQLExecuteQueryOperator)
β
Task 3: EXECUTE DBT PROJECT β runs dbt build on your native Snowflake dbt project
β
Task 4: Post-run row count validation (SQLExecuteQueryOperator)
β
Task 5: Trigger downstream alert / Slack notification / refresh BI layer
β
[Pipeline complete]
Airflow owns the orchestration. Snowflake owns the execution of the dbt transformations. The DBT PROJECT object is what bridges them β because you can trigger it with a SQL command, and Airflowβs SQLExecuteQueryOperator can fire that SQL command.
That SQL command, by the way, is beautifully simple:
EXECUTE DBT PROJECT my_database.my_schema.my_dbt_project
ARGS = 'dbt build'
VERSION = 'LAST';
EXECUTE DBT PROJECT executes the specified dbt project object or the dbt project in a Snowflake workspace using the dbt command and command-line options specified. Snowflake Documentation
One SQL statement. Thatβs all Airflow needs to fire. Let me now show you the full setup to make that work.
Step 1: Snowflake Setup β Roles, Warehouse, and Permissions
I always start here because bad permissions cause the most confusing failures, and they surface late in the process when youβre tired and frustrated.
USE ROLE ACCOUNTADMIN;
-- Create a dedicated role for dbt execution
CREATE OR REPLACE ROLE dbt_executor_role;
GRANT ROLE dbt_executor_role TO ROLE SYSADMIN;
-- Create the service user Airflow will use
CREATE OR REPLACE USER airflow_svc_user
PASSWORD = 'YourStrongPassword123!'
DEFAULT_ROLE = dbt_executor_role
DEFAULT_WAREHOUSE = dbt_transform_wh
COMMENT = 'Airflow service user for dbt orchestration';
GRANT ROLE dbt_executor_role TO USER airflow_svc_user;
-- Create a dedicated warehouse for dbt runs
USE ROLE SYSADMIN;
CREATE OR REPLACE WAREHOUSE dbt_transform_wh
WITH WAREHOUSE_SIZE = 'SMALL'
AUTO_SUSPEND = 120
AUTO_RESUME = TRUE
INITIALLY_SUSPENDED = TRUE;
GRANT ALL ON WAREHOUSE dbt_transform_wh TO ROLE dbt_executor_role;
-- Grant database and schema privileges
GRANT USAGE ON DATABASE analytics_db TO ROLE dbt_executor_role;
GRANT USAGE, CREATE TABLE, CREATE VIEW ON SCHEMA analytics_db.staging TO ROLE dbt_executor_role;
GRANT USAGE, CREATE TABLE, CREATE VIEW ON SCHEMA analytics_db.marts TO ROLE dbt_executor_role;
-- Grant the ability to execute dbt project objects
GRANT EXECUTE DBT PROJECT ON SCHEMA analytics_db.transforms TO ROLE dbt_executor_role;
I made a mistake my first time through β I granted object-level access but forgot the schema-level EXECUTE DBT PROJECT privilege, which is separate. The error message wasnβt obvious. Save yourself that 20-minute debugging session.
Step 2: Deploy Your dbt Project as a Native Snowflake Object
This is the step that feels the most different from traditional dbt Core setup. Youβre not installing dbt on a server. Youβre registering your project inside Snowflake.
Option A: Via Snowsight Workspaces (recommended for first time)
Log into Snowsight, navigate to Workspaces, and connect it to your Git repository:
-- First, create an API integration for GitHub
CREATE OR REPLACE API INTEGRATION github_integration
API_PROVIDER = git_https_api
API_ALLOWED_PREFIXES = ('https://github.com/yourorg/')
ENABLED = TRUE;
-- Create the Git repository object in Snowflake
CREATE OR REPLACE GIT REPOSITORY dbt_project_repo
API_INTEGRATION = github_integration
GIT_CREDENTIALS = my_github_secret
ORIGIN = 'https://github.com/yourorg/your-dbt-project.git';
Option B: Deploy via SQL (great for CI/CD)
-- Create the DBT PROJECT object from your connected Git repo
CREATE OR REPLACE DBT PROJECT analytics_db.transforms.sales_dbt_project
FROM GIT REPOSITORY dbt_project_repo
REF = 'main'
TARGET_PATH = 'models/'
WAREHOUSE = dbt_transform_wh;
Install dbt dependencies:
Install dependencies by executing the dbt deps command within a Snowflake workspace, local machine, or git orchestrator to populate the dbt_packages folder for your dbt Project.
-- Run this once after creating the project, or include in CI/CD
EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
ARGS = 'dbt deps'
VERSION = 'LAST';
A heads up on this: running dbt deps to install packages requires an external access integration when executed inside Snowflake Workspaces, since the runtime needs to reach external package repositories. Alternatively, you can run dbt deps locally or in your CI pipeline and include the populated dbt_packages folder in your deployment artifact.
I found it cleaner to run dbt deps in my GitHub Actions pipeline and commit the dbt_packages folder, rather than configuring external access integrations for every environment. Your call β both approaches work.
Verify it deployed correctly:
-- Check your dbt project versions
SHOW DBT PROJECTS IN SCHEMA analytics_db.transforms;
-- Test execute manually before wiring Airflow
EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
ARGS = 'dbt compile'
VERSION = 'LAST';
If dbt compile completes without error, your project is live and ready to be called by Airflow.
Step 3: Set Up a Real dbt Project Structure
Let me show you what the actual project looks like. Iβm using a sales pipeline as the example β raw orders come in, we stage them, build a fact table, and create a daily summary mart.
dbt_project.yml:
name: 'sales_pipeline'
version: '1.0.0'
config-version: 2
profile: 'snowflake_prod'
model-paths: ["models"]
test-paths: ["tests"]
seed-paths: ["seeds"]
models:
sales_pipeline:
staging:
+schema: staging
+materialized: view
marts:
+schema: marts
+materialized: table
models/staging/stg_orders.sql:
-- Staging model: clean and type-cast raw orders
WITH raw AS (
SELECT * FROM {{ source('raw', 'orders_raw') }}
),
cleaned AS (
SELECT
order_id::VARCHAR AS order_id,
customer_id::VARCHAR AS customer_id,
order_date::DATE AS order_date,
UPPER(TRIM(status)) AS order_status,
amount::DECIMAL(18, 2) AS order_amount,
region::VARCHAR AS region,
CURRENT_TIMESTAMP() AS _loaded_at
FROM raw
WHERE order_id IS NOT NULL
AND order_date >= '2023-01-01'
)
SELECT * FROM cleaned
models/marts/fct_daily_orders.sql:
-- Fact table: daily order summary by region
WITH staged AS (
SELECT * FROM {{ ref('stg_orders') }}
)
SELECT
order_date,
region,
order_status,
COUNT(DISTINCT order_id) AS total_orders,
COUNT(DISTINCT customer_id) AS unique_customers,
SUM(order_amount) AS total_revenue,
AVG(order_amount) AS avg_order_value,
SUM(CASE WHEN order_status = 'RETURNED'
THEN order_amount ELSE 0 END) AS returned_amount,
CURRENT_TIMESTAMP() AS _refreshed_at
FROM staged
GROUP BY order_date, region, order_status
ORDER BY order_date DESC, region
models/staging/sources.yml:
version: 2
sources:
- name: raw database: analytics_db schema: raw_landing tables:
- name: orders_raw description: βRaw orders from the source systemβ columns:
- name: order_id tests:
- not_null
- unique
- name: customer_id tests:
- not_null
- name: order_date tests:
- not_null
- name: amount tests:
- not_null
- name: order_id tests:
- name: orders_raw description: βRaw orders from the source systemβ columns:
models/marts/schema.yml:
version: 2
models:
- name: fct_daily_orders
description: "Daily order summary by region and status"
columns:
- name: order_date
tests:
- not_null
- name: total_orders
tests:
- not_null
- name: total_revenue
tests:
- not_null
This gives us a clean, testable project with source freshness checks and column-level tests. When Airflow executes dbt build, all of this runs β models + tests β in dependency order.
Step 4: Wire It All Together in Airflow
Now the fun part. Iβm going to show you a complete Airflow DAG that:
- Validates raw data arrived in Snowflake
- Fires the native dbt project execution
- Validates row counts on the output marts
- Sends a Slack notification on success or failure
First, install the Snowflake provider if you havenβt:
pip install apache-airflow-providers-snowflake
Set up your Snowflake connection in the Airflow UI (Admin β Connections):
Connection ID : snowflake_analytics
Connection Type : Snowflake
Account : yourorg.us-east-1
Login : airflow_svc_user
Password : YourStrongPassword123!
Schema : transforms
Database : analytics_db
Warehouse: dbt_transform_wh
Role : dbt_executor_role
Now the DAG:
dags/sales_pipeline_dag.py:
from airflow import DAG
from airflow.providers.snowflake.operators.snowflake import SQLExecuteQueryOperator
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.empty import EmptyOperator
from airflow.utils.dates import days_ago
from datetime import datetime, timedelta
import logging
# ββ Default args ββββββββββββββββββββββββββββββββββββββββββββββββ
default_args = {
'owner': 'data-engineering',
'depends_on_past': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
'email_on_failure': True,
'email': ['[email protected]'],
}
SNOWFLAKE_CONN = 'snowflake_analytics'
# ββ SQL snippets βββββββββββββββββββββββββββββββββββββββββββββββββ
RAW_DATA_CHECK_SQL = """
SELECT COUNT(*) AS raw_row_count
FROM analytics_db.raw_landing.orders_raw
WHERE order_date = CURRENT_DATE() - 1;
"""
EXECUTE_DBT_SQL = """
EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
ARGS = 'dbt build --select staging.stg_orders+ --vars "{\\"run_date\\": \\"{{ ds }}\\"}"'
VERSION = 'LAST';
"""
MART_VALIDATION_SQL = """
SELECT
COUNT(*) AS mart_row_count,
MAX(order_date) AS latest_date,
SUM(total_revenue) AS total_revenue
FROM analytics_db.marts.fct_daily_orders
WHERE order_date = CURRENT_DATE() - 1;
"""
ROW_COUNT_GUARD_SQL = """
SELECT
CASE
WHEN COUNT(*) = 0
THEN 'FAIL: No rows found in mart for yesterday'
ELSE 'PASS: ' || COUNT(*) || ' rows present'
END AS validation_result
FROM analytics_db.marts.fct_daily_orders
WHERE order_date = CURRENT_DATE() - 1;
"""
# ββ DAG definition βββββββββββββββββββββββββββββββββββββββββββββββ
with DAG(
dag_id='sales_pipeline_end_to_end',
default_args=default_args,
description='End-to-end sales pipeline: raw β dbt native project β marts',
schedule_interval='0 6 * * *', # 6 AM UTC daily
start_date=days_ago(1),
catchup=False,
tags=['snowflake', 'dbt', 'sales'],
) as dag:
# Task 1: Check raw data arrived
check_raw_data = SQLExecuteQueryOperator(
task_id='check_raw_data_arrived',
conn_id=SNOWFLAKE_CONN,
sql=RAW_DATA_CHECK_SQL,
handler=lambda cursor: logging.info(
f"Raw row count: {cursor.fetchone()[0]}"
),
)
# Task 2: Execute the native dbt project on Snowflake
run_dbt_project = SQLExecuteQueryOperator(
task_id='execute_dbt_project_snowflake',
conn_id=SNOWFLAKE_CONN,
sql=EXECUTE_DBT_SQL,
# Give dbt build enough time for large projects
execution_timeout=timedelta(hours=2),
)
# Task 3: Post-run mart validation
validate_mart_output = SQLExecuteQueryOperator(
task_id='validate_mart_output',
conn_id=SNOWFLAKE_CONN,
sql=ROW_COUNT_GUARD_SQL,
handler=lambda cursor: logging.info(
f"Validation result: {cursor.fetchone()[0]}"
),
)
# Task 4: Run broader stats query (logged for observability)
log_mart_stats = SQLExecuteQueryOperator(
task_id='log_mart_statistics',
conn_id=SNOWFLAKE_CONN,
sql=MART_VALIDATION_SQL,
)
# Task 5: Success marker
pipeline_complete = EmptyOperator(task_id='pipeline_complete')
# ββ Dependencies βββββββββββββββββββββββββββββββββββββββββββββ
(
check_raw_data
>> run_dbt_project
>> validate_mart_output
>> log_mart_stats
>> pipeline_complete
)
Step 5: Running Specific dbt Selectors from Airflow
One of the things I really like about this approach is that you get the full power of dbtβs selector syntax passed straight through the ARGS parameter. You donβt have to run the entire project every time.
Run only staging models:
EXECUTE_STAGING_ONLY = """
EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
ARGS = 'dbt run --select staging.*'
VERSION = 'LAST';
"""
Run a specific model and all its downstream dependencies:
EXECUTE_ORDERS_DOWNSTREAM = """
EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
ARGS = 'dbt build --select stg_orders+'
VERSION = 'LAST';
"""
Run tests only, separate from the model run:
RUN_DBT_TESTS = """
EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
ARGS = 'dbt test --select staging.*'
VERSION = 'LAST';
"""
This means you can split a single DAG into multiple tasks β one for staging, one for marts, one for tests β and get granular retry behavior in Airflow if something fails mid-pipeline. Instead of rerunning everything, Airflow retries only the failed task.
Hereβs that pattern as a DAG:
run_staging = SQLExecuteQueryOperator(
task_id='run_dbt_staging',
conn_id=SNOWFLAKE_CONN,
sql="""
EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
ARGS = 'dbt run --select staging.*'
VERSION = 'LAST';
""",
)
test_staging = SQLExecuteQueryOperator(
task_id='test_dbt_staging',
conn_id=SNOWFLAKE_CONN,
sql="""
EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
ARGS = 'dbt test --select staging.*'
VERSION = 'LAST';
""",
)
run_marts = SQLExecuteQueryOperator(
task_id='run_dbt_marts',
conn_id=SNOWFLAKE_CONN,
sql="""
EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
ARGS = 'dbt run --select marts.*'
VERSION = 'LAST';
""",
)
run_staging >> test_staging >> run_marts
This is how I actually run it in practice. If staging tests fail, marts never execute. If marts fail, I retry marts without re-running staging. Clean dependency management with minimal code.
Step 6: Handling New Versions of Your dbt Project
This is something I didnβt think about until I pushed a breaking change to main and my 6 AM pipeline executed the wrong version.
The DBT PROJECT object is versioned so that each change made to the object via ALTER will add a new version. The versions are named according to the pattern VERSION$<num>.
In practice, your CI/CD pipeline (GitHub Actions, etc.) should update the DBT PROJECT object after any merge to main:
# .github/workflows/deploy_dbt.yml
name: Deploy dbt Project to Snowflake
on:
push:
branches: [main]
jobs:
deploy:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Install Snowflake CLI
run: pip install snowflake-cli-labs
- name: Deploy new dbt project version
env:
SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
SNOWFLAKE_USER: ${{ secrets.SNOWFLAKE_USER }}
SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}
run: |
snow dbt deploy \
--project-name analytics_db.transforms.sales_dbt_project \
--from-git \
--ref main
And in your Airflow SQL, VERSION = 'LAST' always picks up the most recently deployed version automatically. So once CI/CD deploys a new version, the next DAG run picks it up with no Airflow changes needed.
Step 7: Monitoring β What to Watch and Where
Before this setup, I was watching three screens at once when something went wrong. Now itβs mostly one.
In Snowsight:
-- Check recent dbt project execution history
SELECT
query_id,
query_text,
execution_status,
start_time,
end_time,
DATEDIFF('second', start_time, end_time) AS duration_seconds,
error_message
FROM TABLE(
INFORMATION_SCHEMA.QUERY_HISTORY(
END_TIME_RANGE_START => DATEADD('day', -1, CURRENT_TIMESTAMP()),
RESULT_LIMIT => 50
)
)
WHERE query_text ILIKE '%EXECUTE DBT PROJECT%'
ORDER BY start_time DESC;
Row count drift detection (add this as an Airflow task):
-- Compare today's mart row count to yesterday's
-- Flag if it drops more than 20%
WITH today AS (
SELECT COUNT(*) AS cnt
FROM analytics_db.marts.fct_daily_orders
WHERE order_date = CURRENT_DATE() - 1
),
yesterday AS (
SELECT COUNT(*) AS cnt
FROM analytics_db.marts.fct_daily_orders
WHERE order_date = CURRENT_DATE() - 2
)
SELECT
today.cnt AS today_rows,
yesterday.cnt AS yesterday_rows,
ROUND((today.cnt - yesterday.cnt) / NULLIF(yesterday.cnt, 0) * 100, 2) AS pct_change,
CASE
WHEN today.cnt < yesterday.cnt * 0.80
THEN 'ALERT: Row count dropped over 20%'
ELSE 'OK'
END AS status
FROM today, yesterday;
I added this query as a SQLExecuteQueryOperator task right after the mart validation step. If the row count drops by more than 20% compared to the previous day, the task raises a warning in Airflow logs, and the email alert fires.
Not every data quality problem shows up as a dbt test failure. Sometimes the data just quietly shrinks because an upstream feed stopped delivering. This catches that.
What This Setup Actually Changed for Me
I want to be real about this because I think the βbenefitsβ sections in most blog posts are too abstract.
Before: My pipeline had six moving parts. Airflow DAG on one server. dbt installed on a separate instance. profiles.yml with credentials that needed updating every time we rotated passwords. Separate monitoring in CloudWatch for the dbt server. Debugging a failure meant SSH β dbt server β find the log file β cross-reference with Airflow logs.
After: The pipeline has three moving parts β Airflow, Snowflake, and GitHub. The dbt credentials are managed by Airflowβs Snowflake connection, which I was already maintaining. Debugging a failure means clicking into the Airflow task logs (which capture the SQL response from Snowflake) and if I need more detail, running the QUERY_HISTORY query above in Snowsight.
Performance improvements were significant: during preview, result upload usually took approximately 6 to 6.5 minutes. Now, upload completes approximately 8 to 10x faster in around 40 to 45 seconds.
The startup time improvement alone was worth it for me. My morning pipeline used to take 28-32 minutes. It now consistently runs in 18-22 minutes. Thatβs not from faster models β itβs from the reduction in environment spin-up overhead.
A Few Gotchas I Hit Along the Way
1. The EXECUTE DBT PROJECT command is synchronous by default. Airflow will wait for it to complete before marking the task done. For large projects this is fine β you want that behavior. Just make sure your execution_timeout on the Airflow task is set generously enough.
2. Cross-project references donβt work the way you might expect. Cross-project dependencies must be copied into the root of the main project β Snowflake doesnβt support references to external file paths within the DBT PROJECT object. If you have multiple dbt projects, plan your consolidation before deploying.
3. The VERSION = 'LAST' behavior. This always runs the most recently deployed version. If you want to pin to a specific version for stability in production, use VERSION = 'VERSION$3' (or whatever version number). I run LAST in dev and a pinned version in prod, deployed via CI/CD.
4. Warehouse auto-resume and the first task. The first EXECUTE DBT PROJECT of the day can have a few seconds of latency while dbt_transform_wh auto-resumes. I added a lightweight warm-up query as the very first task in my DAG so the warehouse is already running by the time dbt build kicks off:
warm_up_warehouse = SQLExecuteQueryOperator(
task_id='warm_up_warehouse',
conn_id=SNOWFLAKE_CONN,
sql="SELECT CURRENT_TIMESTAMP();",
)
warm_up_warehouse >> check_raw_data >> run_dbt_project >> ...
Costs almost nothing. Saves 5-10 seconds of variability at the start of every run.
Why I Think This Is the Right Direction
I started exploring this because nobody told me to. My teamβs existing setup worked. A reasonable person would have left it alone.
But the more I looked at this setup, the more I kept thinking about the overhead we carry when tools donβt talk to each other natively. Every boundary between systems is a place where credentials leak, latency is added, and debugging gets harder. The native dbt project in Snowflake closes one of those boundaries. Airflow still owns orchestration β which is where it belongs β but the transformation execution lives where the data lives.
For the growing number of organizations that have standardized on Snowflake, the native integration offers something genuinely compelling: one fewer system to run, one fewer vendor to manage, and one fewer boundary between your data and the logic that transforms it.
That sentence landed for me when I read it. Thatβs exactly what this is.
If youβve been running dbt Core on a server and Airflow alongside it and youβve been tolerating that overhead long enough that youβve stopped noticing it β try this weekend rebuild. You might be surprised how much lighter the pipeline feels on the other side.
And if you do try it and hit something weird, drop it in the comments. Iβm still learning this myself.