Dagster Snowflake

Warnings

• breaking Python 3.6 and 3.8 are no longer supported. `dagster-snowflake` dropped Python 3.6 support due to its underlying `snowflake-connector-python` dependency, and Dagster core no longer supports Python 3.8 (EOL 2024-10-07).
  Fix: Upgrade to Python 3.10, 3.11, 3.12, or 3.13 and ensure all Dagster and `dagster-snowflake` packages are compatible with your chosen Python version. The library currently requires Python <3.15, >=3.10.
• gotcha When loading Pandas DataFrames with timestamp columns to Snowflake, `snowflake-connector-python` (v3.5.0+) may cause data corruption if timestamps are not timezone-aware. The `SnowflakePandasIOManager` attempts to mitigate this by assigning UTC if no timezone is present.
  Fix: Ensure all timestamp data in Pandas DataFrames is timezone-aware (e.g., using `.dt.tz_localize('UTC')` or `.dt.tz_convert('UTC')`) before writing to Snowflake via `dagster-snowflake-pandas`.
• gotcha Snowflake costs and query performance can be significant issues if not managed correctly. Common anti-patterns include full table scans, inefficient joins, or using Snowflake for high-concurrency, low-latency applications.
  Fix: Utilize Snowflake's query profiler to optimize SQL queries within your Dagster assets. Implement efficient warehouse usage, consider connection pooling, and separate schemas for different data stages (raw, staging, production). Dagster provides observability to help identify bottlenecks.
• gotcha Schema-level permissions with Snowflake future grants can be complex, especially when dbt/Dagster recreates tables. Database-level future grants may not be sufficient, leading to lost SELECT access for reporting roles.
  Fix: Carefully configure schema-level future grants (e.g., `GRANT SELECT ON FUTURE TABLES IN SCHEMA ... TO ROLE ...`) in Snowflake to ensure consistent access when tables are re-materialized or recreated.
• gotcha Sensitive credentials (account, user, password, private key) for Snowflake should always be managed securely using environment variables or a secrets manager, not hardcoded directly in code.
  Fix: Use `EnvVar` (e.g., `EnvVar("SNOWFLAKE_PASSWORD")`) within Dagster resource configurations to retrieve credentials from environment variables. For more advanced setups, integrate with a dedicated secret management system.
• deprecated The `build_snowflake_io_manager` function is considered a legacy API for constructing I/O managers.
  Fix: For new projects or refactoring, prefer using explicit I/O manager classes like `SnowflakePandasIOManager`, `SnowflakePySparkIOManager`, or `SnowflakePolarsIOManager` directly within your `Definitions` resources.

Install

• pip install dagster-snowflake Base installation
• pip install dagster-snowflake-pandas With Pandas support (for DataFrame I/O)
• pip install dagster-snowflake-pyspark With PySpark support (for PySpark DataFrame I/O)
• pip install dagster-snowflake-polars With Polars support (beta)

Imports

• SnowflakeResource

  from dagster_snowflake import SnowflakeResource

• SnowflakePandasIOManager

  from dagster_snowflake_pandas import SnowflakePandasIOManager

• SnowflakePySparkIOManager

  from dagster_snowflake_pyspark import SnowflakePySparkIOManager

• build_snowflake_io_manager

  from dagster_snowflake import build_snowflake_io_manager

  This is a legacy API for constructing IO managers; newer patterns often use `SnowflakePandasIOManager` or `SnowflakePySparkIOManager` directly.

Quickstart

This quickstart demonstrates how to define a Dagster asset that connects to Snowflake using `SnowflakeResource`. It uses environment variables for secure credential management and performs a simple SQL operation (table creation and data insertion) within the asset's compute function.

import os
from dagster import asset, Definitions, EnvVar
from dagster_snowflake import SnowflakeResource

@asset
def my_snowflake_asset(snowflake: SnowflakeResource):
    """An example Dagster asset that interacts with Snowflake."""
    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        # Example: Create a table if it doesn't exist and insert data
        cursor.execute(
            """
            CREATE TABLE IF NOT EXISTS my_database.my_schema.my_table (
                id INT,
                name VARCHAR
            );"
        )
        cursor.execute("INSERT INTO my_database.my_schema.my_table (id, name) VALUES (1, 'Alice');")
        cursor.execute("INSERT INTO my_database.my_schema.my_table (id, name) VALUES (2, 'Bob');")
        result = cursor.execute("SELECT COUNT(*) FROM my_database.my_table").fetchone()
        print(f"Rows in my_table: {result[0]}")
    return {'rows_processed': result[0] if result else 0}

defs = Definitions(
    assets=[my_snowflake_asset],
    resources={
        "snowflake": SnowflakeResource(
            account=EnvVar("SNOWFLAKE_ACCOUNT"),
            user=EnvVar("SNOWFLAKE_USER"),
            password=EnvVar("SNOWFLAKE_PASSWORD"),
            database=EnvVar("SNOWFLAKE_DATABASE"),
            schema=EnvVar("SNOWFLAKE_SCHEMA"),
            warehouse=os.environ.get("SNOWFLAKE_WAREHOUSE", ""), # Optional
            role=os.environ.get("SNOWFLAKE_ROLE", "") # Optional
        )
    },
)

# To run this locally, set the following environment variables:
# export SNOWFLAKE_ACCOUNT="your-account-identifier"
# export SNOWFLAKE_USER="your-username"
# export SNOWFLAKE_PASSWORD="your-password"
# export SNOWFLAKE_DATABASE="your-database"
# export SNOWFLAKE_SCHEMA="your-schema"
# (Optional) export SNOWFLAKE_WAREHOUSE="your-warehouse"
# (Optional) export SNOWFLAKE_ROLE="your-role"
# Then run `dagster dev -f your_file_name.py`