In Prefect 3.x, the best way is usually to avoid

task.map()

for this case and use a manual

.submit()

loop instead. Why:

task.map()

eagerly materializes iterables, so wrapping a generator/iterator/stream-like object will effectively do the equivalent of

list(...)

internally. That means it is not the right tool for lazy or streaming-style iteration. Recommended pattern

python
from prefect import flow, task
from prefect.futures import wait

@task
def process_item(item):
    return item * 2

@flow
def process_iterable(items):
    futures = []

    for item in items:
        futures.append(process_item.submit(item))

    wait(futures)
    return [f.result() for f in futures]

This gives you: - lazy consumption of the iterable - parallel task execution - no need to resolve the whole iterable up front If the iterable is very large, you may also want to batch submissions so you don't keep an unbounded number of futures in memory:

python
from itertools import islice
from prefect import flow, task
from prefect.futures import wait

@task
def process_item(item):
    return item * 2

def batched(iterable, size):
    it = iter(iterable)
    while batch := list(islice(it, size)):
        yield batch

@flow
def process_iterable_in_batches(items, batch_size=100):
    results = []

    for batch in batched(items, batch_size):
        futures = [process_item.submit(item) for item in batch]
        wait(futures)
        results.extend(f.result() for f in futures)

    return results

For subflows specifically Flows do not have a

.map()

equivalent. If you want to run a subflow for each item: - sync subflows will run one-by-one - async subflows can be run concurrently with

asyncio.gather(...)

But if your goal is simply "do work per item without materializing the iterable first," tasks +

.submit()

are usually the better fit. Example with subflows:

python
import asyncio
from prefect import flow

@flow
async def child(item):
    return item * 2

@flow
async def parent(items):
    coros = []
    for item in items:
        coros.append(child(item))
    return await asyncio.gather(*coros)

Caveat Even though a

.submit()

loop avoids eager materialization of the input iterable, if you keep every future/result forever, memory can still grow. That's why batching is a good pattern for very large or unbounded iterables. Short answer: use a

for

loop with

.submit()

instead of

task.map()

. Relevant docs: - Run work concurrently - Tasks - Flows If you want, I can also show a pattern for: - streaming results as they complete, or - bounded concurrency over an iterator.