When two snowflake assets are being materialized simultaneously in a docker setup, are they in the same session? or are those different sessions? I think they're different sessions?

Just trying to figure out if I'm spamming

ALTER SESSION SET query_tag = "whatever"

Does Dagster include any logic for improving efficiency of simultaneous asset materializations? For instance, if I have two jobs that run at the same time and each has 10 assets that can be materialized independently, does Dagster track performance associated with simultaneous materializations? For instance, Asset A and Asset B can be materialized together faster than Asset B and Asset C, etc... so that, over time, the execution of the jobs is more efficient?

Hello everyone! I’ve started a GitHub discussion for us to talk about how the Snowflake I/O manager should handle Pandas time data. More info in the discussion, but we’d like to get your input on how you’d like Dagster to handle some long-standing issues with how Snowflake stores timestamps. Thank you in advance for your input!

❄️ New in dagster-snowflake 0.19.3 ❄️ A new

SnowflakeResource

follows the new Pythonic Config and Resources system. The existing

snowflake_resource

will continue to be supported. This SnowflakeResource works slightly differently than the

snowflake_resource

. Using the new

SnowflakeResource

you create a

SnowflakeConnection

(snowflake docs) and use that to run queries

from dagster import asset 
from dagster_snowflake import SnowflakeResource 

@asset 
def my_snowflake_asset(snowflake: SnowflakeResource):
    with snowflake.get_connection() as conn:
       return (
            conn.cursor()
            .execute(
                "SELECT * FROM IRIS_DATASET WHERE 'petal_length_cm' < 1 AND"
                " 'petal_width_cm' < 1"
            )
            .fetch_pandas_all()
        )

Also, if you haven’t yet taken a look at the GitHub discussion about handling timestamp data, please let us know your thoughts!

What’s the best way to store and retrieve JSON blobs in Snowflake, using dataframes and the Snowflake I/O manager?

❄️ New in dagster-snowflake 0.19.4 ❄️ The Snowflake resource and I/O manager now have better support for authenticating via private key. Before, if you directly supplied the private key you could run into issues with the newlines in the key. Now you can

base64

encode your key and the resource and I/O manager will decode the key and use it for authentication. If you are already successfully supplying a private key, then your code is still supported and no changes need to be made.

If I have a new column in snowflake is there a way to delete the table and recreate it. I keep having an issue where my source data changes so I need to do that manually and wanted to do it in dagster. I don't mind dropping the table and recreating it everytime either. I am also not using partitions so trying to figure out why it is not dropping my tables

Hi guys, I am using Dagster Cloud and I have restricted IP address access from Snowflake. Is there a way to assign a dedicated IP address to Dagster? or maybe is there a list of IP address that Dagster use to connect to Snowflake?

Is there a way to change the resource's descriptive text? "This resource is for connecting ..."

Is there a simple way to get

snowflake_pandas_io_manager

to only load part of a table for a downstream asset? i.e. add a WHERE clause

Another question: if I have an asset which creates a table in snowflake using the snowflake connector (i.e.

CREATE TABLE ....

), how can I get downstream assets to load that using

snowflake_pandas_io_manager

? i.e. how to declare that the asset lives in snowflake?

Question about trying to update existing tables in snowflake. I'm trying to create workflow where I read from an existing table in snowflake, enrich by adding a few column, and storing the result in another table. Using the

snowflake_pandas_io_manager

as the docs say and time partitions to do this, but it looks like an

@asset

always recreates the table. Was wondering what the suggested route was to go about updating a table on a daily basis in snowflake?

hey all - trying to use the snowflake resource and I can't make heads or tails of what patterns are supported. I expected to be able to run

foo1

from here, but it failed, then i expected to run

foo2

like here but it failed. Then i tried

foo3

from here, and that worked for me. The pattern I'd prefer to encourage others to use is the first -- is that supported?

from dagster import Definitions, asset, OpExecutionContext
from dagster_snowflake import SnowflakeResource

@asset
def foo(context: OpExecutionContext, snowflake: SnowflakeResource):
    result = snowflake.execute_query("select 1")
    <http://context.log.info|context.log.info>(result)

@asset
def foo2(context: OpExecutionContext, snowflake: SnowflakeResource):
    with snowflake.get_connection() as conn:
        result = conn.execute_query("select 1")
    <http://context.log.info|context.log.info>(result)

@asset
def foo3(context: OpExecutionContext, snowflake: SnowflakeResource):
    with snowflake.get_connection() as conn:
        result = conn.cursor().execute("select 1")
    <http://context.log.info|context.log.info>(result)

defs = Definitions(
    assets=[foo, foo2, foo3],
    resources={
        "snowflake": SnowflakeResource(
            account="whatnot-prod",
            user=SNOWFLAKE_USER,
            password=SNOWFLAKE_PASSWORD,
            role="STAGE_ADMIN_ROLE",
            warehouse="STAGE_LOAD_WH",
            database="STAGE",
        )
    }
)

Has anyone played around with linking primary/foreign keys between tables created by dagster in a programmatic way?

In the context of self-dependant assets, what would be the way to make a time-partitioned asset dependant on all past partitions of itself? We are wanting to construct an events table, where in order to calculate events on date t, we need all previous events. I see

