Channels
#prefect-community
Title
# prefect-community
d
David Elliott
10/10/2022, 4:36 PMHey all, wondering if anyone has any idea on how to generate/write a flow dynamically in 2.0? e.g add tasks + dependencies from another graph object…
In 1.0 it used to be possible to just instantiate a flow object and then add tasks / dependencies to the flow object (e.g whilst iterating over a dict or a networkX graph), however now flows are decorators on functions I’m struggling to see how I’d do that? I’ll attach some example code in the 🧵 that shows what I’m trying to achieve. Thanks for any suggestions in advance!
✅ 1
eg code.py
z
Zanie
10/10/2022, 4:42 PM
Perhaps something like this?
Copy code
from prefect import flow, Flow, task
@task
def foo(x):
return 1
def build_flow() -> Flow:
tasks = [foo.with_options(name=f"task_{i}") for i in range(10)]
@flow
def my_flow():
for task in tasks():
task("test")
return my_flow
my_flow = build_flow()👀 1
I didn’t implement the network, but do you see how you could do so?
d
David Elliott
10/10/2022, 4:56 PMThanks! I’m struggling to see how to pass in the
upstreamsI think the difficulty is I need to be able to pass in the result of one task (a PrefectFuture) as an argument to the next task - which I’m not sure is possible at python runtime..? ie previously I could add the task to the flow with
add_task()set_dependency() Copy code
task1 = foo(name="t1", upstreams=[])
task2 = foo(name="t2", upstreams=[task1])z
Zanie
10/10/2022, 5:47 PM
Untested, but like…
Copy code
my_network_graph = {
"task_1" : [],
"task_2" : [],
"task_3" : ["task_1", "task_2"],
"task_4" : ["task_3"],
"task_5" : ["task_4"],
}
from prefect import flow, Flow, task
@task
def foo(x):
return 1
def build_flow(network) -> Flow:
tasks = {}
for name in my_network_graph.keys():
tasks[name] = foo.with_options(name=name)
@flow
def my_flow():
for task, upstreams in tasks.items():
task.submit(wait_for=[tasks[name] for name in upstreams])
return my_flow
my_flow = build_flow(my_network_graph)👀 1
If you’re trying to actually pass the data you’re basically going to have to implement some sort of DAG walker for your network which sounds painful.
Like..
• Run all tasks without upstreams, store their futures
• Find all tasks with upstreams fulfilled, run them and store their futures
• Repeat…
d
David Elliott
10/10/2022, 6:11 PMI’m afraid the above code doesn’t work sadly as it’s passing the un-called tasks in the
wait_for=[]task.submit()open(flow.py,w) as ff.write()I modified your above example in testing, this is what I ended up with:
Copy code
my_network_graph = {
"task_1" : [],
"task_2" : [],
"task_3" : ["task_1", "task_2"],
"task_4" : ["task_3"],
"task_5" : ["task_4"],
}
from prefect import flow, Flow, task
@task
def foo(task_name):
print(task_name)
def build_flow(network) -> Flow:
tasks = {}
for name in network.keys():
tasks[name] = foo.with_options(name=name)
@flow(name="demo_dynamic")
def my_flow():
for task_name, task in tasks.items():
upstream_task_names = network[task_name]
upstream_tasks = [tasks[u] for u in upstream_task_names]
task.submit(task_name=task_name, wait_for=upstream_tasks)
return my_flow
my_flow = build_flow(my_network_graph)
if __name__ == "__main__":
my_flow()z
Zanie
10/10/2022, 6:14 PM
Ah my bad yeah you can pass futures as long as your graph doesn’t have references out of order
Copy code
@flow
def my_flow():
futures = {}
for task, upstreams in tasks.items():
futures[name] = task.submit(wait_for=[futures[name] for name in upstreams])Otherwise it’s more complicated. Can you explain why you’re trying to define your whole structure before beginning work? Dynamic flows are way simpler in v2, you can literally have a for loop that creates tasks in the flow itself and uses runtime logic. I’m struggling to see the value of this particular form of dynamicism.
d
David Elliott
10/10/2022, 6:25 PMYeah for sure - essentially we have an existing static DAG (which is defined in our git repo as a yaml) which needs executing. In Prefect 1.0 we convert that yaml to a prefect flow at CI build time (adding nodes + edges) and register it, then Prefect runs that DAG for us. We’re essentially forced to move to Prefect 2.0 asap as we’re up at 2850 DAG nodes and your upper limit is 3000 (GQL limitations) and so I’m trying to port this existing static-DAG setup over to 2.0 as soon as poss.
Like you say, I could try and write something to execute the DAG at runtime (ie walk the network in a loop submitting tasks for execution as & when their upstreams complete) but I was hoping to avoid needing to do that, given we already know the structure of our DAG and just need it to be executed…!
In terms of the tasks themselves, they’re purely “run this sql file” tasks, no data needs to be passed between them, just need to maintain the dependencies per our dag file
z
Zanie
10/10/2022, 6:35 PM
Interesting! I've got some high priority work so I can't look into a more complete example right now but we should be able to handle this without too much complexity
🙌 1
d
David Elliott
10/10/2022, 9:08 PMThanks! lmk if you have any other thoughts on it 🙂
z
Zanie
10/10/2022, 9:10 PM
I'm also under the impression that we've about doubled the maximum size supported in v1
I'm not sure what limits you're encountering / where, but I recently did some work on Cloud to bump the maximum size of flows.
d
David Elliott
10/12/2022, 10:36 AMThere’s a hard limit, something to do with the GQL backend / api rate limits??, of 3000 nodes in a given flow - initially we hit the 2500 limit and your team upped it to 3k, but said it basically can’t go any higher
z
Zanie
10/12/2022, 4:49 PM
Ah yes sorry I raised it from 2500 to 3000
I think we contemplated bumping it higher, but we are indeed concerned about encountering performance issues after monitoring the earlier bump.
👍 1
3 Views