Channels
#prefect-community
Title
# prefect-community
z
Zach Jablons
03/10/2023, 1:33 AMI'm also trying to figure out how I can defer execution of potentially expensive tasks in a flow unless requested, or to only execute a given task and its parents. 🧵
Copy code
import time
import sys
from prefect import flow, task
from prefect.tasks import task_input_hash
@task(cache_key_fn=task_input_hash, refresh_cache=True)
def a():
print("executing a")
time.sleep(1)
return 2
@task(cache_key_fn=task_input_hash, refresh_cache=True)
def b(result_from_a):
print("executing b")
time.sleep(10)
return result_from_a + 2
@task(cache_key_fn=task_input_hash, refresh_cache=True)
def c(result_from_a):
print("executing c")
return result_from_a + 4
@flow
def test1():
print("Submitting a")
res_a = a.submit()
print("Submitting b")
res_b = b.submit(res_a, wait_for=res_a)
print("Submitting c")
res_c = c.submit(res_a, wait_for=res_a)
return {
'b':res_b,
'a':res_a,
'c':res_c
}
if __name__ == "__main__":
total_tic = time.time()
nodes = test1()
print(nodes[sys.argv[1]].result())
print(f"Took {time.time() - total_tic} seconds")What I'm hoping to be able to do here is to run
test.py cbtest.py bbcThis is the output of
test.py c Copy code
18:58:30.763 | INFO | prefect.engine - Created flow run 'gorgeous-capybara' for flow 'test1'
Submitting a
Submitting b
Submitting c
18:58:30.917 | INFO | Flow run 'gorgeous-capybara' - Created task run 'a-0' for task 'a'
18:58:30.918 | INFO | Flow run 'gorgeous-capybara' - Submitted task run 'a-0' for execution.
/home/zach/.pyenv/versions/prefect_test/lib/python3.8/site-packages/prefect/engine.py:1215: UserWarning: A 'PrefectFuture' from a task
call was cast to a boolean; did you mean to check the result of the task instead? e.g. `if my_task().result(): ...`
if wait_for:
18:58:30.961 | INFO | Flow run 'gorgeous-capybara' - Created task run 'c-0' for task 'c'
18:58:30.962 | INFO | Flow run 'gorgeous-capybara' - Submitted task run 'c-0' for execution.
executing a
18:58:31.011 | INFO | Flow run 'gorgeous-capybara' - Created task run 'b-0' for task 'b'
18:58:31.011 | INFO | Flow run 'gorgeous-capybara' - Submitted task run 'b-0' for execution.
18:58:32.048 | INFO | Task run 'a-0' - Finished in state Completed()
executing b
executing c
18:58:32.145 | INFO | Task run 'c-0' - Finished in state Completed()
18:58:42.167 | INFO | Task run 'b-0' - Finished in state Completed()
18:58:42.212 | INFO | Flow run 'gorgeous-capybara' - Finished in state Completed()
6
Took 11.928142786026001 secondsNote that
bsubmitresulta
Austin Weisgrau
03/10/2023, 5:20 PMA future is returned immediately and the task is sent to the runner for execution, but calling future.result() won't return the result until the task has finished executing
z
Zach Jablons
03/10/2023, 5:21 PMIs there any way to get the runner to delay execution until requested?
a
Austin Weisgrau
03/10/2023, 5:22 PMI think if you don't want b to run unless requested, you could parametrize the flow:
Copy code
@flow
def test1(run_b: bool):
print("Submitting a")
res_a = a.submit()
if run_b:
print("Submitting b")
res_b = b.submit(res_a, wait_for=res_a)
print("Submitting c")
res_c = c.submit(res_a, wait_for=res_a)
result = {
'a':res_a,
'c':res_c
}
if run_b:
result.update({'b': res_b})
return resultz
Zach Jablons
03/10/2023, 5:23 PMYeah, I'm hoping to avoid building that kind of logic for any arbitrary DAG
a
Austin Weisgrau
03/10/2023, 5:26 PMSo if I understand, you don't want to need to tell test1() up front if b is required to run, but have it be inferred by whether or not res_b is used downstream, is that right?
z
Zach Jablons
03/10/2023, 5:26 PMEssentially, yes - I'm hoping to be able to construct a DAG in the flow and have it execute the portions of the DAG that are requested on invocation
(I've built something that does that in the past, but it doesn't have all the cool features Prefect has 🙂 )
a
Austin Weisgrau
03/10/2023, 5:27 PMI'm not sure I understand the difference between what you mean by "requested on invocation" and adding a parameter to the dag
z
Zach Jablons
03/10/2023, 5:28 PMBasically the extension to your example solution for an arbitrary DAG is to have a dictionary of tasks, and their dependencies - but then you have to recursively go through the dependencies and get their outcomes as well
I think that's possible to do, but I was just wondering if there was a way to have Prefect handle that on its own
a
Austin Weisgrau
03/10/2023, 6:03 PMlike a lazy evaluation for task futures 🤔
z
Zach Jablons
03/10/2023, 6:08 PMSomething like that would be really neat, especially if it can work with asynchronous execution while maintaining the DAG ordering correctly.
b
Bianca Hoch
04/03/2023, 7:05 PMHey @Zach Jablons ! It was good talking to you earlier. Feel free to drop the example we reviewed here if it is more up-to-date
z
Zach Jablons
04/03/2023, 7:46 PMSure thing, I think the one I posted above is along the same lines but the example I was showing is below:
Copy code
import time
import sys
from prefect import flow, task
from prefect.tasks import task_input_hash
@task(cache_key_fn=task_input_hash, refresh_cache=True)
def a():
print("executing a")
time.sleep(1)
return 2
@task(cache_key_fn=task_input_hash, refresh_cache=True)
def b(result_from_a):
print("executing b")
time.sleep(10)
return result_from_a + 2
@task(cache_key_fn=task_input_hash, refresh_cache=True)
def c(result_from_a):
print("executing c")
return result_from_a + 4
@flow
def flow_a():
return a.submit()
@flow
def flow_b(result_from_a):
return b.submit(result_from_a)
@flow
def flow_c(result_from_a):
return c.submit(result_from_a)
@flow
def main_flow():
res_a = flow_a()
res_b = flow_b(res_a)
res_c = flow_c(res_a)
return {
'a': res_a,
'b': res_b,
'c': res_c
}
if __name__ == "__main__":
total_tic = time.time()
nodes = main_flow()
print(nodes[sys.argv[1]].result())
print(f"Took {time.time() - total_tic} seconds")t
Taylor Curran
04/03/2023, 9:51 PMHi @Zach Jablons on the topic of caching, I wonder if this custom caching example might help:
https://github.com/taylor-curran/flow-patterns/blob/main/flows/caching/custom_cache_key_function.py
A list of tasks is passed as a parameter and those tasks are skipped thanks to the function static_cache_key.
z
Zach Jablons
04/03/2023, 10:01 PMOh that's interesting, but doesn't quite solve my issue - ideally my user would specify the tasks they want outputs for, and Prefect would figure out what tasks need to be executed to achieve that - rather than the user having to determine which list of tasks (which, it could be a lot) should be skipped (and then specify that in the hash key)
👍 1
t
Taylor Curran
04/03/2023, 10:22 PMDo you think it might work if the cache function checked for conditions instead of checked for user input?
z
Zach Jablons
04/03/2023, 10:26 PMWhat do you mean by conditions?
t
Taylor Curran
04/03/2023, 10:28 PMconditions that would indicate wether or not the task needs to run
But hmm im jumping to conclusions, let me read through this thread a little more thoroughly first 😅
I see that the condition would be wether or not the result is used
Like
Copy code
@flow
def main_flow():
res_a = flow_a()
res_b = flow_b(res_a)
res_c = flow_c(res_a)
return {
'a': res_a,
'b': res_b,
'c': res_c
}flow_b'b': res_bPrefect to my knowledge does not have something like that out of the box. But I’ll ask around — someone might have an idea. Also this page in the docs is a good overview of results and futures in Perfect our very own @Zanie wrote them.
z
Zach Jablons
04/04/2023, 2:00 PMYeah, I think the key thing that I want to be able to do here is define the DAG separately from executing the DAG, and this is an example of why.
The ideal interface I'm trying to achieve here is that our team would only have to define a) what configuration they want the DAG to run with and b) what nodes they want results from, and then Prefect would take care of the rest w/o executing anything unnecessarily
z
Zanie
04/04/2023, 6:07 PM
👋 Hey Zach, I’m going to think about this feature but it’s probably not something that can be hacked together using our existing interfaces.
You could of course write a generic flow that walks a DAG and executes it dynamically then define the DAG separately but to use Prefect itself for lazy execution we’d need to make some substantial changes to how submission for execution happens.
z
Zach Jablons
04/04/2023, 6:11 PMOk, that's helpful to know - thanks