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"
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,
if __name__ == "__main__":
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?
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?