in prefect 2 0 what is the best way to chain results between

Question

in prefect 2 0 what is the best way to chain results between map tasks Specifically how should errors on specific iterations be handled if the remaining results are still desired This seemed seemless in prefect 1 x but I am running into issues while migrating to 2 0

Khuyen Tran · Answer

You could add ` submit ` to each task to make the tasks inside your flow run asynchronously <https docs prefect io concepts task runners using a task runner>

Neil Natarajan · Answer

I see so task mapping in 2 0 is actually just synchronous And submit is actually async now

Neil Natarajan · Answer

< prefect admin>

Jeff Hale · Answer

Check out the examples of mapping <https docs prefect io api ref prefect tasks prefect tasks Task map|here > Will submit to the task runner automatically

Nate · Answer

< Neil Natarajan> it should execute concurrently by default for example ``` task def square x int gt int return x 2 flow def my flow numbers = i for i in range 10 squared numbers = square map numbers my flow ``` will give logs like ```18 27 44 809 | INFO | prefect engine Created flow run elegant coua for flow my flow 18 27 45 846 | INFO | Task run square 2b9b4656 0 Finished in state Completed 18 27 45 904 | INFO | Task run square 2b9b4656 2 Finished in state Completed 18 27 45 912 | INFO | Task run square 2b9b4656 8 Finished in state Completed 18 27 45 915 | INFO | Task run square 2b9b4656 5 Finished in state Completed 18 27 45 925 | INFO | Task run square 2b9b4656 3 Finished in state Completed 18 27 45 936 | INFO | Task run square 2b9b4656 6 Finished in state Completed 18 27 45 936 | INFO | Task run square 2b9b4656 7 Finished in state Completed 18 27 45 957 | INFO | Task run square 2b9b4656 4 Finished in state Completed 18 27 46 015 | INFO | Task run square 2b9b4656 9 Finished in state Completed 18 27 46 140 | INFO | Task run square 2b9b4656 1 Finished in state Completed 18 27 46 252 | INFO | Flow run elegant coua Finished in state Completed All states completed ```

Neil Natarajan · Answer

Thank you all for the help

Neil Natarajan · Answer

Circling back to the original question if there are two map tasks that chain input like this example ``` task name= map task def map task x y if x == 2 raise Exception return x + y task name= map task 2 def map task 2 nr print f Map Task 2 time time nr time sleep 5 print f Map Task 2 time time nr flow name= outer flow task runner=DaskTaskRunner def outer flow numbers static num futures = map task map x=numbers y= static num len numbers map task 2 map nr=futures if name == main workflow result = outer flow 1 2 3 4 5 2 ```

Neil Natarajan · Answer

In this example I still want the results for iterations of x = 1 3 4 5 but when I run this the entire workflow fails and none of the other iterations run

Khuyen Tran · Answer

It is because `map task 2 map nr=futures ` is blocked by `futures = map task map x=numbers y= static num len numbers ` For example ```from prefect import flow task from time import sleep task name= Task 1 def task 1 sleep 5 task name= Task 2 def task 2 sleep 2 flow def my flow task 1 task 2 if name == main my flow ``` Output ```09 37 44 528 | INFO | prefect engine Created flow run prehistoric hyrax for flow my flow 09 37 45 275 | INFO | Flow run prehistoric hyrax Created task run Task 1 1619c4ff 0 for task Task 1 09 37 45 276 | INFO | Flow run prehistoric hyrax Executing Task 1 1619c4ff 0 immediately 09 37 50 594 | INFO | Task run Task 1 1619c4ff 0 Finished in state Completed 09 37 50 679 | INFO | Flow run prehistoric hyrax Created task run Task 2 49f2f021 0 for task Task 2 09 37 50 680 | INFO | Flow run prehistoric hyrax Executing Task 2 49f2f021 0 immediately 09 37 52 938 | INFO | Task run Task 2 49f2f021 0 Finished in state Completed 09 37 53 041 | INFO | Flow run prehistoric hyrax Finished in state Completed All states completed ``` Task 2 will wait for Task 1 to execute

Khuyen Tran · Answer

But adding `submit` will make the task run concurrently ```from prefect import flow task from time import sleep task name= Task 1 def task 1 sleep 5 task name= Task 2 def task 2 sleep 2 flow def my flow task 1 submit task 2 submit if name == main my flow ``` ```09 38 35 746 | INFO | prefect engine Created flow run amiable marmot for flow my flow 09 38 36 462 | INFO | Flow run amiable marmot Created task run Task 1 1619c4ff 0 for task Task 1 09 38 36 463 | INFO | Flow run amiable marmot Submitted task run Task 1 1619c4ff 0 for execution 09 38 36 603 | INFO | Flow run amiable marmot Created task run Task 2 49f2f021 0 for task Task 2 09 38 36 603 | INFO | Flow run amiable marmot Submitted task run Task 2 49f2f021 0 for execution 09 38 38 850 | INFO | Task run Task 2 49f2f021 0 Finished in state Completed 09 38 41 711 | INFO | Task run Task 1 1619c4ff 0 Finished in state Completed 09 38 41 810 | INFO | Flow run amiable marmot Finished in state Completed All states completed ```

Khuyen Tran · Answer

So to make sure your second task still runs when the first task fails do something like this ``` task name= map task def map task x y if x == 2 raise Exception return x + y task name= map task 2 def map task 2 nr print f Map Task 2 time time nr time sleep 5 print f Map Task 2 time time nr flow name= outer flow def outer flow numbers static num futures = map task submit x=num y=static num for num in numbers map task 2 submit nr=num for num in futures if name == main workflow result = outer flow 1 2 3 4 5 2 ```

Neil Natarajan · Answer

Understood that requires then a lot of code change when migrating from 1 gt 2 If the same behavior could be added to the map function that would make the migration and experience smoother imo

Neil Natarajan · Answer

Thank you for the help

Zanie · Answer

Hey Neil can you open an enhancement request on GitHub We should be able to find a way to support chained mapping with errors

Neil Natarajan · Answer

Yes I can totally do that

Neil Natarajan · Answer

Thank you