If I do nothing with it, that works fine. If I try to include it as a parameter to an asset definition in code location 2, like this....

@asset(required_resource_keys={"snowflake"}, compute_kind="Python")
def calc_jira_kpis(context: OpExecutionContext, jira_tickets_source_asset: pd.DataFrame) -> None:
    """
    Calculate Jira-derived KPIs.
    """
    .... stuff
    return None

I get an error:

dagster._core.errors.DagsterInvalidDefinitionError: Input asset '["jira_tickets_source_asset"]' for asset '["calc_jira_kpis"]' is not produced by any of the provided asset ops and is not one of the provided sources

What do I need to do to get Dagster to recognize this asset by key? I was following these docs: https://docs.dagster.io/concepts/assets/software-defined-assets#defining-external-asset-dependencies

I’m not sure what it means by “one of the provided sources.” Do I need to register code location 1 with code location 2? I thought dagster was supposed to know about both?

I know I've asked this before, but maybe something has changed.... Is there a way to force the snowflake connector to run bootstrap queries each time it connects? I need it to always run

USE SECONDARY ROLES ALL

before executing queries.

I cannot, for example, use

context.resources.snowflake.execute_queries(sql_queries=queries)

because it doesn't first

USE SECONDARY ROLES ALL

👋

Hey all, I have a question about dbt and asset keys for Snowflake IO Manager. Specifically, if I want an asset to produce data for a source in dbt, I use asset keys for that, but then asset keys simultaneously are used by the snowflake IO manager to provide schema info. Right now, it looks like these two will conflict in my project and I'm not sure what to do about it

Is there a way to override Snowflake-materialized assets checking their

key_prefix

for schema information, and only use the schema value from the passed Snowflake IO manager? We run & test our pipelines in different schema configurations based on local dev, branch qa, and production, but ideally would like the visual organization of the assets in the UI to stay consistent between environments. So the goal would be to have static

key_prefixes

and dynamic io_manager schema configuration, that won't overlap in values

❄️ New in dagster-snowflake 0.18.7 ❄️ In this release, we have added new

SnowflakePandasIOManager

(api docs) and

SnowflakePySparkIOManager

(api docs) that follow the new Pythonic Config and Resources system. The existing

snowflake_pandas_io_manager

and

snowflake_pyspark_io_manager

are still and will continue to be supported. However, we think that the new Pythonic system has huge ergonomic benefits so we recommend you check it out! You can read more about the Pythonic Config and Resources system here , and expect to see even more resources (including updated guides) in the coming weeks!

We use Postgres for metadata/asset/job tracking, running in a Docker container in a multi-container deployment. However, most of our work is done in Snowflake. I was thinking it would be useful to provide a way (in dagit?) to export the Postgres metadata tables to Snowflake for easier exploration. I'd like to be able to look at resource usage by asset materialization, for example, and that data is stored in Postgres. However, I would need to get it into Snowflake to support reporting anyhow. Does anybody know if there's a way to do this?

How are snowflake sessions managed in Dagster? If I have two subsequent assets that each require a snowflake resource that is set up with

snowflake_resource.configured

, are the subsequent asset materializations using the same snowflake session? I'm having a problem trying to run two jobs in different warehouses simultaneously when I know I have access to both warehouses. I'm getting the

Object does not exist, or operation cannot be performed.

error when I try to call

USE WAREHOUSE ..

when I can easily use that warehouse in DBeaver, for instance. Is there a way to print the connection URL that the snowflake connector is using?

❄️ New in

dagster-snowflake

0.19.0 ❄️ Happy 1.3 release everyone! • The Snowflake tutorial and reference page have been updated to use the pythonic resources and config versions of Snowflake I/O managers. If you are still using the pre-0.19.0 versions of these I/O managers, you can still see the older versions of the tutorial and reference page by selecting a prior version of the documentation in the dropdown at the top of the page • This release includes a breaking change for the

SnowflakePandasIOManager

and any other I/O managers using the

SnowflakePandasTypeHandler

. Due to a longstanding issue with the snowflake-python connector, Timestamp data cannot be directly stored in Snowflake (the data get’s garbled and stored as non-sensical dates). Prior to this release, we converted all timestamp data to strings when storing dataframes, and did the reverse conversion when loading dataframes. However, you can also avoid the issue by ensuring that timestamp data has a timezone attached to it. We think this is a less invasive change, and so the

SnowflakePandasIOManager

will now attach the UTC timezone to timestamp data that does not already have a timezone. However, if you would like to continue storing timestamp data as strings, or you have already materialized timestamp data and do not want to migrate the corresponding Snowflake table to a table with a TIMESTAMP column, you can set the configuration value

store_timestamps_as_strings=True

. If you have materialized an asset with timestamp data that was stored as strings and would like to migrate the corresponding table so that you can store timestamp data as timestamps, you can follow the instructions in the migration guide

Is there a way I can use the dagster_snowflake resource with the new Python resources api? This is what I would like to do:

from dagster import asset
from dagster_snowflake import SnowflakeResource

@asset
def my_snowflake_thing(snowflake: SnowflakeResource):
    results = snowflake.execute_query("select 1 as foo")
    results_as_df = snowflake.execute_query("select 2 as bar", as_df=True)

    # do things

What would be the best way to extract a data version (ideally generated from an

updated_at

field or similar) from an

observable_source_asset

that has a Snowflake IO manager?

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?