Quick clarifier: are you using Prefect 1.x?

task.map()

is a 1.x feature; in Prefect 3.x there’s no

task.map()

and you’d typically submit tasks in a loop. Assuming Prefect 3.x, if each task run returns a list and you want a single flattened list, create a small downstream task to flatten after all runs complete:

from itertools import chain
from prefect import flow, task

@task
def produce(n):
    # returns a list per input
    return [n, n + 1]

@task
def flatten(list_of_lists):
    return list(chain.from_iterable(list_of_lists))

@flow
def main():
    # launch in parallel
    futures = [produce.submit(i) for i in range(5)]
    # collect results
    lists = [f.result() for f in futures]
    # flatten
    flat_future = flatten.submit(lists)
    return flat_future.result()

if __name__ == "__main__":
    print(main())

If you are on Prefect 1.x and truly using

task.map()

, share a small snippet and I’ll tailor it. The idea is the same: run a downstream task that takes the list-of-lists and flattens it; the only wrinkle is making sure that downstream task is not auto-mapped.