👋 Hi, I’m new here, trialing Prefect 2.0, and have a couple of question. I’m writing a flow to copy data into a bunch of Snowflake tables. I wanted to use the Snowflake connection instead of managing a connection myself. Seems like

snowflake_query

and

snowflake_multiquery

are their own tasks, so I can’t use them as part of a different task? Which means that my flow looks like this:

do_some_stuff()
for table_name in tables:
    queries = build_queries(table)
    snowflake_multiquery(queries)

These tasks are all running sequentially, but there are a lot of tables so I’d like the

snowflake_multiquery

tasks for each table to run concurrently. How can I make that happen?

I think you can use list comprehension instead:

[snowflake_multiquery(x) for x in queries]

and that might be parallel. You can use a task in another task by calling

task.fn

but it’s just the python under the hood, not a task with retries/caching/checkpointing

To clarify — each of the query in

queries

needs to be run in order (they’re, like, “create table x”, “copy into table x”, etc) but the queries for each table that I’m concerned about should be run concurrently — we can load data into table X concurrently with loading into table Y and table Z. So maybe a list comprehension like this would work?

[snowflake_multiquery(build_queries(t) for t in tables]

Though I don’t understand how that’s functionally different from the for loop I posted above. Can you explain if/why that list comprehension would have tasks running concurrently but not the for loop?

I think the issue might be overwriting queries each time. Could you try:

do_some_stuff()
for table_name in tables:
    snowflake_multiquery(build_queries(table))

Also, can you make sure you are using the ConcurrentTaskRunner? I think the default may have changed to SequentialTaskRunner

logs indicate it is using the ConcurrentTaskRunner. Trying your suggestion, one sec

Are you using a subflow somewhere?

Nope — it’s a pretty simple flow I think

Let me ask a teammate

If the solution is to update my

build_queries()

task to build the queries and then run them with

snowflake_multiquery.fn()

, that should be fine — I mostly wanted to avoid having to work with the snowflake python connector directly

We require you to call

.submit(…)

on your tasks to send them to the task runner now