This channel is a place where you can get the updates on using Dagster with Snowflake, ask questions, and discuss with other members of the community. • ❄️ check out our getting started guide • 📚 take a look at the

dagster-snowflake

API docs

Would love to kickstart this channel with a discussion: how is everyone handling loading data into snowflake? The IO Managers provided out of the box load data directly into tables, but I’ve always preferred using stages, and wound up enhancing the S3 IO Manager with a dummy script that will run a

COPY INTO

command to materialize the assets. Would love to see what other approaches people have found.

Some of our replication pipelines use stages and temp tables, but for our data scientists, they mostly use this `write_pandas` function. We use assets but handle all the writing of resources to Snowflake within the body of the asset rather than using IO Managers.

This message was deleted.

Plus 1 to the above - when it comes to data scientist/analysts they tend to use

write_pandas

and somewhere on the body of the func. And we have some legacy S3 to staging stuff. However, as a data eng, anything new I build (that’s an asset) I prefer to use the snowflake IO manager and directly to the table.

Interesting. May I ask if that was an explicit decision or if it just happened more or less organically?

I didn't understand IO Managers and assets initially, so I started off building around them. But once I did understand them, I couldn't really envision the data scientists / analysts using them effectively. It hides a lot of pretty critical logic imo, so just from a transparency perspective I don't love it, but could also be some of the use cases we have don't really support it (e.g. passing metadata around to orchestrate Sagemaker jobs)

I would assume that having abstraction for data analysts/scientist is a good thing because each could end up using different methods saving data to snowflake or S3, etc for pickup by other tools causing a lot of inefficiencies or just bad implementations. My hope is that by having data eng provide these very common tasks (saving and getting data via io manager) that scientist/analyst have one less thing to worry about BUT it does require a change in mindset which some people will definitely not want (or can’t) do.

Agreed, and I think it makes sense once you have a certain scale of assets. Right now, we have a small number assets across a variety of systems -- I would basically be writing 7 different IO managers right now, to support 4 instances of each. I think something like

pandas.DataFrame -> snowflake_table

is pretty easy to grok, but it's a little less clear for some other types of assets. dbt cloud job, Sagemaker training job, Rockset query lambda, etc. A good comparison here is dbt -- they essentially offer a single IO Manager per project, and you get a ton of mileage out of it

I’m very interested in this as I’m preparing to onboard a data scientist (and later more people) into dagster/the rest of the infrastructure. I agree that it takes a while to grasp assets, resources, and IO managers, but I like the idea of removing this complexity so they can focus on just returning a

pd.DataFrame

or json object and I can ensure everything integrates with the rest of the datalake in s3, but I’m curious to see what kinds of conversations I’ll have.

We use Meltano EL process from MySQL to Snowflake and Use DBT for transformation there-by doing ELT. All managed with 3 step Dagster Flow:: Meltano EL ( Multi-Assets ) -> Data Verification ( Multi-Assets ) -> DBT ( Multi-Assets outputs ) - Final Notification via Triggers to Slack

I’m trying to get

dagster_snowflake_pandas

working on my local machine but I keep getting this error:

snowflake.connector.errors.OperationalError: 250001: Could not connect to Snowflake backend after 0 attempt(s).Aborting

I’m able to successfully connect using

snowflake.connector

in a REPL using the same account, user, and password so I really don’t know what to do next.

is there any way to use the io_manager but not use pandas…I can’t find any documentation on TypeHandlers. I need to load a dbt asset as an upstream asset and I believe I need an io_manager for this, since dagster won’t find a locally stored asset with that key

When using

dagster-snowflake-pandas

, when an

@asset

function returns a Data Frame, does Dagster download the data into Dataframe and hold it in memory? If yes, what mechanism does it use to download and build a Panda dataframe? Can it be possible to convert into a Polars dataframe ?

I’m having an issue where datetime Pandas dataframe columns are not being created as datetime columns in Snowflake. I created a basic proof of concept:

@asset
def dagster_df_dates():
    return pd.DataFrame(pd.date_range("2023-01-01", "2023-01-31"), columns=["A"])

In Snowflake this column is created as a varchar. What am I doing wrong?

Is this

define_asset_job()

deprecated? I can't find it in the documentation. I'm wondering how the

"all_assets_job"

is created on that line...

or can somebody point me to the docs for

define_asset_job()

?

Is there a way to force a

snowflake_resource

to run bootstrap queries each time it connects? For example, each time I connect, I need to run

USE SECONDARY ROLES ALL;

because it provides me the read access I need for pulling data from schemas that are not my default schema.

❄️ New Snowflake I/O manager features in this week’s release ❄️ The Snowflake I/O manager now support PySpark DataFrames! To use, install the

dagster-snowflake-pyspark

package and use the

snowflake_pyspark_io_manager

API docs are here, with more docs coming soon! Here’s a full code example to get you started:

