Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Is there a good way to have a task generate two lists (e.g. error_items, and a processed_items as output and map each list to a different downstream task?

Hi <@U011KE2QA5V>! You can index the output of a task like this:

```from prefect import task, Flow
@task
def two_lists():
    return {
        'list_one': [1, 2, 3],
        'list_two': [101, 102, 103]
    }

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

with Flow('example') as flow:
    lists = two_lists()
    x = add_one.map(lists['list_one'])
    y = add_one.map(lists['list_two'])

state = flow.run()
assert state.result[x].result == [2, 3, 4]
assert state.result[y].result == [102, 103, 104]```

My example returns a dictionary, but if you return a list or tuple and index as `lists[0], lists[1]` it’ll also work

this also works if you don’t want to return a `dict` (I use something similar in my flow)
```both_items = task_that_generates_two_lists(..) # returns two lists

error_items, processed_items = both_items[0], both_items[1]

error_items_task.map(error_items)
processed_items_task.map(processed_items)```

Awesome, thanks guys! What does this look like using the imperative api? do I just do a `error_items_task.set_upstream(two_lists['list_one'], key="data", mapped=True)`