Sylvain Hazard

11/22/2021, 2:13 PM
Hey again, Got another question, sorry for bothering. How can I get the result of a subflow to pass as input to another task ? I'd like to do that because I have a specific task in my flow that is quite resource hungry and I would like to run it as a separate flow in order to optimize costs. The issue is that the output of this task is currently used downstream and I'd like to do the same with it running as its own flow. I may have missed it in the documentation, but I don't think I saw this use case described. Basically, I'd like to do something like this :
with Flow(name="Parent Flow") as flow:
    docs = Parameters("docs", default=[1, 2, 3])
    small_task = SmallTask().map(docs)
    big_task ="Child flow"), project_name="whatever", parameters=small_task)
    wait_big_task =, raise_final_state=unmapped(True))
    another_small_task = AnotherSmallTask().map(wait_big_task)
but I'm not sure how or where to define the result of a flow in order to use it downstream.

Anna Geller

11/22/2021, 2:28 PM
@Sylvain Hazard in general, flows don’t return any data, but you can get results of a task from a specific flow run. You could use the get_task_run_results task, as shown here:
from prefect import Flow
from prefect.tasks.prefect import create_flow_run, get_task_run_result

with Flow("parent") as parent_flow:
    child_run_id = create_flow_run(flow_name="child")

    child_data = get_task_run_result(child_run_id, "create_some_data-1")
    # At runtime, `child_data` will be `[0, 1, 2, 3, 4]`
Docs: •

Sylvain Hazard

11/22/2021, 2:33 PM
Awesome, thanks so much !
🙌 1