from dagster_snowflake_pyspark import snowflake_pyspark_io_manager
from pyspark import SparkFiles
from pyspark.sql import (
    DataFrame,
    SparkSession,
)
from pyspark.sql.types import (
    DoubleType,
    StringType,
    StructField,
    StructType,
)

from dagster import Definitions, asset

SNOWFLAKE_JARS = "net.snowflake:snowflake-jdbc:3.8.0,net.snowflake:spark-snowflake_2.12:2.8.2-spark_3.0"


@asset
def iris_dataset() -> DataFrame:
    spark = SparkSession.builder.config(
        key="spark.jars.packages",
        value=SNOWFLAKE_JARS,
    ).getOrCreate()

    schema = StructType(
        [
            StructField("Sepal length (cm)", DoubleType()),
            StructField("Sepal width (cm)", DoubleType()),
            StructField("Petal length (cm)", DoubleType()),
            StructField("Petal width (cm)", DoubleType()),
            StructField("Species", StringType()),
        ]
    )

    url = "<https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data>"
    spark.sparkContext.addFile(url)

    return spark.read.schema(schema).csv("file://" + SparkFiles.get("iris.data"))


defs = Definitions(
    assets=[iris_dataset],
    resources={
        "io_manager": snowflake_pyspark_io_manager.configured(
            {
                "account": "abc1234.us-east-1",
                "user": {"env": "SNOWFLAKE_USER"},
                "password": {"env": "SNOWFLAKE_PASSWORD"},
                "database": "FLOWERS",
                "warehouse": "PLANTS",
                "schema": "IRIS",
            }
        )
    },
)

as always, if you have an feedback, bug reports, or feature requests, let me know here!

@jamie pandas.DataFrame.append is deprecated: https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.append.html It's used in

dagster-snowflake

for things like

context.resources.snowflake.execute_queries()

as you can see here, when the

use_pandas_result

flag is True https://docs.dagster.io/_modules/dagster_snowflake/resources#SnowflakeConnection.execute_queries

execute_queries

is a bit problematic when you're appending to the same data frame. That might be a bug? Seems like it's assuming that each query will return a separate dataframe and that those will be appended to a list -- which is similar to how it works if the

use_pandas_result

is False. However,

result

is initialized as a DataFrame and not as a list when the flag is true

hence, the only way that really seems to work is if the queries are returning the same format of data frame and/or result.

If

use_pandas_result

is True, it's probably best to just return a list of DataFrames

I wrote it up here: https://github.com/dagster-io/dagster/issues/12084

I've used Dagster with Snowflake before, but am now using SDA's. I'm trying to make sure I'm thinking about and using them in the right way. When I'm running an op that is strictly using data within snowflake (i.e. just running a query within Snowflake that creates a new table), should I be using an asset? Or is this simply an op?

I originally opened a discussion about how other snowflake users are loading data. After some playing around and now having onboarded a data scientist into the platform, I wanted to share the pattern I ultimately adopted. The code below is simplified to keep it high-level.

@usable_as_dagster_type
class UtilsSnowInput(pydantic.BaseModel):
    class Config:
        arbitrary_types_allowed = True

    dest_namespace: str
    data: pd.DataFrame

# s3_io_manager enhanced:
def handle_output(self, context: OutputContext, obj):
    ... # s3 load specific logic

    if isinstance(obj, UtilsSnowInput):
        context.log.debug(f"Attempting snowflake upload")
        parquet_path = self._upload_df(obj) # loads obj.data as parquet
        yield MetadataEntry(
            "S3 parquet storage path",
            value=MetadataValue.path(f"s3://{self.bucket}/{parquet_path}"),
        )
        # utils_snow is instantiated with a snowflake stage path as an optional parameter, runs a COPY INTO command into dest_namespace
        yield from self.utils_snow.copy_into_landing_area(
            context,
            parquet_path,
        )

# utils_snow.copy_into_landing_area wraps a COPY INTO command, cleans up the landing area, and yields some more metadata such as number of rows
def _get_copy_into_statement(
    self,
    remote_filepath: str,
    dest_namespace: str,
):
    return (
        f"COPY INTO {dest_namespace}(RAW) FROM @{self.stage}\n"
        f"FILES =('{remote_filepath}')\n"
        f"FILE_FORMAT = (type = '{remote_filepath.split('.')[-1]}');"
    )

This effectively means the users are free to write logic as they see fit, the only requirement is returning a

UtilsSnowInput

with the required parameters

FYI, I was not expecting using

key_prefix

to try to specify a schema for storing data in Snowflake. That probably should be mentioned in the documentation https://docs.dagster.io/integrations/dbt/reference#adding-a-prefix-to-asset-keys

image.png

My expectation was that it would materialize tables with the

dagster_

prefix in the schema I already had specified in my connection parameters