@Kevin Kho, I got the apply_map working, but when it runs the task diagrams are on top of one another (first pic) and the flow names seem to just list "Mapped Child X" in most cases (second pic). Is there a way for me to split the mapped flows apart in the diagram or to have the names be more descriptive? As you can see in the bottom two cases I have mapped names going to the sub-task called, but all the various Mapped Child names aren't particularly helpful. These correlate roughly to the inc_or_negate and case statements in your https://docs.prefect.io/core/concepts/mapping.html#complex-mapped-pipelines example (alone with a create_flow_run and wait_for_flow_run in each mapped instance too); is there a way for me to give that more sensical names?

FYI, I just had my first instance of one of the mapped flows failing; it really is harder to track which went wrong without some better naming here.

So I have a suggestion that might help, you can template a name inside the Flow argument by using the

task_args

like this:

from prefect import Flow, task, case, apply_map
from prefect.tasks.control_flow import merge

@task
def inc(x):
    return x + 1

@task
def negate(x):
    return -x

@task()
def is_even(x):
    if x == 2:
        raise ValueError()
    return x % 2 == 0

def inc_or_negate(x):
    cond = is_even(x)
    # If x is even, increment it
    with case(cond, True):
        res1 = inc(x, task_args={"name":"test-{x}"})
    # If x is odd, negate it
    with case(cond, False):
        res2 = negate(x)
    return merge(res1, res2)

with Flow("apply-map example") as flow:
    result = apply_map(inc_or_negate, range(4))

flow.run()

This should at least help with create_flow_run and wait_for_flow_run. The case is special so it doesn’t work for that.

Hmm, if I do that then it looks like I need an f-string.

current_flow = create_flow_run(_flow_name_ = flow_name,_run_name_ = _f_'{datetime.now()} {flow_name} attempt.',_project_name_ = PROJECT_NAME,_task_args_={"name":"Retry: {flow_name}"})

wait_for_flow_run(current_flow, _raise_final_state_=True,_task_args_={"name":"Wait for Retry: {flow_name}"})

in those cases flow_name is one of the mapped values that I pass in to my equivalent of the

inc_or_negate

function

Also, I don’t think you want to f-string in the

create_flow_run

for the

run_name

, there is a chance that gets fixed to registration time

Maybe you can modify wait_for_flow_run by making a wrapper so you can pass in your name to the task call, and then you can use it in the template lol

Ah sorry it should be the keyword argument that is the input. So it should be:

wait_for_flow_run(current_flow, _raise_final_state_=True,_task_args_={"name":"Wait for Retry: {flow_run_id}"})

Like in this doc