https://prefect.io logo
d

Deepanshu Aggarwal

10/21/2022, 5:17 AM
if 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 AM
the 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 AM
but 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 AM
explore 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
Copy code
# 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 AM
Copy code
def 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 PM
yes 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 PM
but 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 PM
alright. ill try this out
🙌 1
3 Views