Help needed to get dataframe loaded into Snowflake

[dagsir[bot]]

I am trying to get a dataframe loaded into snowflake. This is approximately what our code looks like. I have tried to follow the directions closely.

Still when I run the code nothing shows up in snowflake and I think it's still using filesystem as an IO manager. Any ideas?

Snippet of logs in a reply

---

The question was originally asked in Dagster Slack.

[jamiedemaria]

Logs from the original slack message:

2023-11-07 15:32:05 +0700 - dagster - DEBUG - all_assets_job - fca3f066-76c3-40a3-aaf2-664e9624b26e - 34699 - enriched_user_feedbacks - STEP_OUTPUT - Yielded output "result" of type "DataFrame". (Type check passed).
2023-11-07 15:32:05 +0700 - dagster - DEBUG - all_assets_job - fca3f066-76c3-40a3-aaf2-664e9624b26e - enriched_user_feedbacks - Writing file at: /Users/jeremiah/code/sn_dagster/tmpl6fuq9uy/storage/enriched_user_feedbacks/2023-11-06 using PickledObjectFilesystemIOManager...
2023-11-07 15:32:05 +0700 - dagster - DEBUG - all_assets_job - fca3f066-76c3-40a3-aaf2-664e9624b26e - 34699 - enriched_user_feedbacks - ASSET_MATERIALIZATION - Materialized value enriched_user_feedbacks.
2023-11-07 15:32:05 +0700 - dagster - DEBUG - all_assets_job - fca3f066-76c3-40a3-aaf2-664e9624b26e - 34699 - enriched_user_feedbacks - HANDLED_OUTPUT - Handled output "result" using IO manager "io_manager"
2023-11-07 15:32:05 +0700 - dagster - DEBUG - all_assets_job - fca3f066-76c3-40a3-aaf2-664e9624b26e - 34699 - enriched_user_feedbacks - STEP_SUCCESS - Finished execution of step "enriched_user_feedbacks" in 12.85s.

[jamiedemaria]

Based on the logs, it does look like the filesystem IO manager is being used. However, the code snippet you shared looks correct. I would try reloading your code location in the UI. Under the Deployment tab, there will be a button on the far right next to your deployment named. The button says Reload.

[jamiedemaria]

Try adding a defs = in front of your Definitions instantiation too. I didn't notice that before in the code snippet you shared

defs = Definitions(
    assets=[enriched_user_feedbacks],
    resources={
        "io_manager": SnowflakePandasIOManager(
            account=EnvVar('SNOWFLAKE_ACCOUNT'),  # required
            user=EnvVar('SNOWFLAKE_USER'),  # required
            password=EnvVar('SNOWFLAKE_PASSWORD'),  # password or private key required
            database=EnvVar('SNOWFLAKE_DATABASE'),  # required
            role=EnvVar('SNOWFLAKE_ROLE'),  # optional, defaults to the default role for the account
            schema=EnvVar('SNOWFLAKE_SCHEMA'),  # optional, defaults to PUBLIC
        )
    },
)

[JeremiahR]

Ok I did that, it still didn't work. Just to check I renamed the io_manager to snowflake_io_manager and the logs still report that it is using io_manager (the filesystem one).

Here is the code (biz logic removed) https://gist.github.com/JeremiahR/2f74242c1dd43ed00d7d659d32eaac22

There is no dagster specific code missing from this snippet. So I wonder: Do I need to be defining an asset_job? Do I need to be sure to import or define other things carefully?

The defs part makes me think we might be missing some magic code here.

Thank you for any help.

[JeremiahR]

Just to be clear I reloaded and re-ran a few times. Same bug.

[jamiedemaria]

In the new snippet, since you renamed the snowflake io manager to have the key "snowflake_io_manager" you need to tell the enriched_user_feedbacks asset to use that io manager (otherwise it defaults to using the io manager with the key "io_manager")

@asset(
    partitions_def=DailyPartitionsDefinition(
        start_date=config.SKIP_CHUNK_PARTITION_START_DATE
    ),
    metadata={"partition_expr": "TO_TIMESTAMP(CREATED_DATE::INT)"},
    io_manager_key="snowflake_io_manager"
)
def enriched_user_feedbacks(context) -> DataFrame:
    """
    Download latest user feedbacks data set for the latest date
    """
    ...
    return df

There shouldn't be any other "magic" code. The order of imports shouldn't matter or anything like that.

Once you try updating the io manager key on the asset, if that doesn't work would you be able to put together a simple code snippet that I could also run so that I can debug with some real code. You could use the example iris asset from the tutorial https://docs.dagster.io/integrations/snowflake/using-snowflake-with-dagster

[JeremiahR]

Resolved, thank you. I also had to figure out that because we are developing a module, defs= should be defined in the top level module init in order for the asset to see it. Either way thank you for the help.