Thread
#prefect-community
    a

    Aaron Pickering

    8 months ago
    Hi everyone, I'm trying to use "SnowflakeQueriesFromFile" in a task and I'm getting a strange error. Not sure how to start debugging this, any ideas? Could it be something to do with the file path?
    "Failed to load and execute Flow's environment: FlowStorageError('An error occurred while unpickling the flow:\n NameError("name \'err\' is not defined")')"
    The task itself is straightforward, it looks like this:
    snowsql_obj = SnowflakeQueriesFromFile(account=SNOWFLAKE_ACCOUNT, user=SNOWFLAKE_USER, password=SNOWFLAKE_PWD, file_path="../../sql/amplitude_raw.sql")
    snowsql_obj.run()
    Anna Geller

    Anna Geller

    8 months ago
    You issue is because of flow storage. This thread may help you. You need to make sure your agent can access both your flow file and your query file. An absolute path would be better than relative (atm you use a relative path). If you need an example for the task:
    from prefect import Flow, Parameter, task
    from prefect.tasks.snowflake import SnowflakeQueriesFromFile
    from prefect.tasks.secrets import PrefectSecret
    
    
    snowflake_query = SnowflakeQueriesFromFile(
        database="DEV", schema="jaffle_shop", autocommit=True, file_path="sample_query.sql"
    )
    
    
    @task(log_stdout=True)
    def print_data(x):
        print(x)  # prints a list of tuples
    
    
    with Flow("snowflake_tasks_example") as flow:
        account = PrefectSecret("SNOWFLAKE_ACCOUNT_ID")
        user = PrefectSecret("SNOWFLAKE_USER")
        pwd = PrefectSecret("SNOWFLAKE_PASS")
        role = PrefectSecret("SNOWFLAKE_ROLE")
        warehouse = Parameter("SNOWFLAKE_WAREHOUSE", default="COMPUTE_WH")
        data = snowflake_query(
            account=account, user=user, password=pwd, role=role, warehouse=warehouse,
        )
        print_data(data)
    
    if __name__ == "__main__":
        flow.run()
    a

    Aaron Pickering

    8 months ago
    Thanks! Thats very helpful. So, I know that my flow storage can be accesssed (because I can run the flow properly without the snowflake task). I guess this means that the sql file cannot be accessed. Since I'm using an S3 bucket for the storage, do I need to place the SQL file in the same s3 bucket?
    Anna Geller

    Anna Geller

    8 months ago
    No, it must be placed on the agent. Afaik this task expects a local file, so S3 path won’t work. Perhaps you can add a task before that and download this query file from S3?
    alternatively, this task can load your query from string https://docs.prefect.io/api/latest/tasks/snowflake.html#snowflakequery
    a

    Aaron Pickering

    8 months ago
    Thanks for the info, I will check out the resources. Much appreciated!
    FYI @Anna Geller you were correct. It was a problem with the path and the docker build. The given path should match the location of the file path on the agent. Works well now, thanks again!