AllPartitionMapping

, but I think that would get confused with backfills

hi! I'm brand new to Dagster and have been trying to get some Snowflake examples to work, but it seems like most the examples in the docs don't work. Is it just me? Examples in 🧵

Hello 👋 . I've been using the

SnowflakeConnection

class for a few jobs - trying to query some data and then do stuff with it. I've been using the

execute_query

method to extract the data I need from the warehouse. It seems like all of the sudden the method is no longer recognized:

AttributeError: 'SnowflakeConnection' object has no attribute 'execute_query'

I'm curious as to why this might've happened. Not sure if its something we did on our side. Or maybe it has to do with some update? Not sure. Any help would be very appreciated

Is anyone using Snowflake streams or tasks to run things? Do you integrate it with Dagster in any way?

And / or is anyone looking into using dynamic tables? Using Dagster as a way to manage dynamic tables w/ monitoring might be pretty cool.

I was playing around with dagster snowflake pre built SnowflakePandasIOManager and was wondering how the write works? Is doing upserts or (create or replace the existing table in the schema)?

Hello, I'm using SnowflakeResource and would like to submit multiple statements in one transaction. There is a pattern for this in the Snowflake docs, using the

autocommit=False

argument to

snowflake.connector.connect

(https://docs.snowflake.com/developer-guide/python-connector/python-connector-example#using-context-manager-to-connect-and-control-transactions). How can I get a Snowflake connection with

autocommit=False

from

SnowflakeResource.get_connection()

?

Is it possible to create a Snowflake IO manager with a schema configuration that assigns a given schema per environment, or

None

on prod (so that it infers the schema from the asset keys like normal)? We'd like to avoid giving our assets variable key names per enviroment, so our Asset graph correctly reflects the state it would be in prod. We use dagster-dbt so we aren't in full control of how the asset dependencies are managed. We've tried using a variable configuration on a

SnowflakePandasIoManager

but when we're in a dev environment and try to override the asset key-defined schema named, we get an error for trying to define the schema in multiple places at once

I’ve got some complex business logic I’m trying to implement in a Snowflake Python UDTF. It needs to run as a UDF and not in a full blown dagster project. However, I think the ops/graph abstraction would be very helpful in clarifying this logic. I’m curious whether anyone has anyone tried to implement these methods in a Python UDF?

Has there been any work on using snowpark as an alternative to the snowflake python connector/ sqlalchemy in the IOManager backend? We’ve been using it internally and have seen some significant performance benefits. It’s also quite expressive when it comes to metadata operations. An example might be the flow control for what to do when the schema of an asset changes between code versions. Currently I believe the response is “panic”. It would be fairly easy in this framework to establish some lightweight checks like “do the schemas match. Yes? Cool. No? Check whether we’re supposed to throw an exception or replace the table” Doing so also opens up the possibility of using snowpark dataframes instead of pandas data frames. For operations which are computationally light, but extremely bandwidth intensive (add 1 to every row in this terabyte of data, that sort of thing), it would be great to have the ability to leverage dataframe abstractions on snowflake compute rather than needing to pull the data down locally.

cc: @Joe @josh Question for teams here… What percentage of your pipeline activity is: 1. Dagster -> Snowflake 2. Dagster -> Dbt -> Snowflake We’re building a cost attribution solution for #2 but it seems like there is still a pretty substantial focus on #1 for some teams, as well as workloads that aren’t really meant for dbt

Hiya, a question for the channel - how do you use Dagster for data ingestion with Snowflake? Do you combine it with pipes/streams, or use Dagster standalone for data ingestion?

Hey All, Apologies if this has been asked/answered before.. I'm looking to test some partition strategies for a project, however, in my use case both the source and destination would be in Snowflake. For example, I have a source revenue table and I want to get data for a daily partition, apply some logic, and then save as another table. I'm assuming that this also would allow for rebuilding of these "historical" partitions in my destination table if something were to change in the source data, right? Has anyone done this and if so, would you be willing to provide a loose scaffold of what this asset config might look like?

Hi Everyone.... I”m currently updating some of my snowflake resources from

snowflake_resource

to the Pythonic

SnowflakeResource

and I used to use something like

snowflake.execute_query(...)

but now it looks like there is no

execute_query

method for the Pythonic version and I now have to get a connection and query the db directly. Is this correct?

def get_one(snowflake_resource: SnowflakeResource):
      with snowflake_resource.get_connection() as conn:
                # conn is a snowflake.connector.Connection object
                conn.cursor().execute("SELECT 1")

And to follow up that - is the expectation/migration then to instead of using

execute_query(...use_pandas_result=true)

we should use

cursor.fetch_pandas_all()

? Just wanted to make sure I'm not missing some documented/recommended approach. Thanks

Hi I'm just getting started on a Dagster POC where I need to upload lot of data from S3 to Snowflake. One questions I have is: if I set up a Dagster job to read a partitioned asset from S3 and write it to Snowflake using the I/O manager, how does Dagster handle that under the hood? Will it batch files together when uploading. I will have potentially thousands of files in a hourly partitioned asset in S3