Airflow + Snowflake: Queued Queries and the 45-Minute Wait

The nightly ELT DAG ran fine. Every task turned green. It just started 45 minutes late, every single night, and nobody noticed until the dashboard numbers stopped matching the morning standup.

Forty-five minutes of queuing against a Snowflake warehouse that wasn't there when the first query landed. Not an error, not a failure — a queue. That's the most dangerous way Snowflake DAGs break: quietly, politely, and on a schedule you can't see in the task logs.

The fix is a contract, not a prayer: a dedicated warehouse with auto-resume, wired through the connection, monitored by QUEUED state. This article builds that contract from the connection up — SnowflakeOperator, the ELT pattern, warehouse sizing, and the cost controls that keep the bill boring.

1. The Airflow → Snowflake Connection Anatomy

The connection is the contract between Airflow and Snowflake. It holds account, user, password or key, warehouse, database, and schema — everything a query needs to know about where it runs. Get the warehouse here first: it's the default every operator and hook inherits.

Create it once, reference the id forever. The DAG never touches credentials, and changing the warehouse for an environment is a config change, not a code change.

Two fields are the classic traps: the account identifier must include the region and cloud provider (company.us-east-1, not company), and the schema path matters if your ELT writes to a specific database.

``bash airflow connections add snowflake_prod \ --conn-type snowflake \ --conn-host company.us-east-1 \ --conn-login etl_airflow \ --conn-extra '{"warehouse": "ETL_WH", "database": "ANALYTICS", "schema": "RAW", "role": "ETL_ROLE"}' ``

For production, pull the password from a secrets backend instead of passing it on the command line.

2. SnowflakeOperator Basics

SnowflakeOperator is a task that runs one SQL statement: run this SQL, on this warehouse, with this connection, retried like any Airflow task. Its output is the query's row count or result set — handy for the next task to consume.

It's the right tool for standard steps: create tables, truncate partitions, run stored procedures, load from stage. For anything with Python logic around the SQL, reach for SnowflakeHook inside a task instead.

The operators in the snowflake provider also handle a lot of the state for you: running, failed, and retried queries are all tracked at the task level, so the 45-minute queue incident shows up as a long-running task — visible, if you look.

Two precedence rules matter in production. Any parameter passed to the operator — warehouse, database, schema, role — takes priority over the same field in the connection, so the connection stays the default and the operator handles the exceptions. And a sql argument holding multiple statements needs split_statements=True (or pass a list of SQL strings), while parameters={...} binds values into the statement.

One deprecation heads-up from the provider docs: SnowflakeOperator is deprecated and will be removed in a future provider version — the stable docs now document SQLExecuteQueryOperator for the same job, with identical connection metadata and the same argument-over-connection precedence. Plan the migration, and know the check operators (SnowflakeCheckOperator, SnowflakeValueCheckOperator) for in-warehouse assertions instead of hand-rolled SQL checks.

from datetime import datetime, timedelta

from airflow.decorators import dag
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator


@dag(
    schedule="@daily",
    start_date=datetime(2026, 7, 1),
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "snowflake_conn_id": "snowflake_prod",
    },
)
def refresh_marts():
    build_orders = SnowflakeOperator(
        task_id="build_orders_mart",
        sql="""
            CREATE OR REPLACE TABLE ANALYTICS.MARTS.ORDERS_DAILY AS
            SELECT DATE(order_ts) AS day, COUNT(*) AS orders, SUM(amount) AS revenue
            FROM ANALYTICS.RAW.ORDERS
            WHERE DATE(order_ts) = '{{ ds }}'
            GROUP BY 1
        """,
        warehouse="ETL_WH",
    )

    build_orders


refresh_marts()

3. ELT: Load Raw, Transform In-Warehouse

ETL pulls transformation into Python; ELT loads raw and transforms inside the warehouse with SQL or dbt. For Airflow + Snowflake, ELT wins almost every time: Snowflake is the fastest engine for this shape of work, and the DAG stays thin — load, then run SQL, then done.

The pattern: one task copies raw data into a landing table, a second runs the transforms as SQL statements, and a third asserts row counts. The Python in the DAG shrinks to orchestration.

The hook reuse matters: every task in this DAG shares the snowflake_prod connection, so the warehouse, database, and schema stay consistent across the whole run. No drift, no 'why is this task in the wrong schema' mysteries.

from datetime import datetime, timedelta

from airflow.decorators import dag, task
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator


