Channels
announcements
ask-marvin
best-practices-coordination-plane
data-ecosystem
data-tricks-and-tips
events
find-a-prefect-job
geo-bay-area
geo-berlin
geo-boston
geo-chicago
geo-colorado
geo-dc
geo-israel
geo-japan
geo-london
geo-nyc
geo-seattle
geo-texas
gratitude
introductions
marvin-in-the-wild
pacc-clearcover-june-12-2023
pacc-may-31-2023
ppcc-may-16-2023
prefect-ai
prefect-aws
prefect-azure
prefect-cloud
prefect-community
prefect-contributors
prefect-dbt
prefect-docker
prefect-gcp
prefect-getting-started
prefect-integrations
prefect-kubernetes
prefect-recipes
prefect-server
prefect-ui
random
show-us-what-you-got
Title
d
Deepanshu Aggarwal
10/21/2022, 5:17 AMif after running a concurrent task run i want to run another task ( after completion of previous concurrent task run) with the output of 1st task in concurrent task run. Is it possible ?
✅ 1
a
Anna Geller
10/21/2022, 11:39 AMthe simplest might be to combine those two into one task
e.g. when I do "extract & load" step without any transformations in between, I'd typically put them into one task since the sole purpose of such task is to get this data into X target (the target is typically even named after source anyway so observability aspect remains)
d
Deepanshu Aggarwal
10/21/2022, 11:42 AMbut what i wanna achieve is
task A (array output)-> Map(task B) -> task C(output of 1st run of task B)
i want the sequential run as well
a
Anna Geller
10/21/2022, 11:43 AMexplore various designs and see what's working best, no right or wrong here
what you suggest seems doable by doing a for loop without submit for task C and another for loop with submit for task B
# pseudo code
for i in task_a_output:
task_b.submit(i, ...)
for i in task_a_output:
task_c(i...)d
Deepanshu Aggarwal
10/21/2022, 11:46 AMdef spends_flow(event):
scheduler_output = run_scheduler(event)
executor_output = run_executor.map(scheduler_output['body_array'])
this is what im doing right now. for concurrent run
But i want to have one more task (not map state) after all the run_executor tasks have run
something like
def spends_flow(event):
scheduler_output = run_scheduler(event)
executor_output = run_executor.map(scheduler_output['body_array'])
run_task_c(executor_output[0])are you suggesting to loop over the output and use task.submit ? would it still be concurrent ?
🙌 1
a
Anna Geller
10/21/2022, 12:07 PMyes exactly! calling a .submit() on a task submits it to a task runner - the default one is concurrent task runner
without .submit() you run things sequentially in a flow run process
d
Deepanshu Aggarwal
10/21/2022, 12:09 PMbut before i run task c i want to make sure that all concurrent runs of task b have completed..
a
Anna Geller
10/21/2022, 12:57 PM.submit().wait() then
to wait for completion and then start sth else
d
Deepanshu Aggarwal
10/21/2022, 12:58 PMalright. ill try this out
🙌 1