https://prefect.io logo
Title
a

Arsenii

02/18/2020, 6:22 AM
Hey all, a quick question Let's say I
map
a function A over a list L, and then want another function B to continue working on elements of L in a parallel manner after A has finished. Is my understanding correct, that in order to implement this, function A should basically return the same element it got as an input, and pass it to B? i.e. in Python/pseudocode
def func_A(list_element):
    do_stuff(list_element)
    return list_element

def func_B(list_element):
    list_element+=2 

with Flow(...):
    L = Parameter("list", default=[1,2,3])
    first_step = func_A.map(L)
    second_step = func_B.map(first_step)
This seems to work, but I wonder whether this is the most efficient way to do this. I also noticed that, if I have several consecutive `map`s set up in the above manner
A -> B -> C -> D-> ...
, it will not begin the next step until the previous step has finished executing for all mapped elements. Is there a way to go around this? Is it because I'm using LocalExecutor and not something like DaskExecutor? Thanks!
a

Alex Goodman

02/18/2020, 3:24 PM
The short answer is we don't support this yet. We're currently working towards refactoring Mapping and have captured this as a desired feature to work towards (https://github.com/PrefectHQ/prefect/issues/2041). A workaround in the meantime would be to refactor tasks to include all of the "downstream" work into a single task:
import prefect


def func_A(element):
    return element * 2


def func_B(element):
    return element + 4


@prefect.task
def func_A_B(element):
    return func_B(func_A(element))


with prefect.Flow("thing") as flow:
    L = prefect.Parameter("list", default=[1, 2, 3])
    result = func_A_B.map(L)


flow.run()
👍 1