@dag(
    schedule="@daily",
    start_date=datetime(2026, 7, 1),
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
)
def elt_orders():
    load_raw = SnowflakeOperator(
        task_id="load_raw_orders",
        snowflake_conn_id="snowflake_prod",
        sql="""
            COPY INTO ANALYTICS.RAW.ORDERS
            FROM @ETL_STAGE/orders_{{ ds }}/
            FILE_FORMAT = (TYPE = 'CSV', SKIP_HEADER = 1)
        """,
        warehouse="ETL_WH",
    )

    transform_orders = SnowflakeOperator(
        task_id="transform_orders",
        snowflake_conn_id="snowflake_prod",
        sql="""
            INSERT INTO ANALYTICS.MARTS.ORDERS_DAILY
            SELECT DATE(o.order_ts) AS day, COUNT(*) AS orders, SUM(o.amount) AS revenue
            FROM ANALYTICS.RAW.ORDERS o
            WHERE DATE(o.order_ts) = '{{ ds }}'
            GROUP BY 1
        """,
        warehouse="ETL_WH",
    )

    @task
    def assert_loaded() -> int:
        hook = SnowflakeHook(snowflake_conn_id="snowflake_prod", warehouse="ETL_WH")
        count = hook.get_first("SELECT COUNT(*) FROM ANALYTICS.MARTS.ORDERS_DAILY WHERE day = '{{ ds }}'")[0]
        if count == 0:
            raise ValueError("No rows loaded for the current interval")
        return count

    load_raw >> transform_orders >> assert_loaded()


elt_orders()

4. Warehouse Auto-Suspend vs DAG Timing

Warehouses suspend after idle time and resume on demand — unless auto-resume is off, in which case the first query queues until someone or something starts the warehouse. That queue is exactly the 45-minute incident.

The timing trap: auto-resume is not instant. A suspended warehouse takes seconds to start, and every query that arrives during startup waits its turn. Nightly DAGs that fire into a cold warehouse pay this tax every run, silently.

The contract: the DAG's connection names a warehouse with AUTO_RESUME = TRUE and a short AUTO_SUSPEND — long enough to not thrash between tasks in one run, short enough to not burn credits overnight. Then the DAG's schedule becomes the only timing variable left.

``sql CREATE WAREHOUSE IF NOT EXISTS ETL_WH WITH WAREHOUSE_SIZE = 'MEDIUM' AUTO_SUSPEND = 60 AUTO_RESUME = TRUE INITIALLY_SUSPENDED = TRUE; ``

5. S3ToSnowflakeOperator for Bulk Loads

When data lands in S3, the amazon provider's S3ToSnowflakeOperator loads it straight into Snowflake in one task: it reads the S3 key, stages it, and runs COPY INTO with your file format and stage settings.

The value is a full Airflow task wrapped around a data transfer — retries, logging, and lineage for free. And it removes the download-reupload hop most teams add by hand.

Watch the connection pair: this operator needs both an S3 connection (aws) and the Snowflake connection, so both must be configured in the same environment. The warehouse still comes from the Snowflake connection, so the sizing contract from section 4 still applies.

from datetime import datetime, timedelta

from airflow.decorators import dag
from airflow.providers.amazon.aws.transfers.s3_to_snowflake import S3ToSnowflakeOperator


@dag(
    schedule="@daily",
    start_date=datetime(2026, 7, 1),
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
)
def load_from_s3():
    load_events = S3ToSnowflakeOperator(
        task_id="s3_events_to_snowflake",
        s3_keys=["events/{{ ds }}/events.json"],
        s3_bucket="analytics-ingest-prod",
        table="ANALYTICS.RAW.EVENTS",
        stage="ETL_STAGE",
        file_format="(TYPE = 'JSON')",
        aws_conn_id="aws_prod",
        snowflake_conn_id="snowflake_prod",
    )

    load_events


load_from_s3()

6. Cost Controls: Warehouses, Retries, Schedule Discipline

Snowflake bills per credit per active second. The cost levers for an Airflow workload: warehouse size, auto-suspend timeout, retry behavior, and what the schedule actually runs.

Right-size the warehouse to the job, not to the fear. An X-Small that finishes in 20 minutes costs one credit; an X-Large idling through the same job costs sixteen. Watch QUEUED time before sizing up — a queue is usually a concurrency problem, and the fix is a second cluster or a separate warehouse, not a bigger one.

Retries multiply cost: each retry re-runs the query and may spin the warehouse back up. Keep retries for transient resume errors, set retry_delay so the warehouse can warm, and use a resource monitor with a hard cap so a runaway DAG can't spend the quarter.

7. Monitoring QUEUED State

The 45-minute incident hid because nobody watched the queue. Task success was green; the queue was invisible. The fix: a query that measures QUEUED time per query, checked like any other metric.

The cleanest query pulls from SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY: elapsed, queued, and executed times per query. Any query with queued_overload_time above a threshold — 60 seconds is a good starting alarm — gets a ticket.

Make it a habit, not an incident response: a weekly review of QUEUED time per warehouse tells you when to resize, when to split warehouses, and when your schedule is outrunning the warehouse's sleep cycle.

SELECT
    query_id,
    warehouse_name,
    user_name,
    start_time,
    queued_overload_time AS queued_ms
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE start_time >= DATEADD(day, -1, CURRENT_TIMESTAMP())
  AND queued_overload_time > 60000
ORDER BY queued_overload_time DESC;