Thanks for sharing the snippet! A couple quick fixes and patterns to make this work well in Prefect 3.x: Key points - Task.map exists in Prefect 3.x and is the recommended way to run the same task over a list of inputs concurrently. - Use a task runner on your flow (ThreadPoolTaskRunner for I/O-bound, ProcessPoolTaskRunner for CPU-bound) to actually get parallelism. - Your chunking logic has a couple bugs (it appends too early and appends inside the loop). See a corrected version below. - Avoid naming collisions (your

@task def transform(...)

recursively calls itself). Use distinct names like

transform_batch

or

transform_one

. A corrected “batched” version (map over chunks)

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

# Simple chunker
def chunked(seq, size):
    for i in range(0, len(seq), size):
        yield seq[i:i+size]

# Replace with your real ID source
def get_ids_to_process() -> list[str]:
    return [f"id-{i}" for i in range(1, 51)]

@task
def get_id_chunks(chunk_size: int) -> list[list[str]]:
    ids = get_ids_to_process()
    return list(chunked(ids, chunk_size))

@task
def load_for_ids(ids: list[str]) -> list[dict]:
    return [do_load(i) for i in ids]  # implement do_load()

@task
def transform_batch(batch: list[dict]) -> list[dict]:
    return [do_transform(item) for item in batch]  # implement do_transform()

@task
def unload_batch(batch: list[dict]) -> None:
    for item in batch:
        write_to_s3(item)  # implement write_to_s3()

@flow(task_runner=ThreadPoolTaskRunner(max_workers=16))  # or ProcessPoolTaskRunner for CPU-bound
def my_flow():
    id_chunks = get_id_chunks(10)  # runs inline and returns list[list[str]]
    loaded = load_for_ids.map(id_chunks)
    transformed = transform_batch.map(loaded)
    unload_batch.map(transformed)

If each step is naturally “per ID,” map per item instead (often simpler)

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

def get_ids_to_process() -> list[str]:
    return [f"id-{i}" for i in range(1, 51)]

@task
def load_one(i: str) -> dict:
    return do_load(i)

@task
def transform_one(d: dict) -> dict:
    return do_transform(d)

@task
def unload_one(d: dict) -> None:
    write_to_s3(d)

@flow(task_runner=ThreadPoolTaskRunner(max_workers=16))
def my_flow():
    ids = get_ids_to_process()
    loaded = load_one.map(ids)
    transformed = transform_one.map(loaded)
    unload_one.map(transformed)

Notes - Choose ThreadPoolTaskRunner for I/O-bound work (API calls, S3, DB), ProcessPoolTaskRunner for CPU-bound transforms. - If you don’t need orchestration for

get_id_chunks

, you can make it a plain function instead of a task. - If you need to cap concurrency (e.g., API limits), you can control it via the task runner’s

max_workers

or Prefect concurrency limits (tags/global). If you share whether your workload is I/O- or CPU-bound and any rate limit constraints, I can recommend exact task runner settings and a final pattern.

Great question — what you’re seeing is expected. - Prefect will make each downstream task (transform/unload) eligible to run as soon as its upstream (load) finishes. - However, with a single task runner and lots of ready “load” work, the pool is usually saturated by load tasks first. That often looks like “do all loads (16 at a time), then transforms, then unloads” even though the graph technically allows pipelining. If you want true “16 pipelines” (each worker does load -> transform -> unload for an ID before moving on), the simplest and most reliable approach is to compose the three steps into a single task (or subflow) and map that:

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def process_one(i: str):
    d = do_load(i)
    t = do_transform(d)
    write_to_s3(t)

def get_ids_to_process() -> list[str]:
    return [f"id-{i}" for i in range(1, 201)]

@flow(task_runner=ThreadPoolTaskRunner(max_workers=16))
def my_flow():
    ids = get_ids_to_process()
    process_one.map(ids)

Other options if you want to keep separate tasks: - Apply a concurrency limit to the load task (e.g., via tags) so it can’t monopolize the pool; this leaves capacity for transform/unload to run as each load finishes. - Use a subflow that chains the three tasks per ID, then map the subflow across IDs. - Use a sliding-window pattern with manual

.submit

and

as_completed

to prioritize downstream work, but that’s more complex than composing into one task. TL;DR: Your observation is normal; for strict per-item pipelining, map a composite task (or subflow) so each worker processes a full item pipeline end-to-end.

Short answer: There isn’t a built-in “pipeline width” knob for mapped chains with the ThreadPoolTaskRunner. By default, mapped tasks are all submitted eagerly, and the thread pool will happily run the earlier-submitted stage (e.g., stage1) until workers free up. Even though each downstream mapped item depends only on its corresponding upstream item, the pool ordering can make it look like “all of stage1, then stage2”. Good news: you can get the “N items go end-to-end” behavior with a couple of patterns: Options to get eager, bounded pipelining 1) Collapse the per-item pipeline into a single mapped task - Ensures each item runs stage1 → stage2 → sink in one worker, with max N concurrent items.

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def stage1(x): ...
@task
def stage2(y): ...
@task
def sink(z): ...

@task
def per_item(item):
    # Inline task calls run synchronously inside the same worker
    return sink(stage2(stage1(item)))

@flow(task_runner=ThreadPoolTaskRunner(max_workers=5))
def my_flow(items):
    per_item.map(items)

2) Rolling window with submit + as_completed - Keep only N stage1 items in-flight; as each finishes, immediately kick off its downstream and backfill stage1 by one. This yields “eager” per-item progression and avoids OOM.

from prefect import flow, task
from prefect.futures import as_completed
from prefect.task_runners import ThreadPoolTaskRunner

@task
def stage1(x): ...
@task
def stage2(y): ...
@task
def sink(z): ...

@flow(task_runner=ThreadPoolTaskRunner(max_workers=5))
def my_flow(items, width=5):
    items = list(items)
    in_flight = [stage1.submit(i) for i in items[:width]]
    todo = items[width:]

    for fut in as_completed(in_flight):
        s2 = stage2.submit(fut)   # pass the future; dependency is automatic
        sink.submit(s2)

        # keep the window full
        if todo:
            in_flight.append(stage1.submit(todo.pop(0)))

3) Batch by chunks - Simple and memory-safe: run the whole pipeline for small batches before moving on.

from itertools import islice

def batched(it, n):
    it = iter(it)
    while True:
        batch = list(islice(it, n))
        if not batch:
            break
        yield batch

@flow(task_runner=ThreadPoolTaskRunner(max_workers=5))
def my_flow(items):
    for batch in batched(items, 5):
        out = sink.map(stage2.map(stage1.map(batch)))
        out.result()  # force completion and free references per batch

Alternative: leave capacity for downstream with tag-based concurrency limits - If you’re on Prefect Cloud/Server and can use tag limits, you can cap how many “stage1” tasks run concurrently so downstream can run alongside. - Example (CLI):

prefect concurrency-limit create stage1 --limit 2

- Docs: - Concepts: Tag-based concurrency limits - How-to: Configure tag-based concurrency limits Notes and references - Mapping, futures, and rolling patterns: Using map and Futures API (as_completed, wait, PrefectFutureList) - Each downstream mapped item depends on its corresponding upstream item (passing a future as a parameter creates that dependency), but the executor processes tasks in submission order unless you orchestrate as shown above. If you can share a minimal snippet of your flow (stages and how they’re chained), I can tailor one of these patterns to your exact setup.

Yep! Here’s a subflow pattern that runs “pipelines” in parallel — each pipeline does load -> transform -> unload for a single ID — with up to 16 concurrent pipelines. Note: flows don’t support .map/.submit. To run subflows concurrently in Prefect 3.x, make the parent flow async and use asyncio.gather (with a semaphore to cap concurrency).

import asyncio
from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

# --- Your per-step tasks ---
@task
def load_one(i: str) -> dict:
    # replace with your real implementation
    return {"id": i, "value": do_load(i)}

@task
def transform_one(d: dict) -> dict:
    # replace with your real implementation
    return {"id": d["id"], "value": do_transform(d["value"])}

@task
def unload_one(d: dict) -> None:
    # replace with your real implementation
    write_to_s3(d)

# --- Subflow: one full pipeline per ID ---
@flow(task_runner=ThreadPoolTaskRunner(max_workers=16))  # choose ThreadPool vs ProcessPool based on workload
async def process_one_subflow(item_id: str) -> None:
    # Sequential per-item pipeline inside the subflow
    d = load_one(item_id)
    t = transform_one(d)
    unload_one(t)

# --- Parent flow: run many subflows concurrently (16 at a time) ---
@flow
async def parent_flow(ids: list[str], max_concurrent_pipelines: int = 16):
    sem = asyncio.Semaphore(max_concurrent_pipelines)

    async def run_one(i: str):
        async with sem:
            await process_one_subflow(i)

    await asyncio.gather(*(run_one(i) for i in ids))

# Example invocation
if __name__ == "__main__":
    ids = [f"id-{i}" for i in range(1, 101)]
    # This will run up to 16 concurrent pipelines; each pipeline runs load->transform->unload for one ID
    parent_flow(ids)

Why this works - Each subflow is a full per-item pipeline, so as soon as load finishes for that item, its transform and unload proceed without waiting on other items. - The parent async flow uses a semaphore to keep at most 16 pipelines running at once. - The subflow has its own task runner; pick ThreadPoolTaskRunner for I/O-bound work or ProcessPoolTaskRunner for CPU-bound transforms. If you’d like, I can adapt this to your exact functions (

do_load

,

do_transform

,

write_to_s3

) and suggest the right task runner based on whether you’re I/O- or CPU-bound. For more on subflows and async concurrency: Prefect flows and subflows.

Absolutely — here are two clean ways to “collapse per-item pipeline into a single unit” using a subflow, so your pool runs N end-to-end pipelines instead of stage-by-stage waves. Option A: Subflow does the whole pipeline (sync) - Use a subflow to encapsulate load → transform → unload for one item. - Map the task that invokes the subflow, or submit subflows directly if you prefer.

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def load_one(i: str) -> dict:
    ...

@task
def transform_one(d: dict) -> dict:
    ...

@task
def unload_one(d: dict) -> None:
    ...

@flow
def process_item_flow(i: str) -> None:
    d = load_one(i)            # direct call; runs synchronously in this subflow
    t = transform_one(d)
    unload_one(t)

@task
def run_subflow_for_item(i: str) -> None:
    # Call the subflow synchronously so one thread processes the whole pipeline
    process_item_flow(i)

@flow(task_runner=ThreadPoolTaskRunner(max_workers=5))
def parent_flow(ids: list[str]):
    # Map the wrapper task so you get exactly 5 end-to-end pipelines in flight
    run_subflow_for_item.map(ids)

Why use the wrapper task? Tasks have

.map()

, flows don’t. This keeps your “pipeline width” equal to

max_workers

. Option B: Async subflow with concurrent scheduling - If your steps can be async (or you just want concurrent subflows controlled by the pool), you can still encapsulate the pipeline in a subflow and run it from a task.

import asyncio
from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def load_one(i: str) -> dict:
    ...

@task
def transform_one(d: dict) -> dict:
    ...

@task
def unload_one(d: dict) -> None:
    ...

@flow
async def process_item_flow(i: str) -> None:
    # Tasks inside a subflow can still be called directly (blocking per subflow),
    # or with .submit() if you want intra-subflow concurrency.
    d = load_one(i)
    t = transform_one(d)
    unload_one(t)

@task
def run_async_subflow(i: str) -> None:
    # Run the async subflow synchronously from a task
    asyncio.run(process_item_flow(i))

@flow(task_runner=ThreadPoolTaskRunner(max_workers=5))
def parent_flow(ids: list[str]):
    run_async_subflow.map(ids)

Notes and tips - If you’d rather submit subflows directly, you can use

process_item_flow.submit(i)

in a loop. That’s great when you’re distributing subflows to remote workers via a work pool. But for local “N pipelines in a thread pool” semantics, the “task that calls a subflow” pattern above is the simplest. - If your aim is memory safety, this approach ensures each worker only holds one item’s data at a time through the pipeline. - Choose ThreadPoolTaskRunner for I/O-bound work; switch to ProcessPoolTaskRunner for CPU-bound transforms. If you share whether your steps are I/O- or CPU-bound and whether you’re running locally or via workers, I can tailor the final pattern (including error handling and retries) to your setup.

Short answer: your suspicion is correct. In Prefect 3.x, when you pass futures (including a PrefectFutureList from mapping) as arguments to a subflow, Prefect resolves them at the flow boundary. That means the child flow receives fully materialized data, not unresolved futures. There isn’t a way to “chain” unresolved futures across subflows by just passing them as parameters. A few specifics and options: - PrefectFutureList does exist and is returned by mapped tasks. Passing it to a subflow will trigger resolution to a list of results before the subflow starts. - There is an annotation

quote(...)

that can prevent resolution of a parameter, but that just passes the future object itself into the subflow; it won’t auto-wire orchestration to “continue” those futures inside the subflow. In remote/infrastructure scenarios it may not even be serializable. It’s generally not the pattern you want for chaining computations. - Flows do support a

wait_for=[futures]

kwarg to create dependencies without passing data. This waits for upstream futures to complete before the subflow runs, but it does not pass their results as parameters. Recommended patterns for “chained processing” without pulling large results into memory at subflow boundaries: 1) Keep the mapped chain in a single flow (or a single subflow) as tasks - This preserves the task DAG so Prefect coordinates execution without materializing large intermediates at a Python call boundary.

from prefect import flow, task

@task
def step_a(x): 
    return x * 2

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

@flow
def main(items):
    a_futs = step_a.map(items)      # PrefectFutureList
    b_futs = step_b.map(a_futs)     # stays in the same DAG
    return b_futs.result()

2) If you must split into multiple subflows, pass small references (paths/IDs) and persist the heavy results - Configure your tasks to persist outputs (e.g., upload to S3/GCS, or rely on Prefect result storage) and return only lightweight keys. The second subflow loads what it needs.

from prefect import flow, task

@task
def produce_and_store(x) -> str:
    # write a large artifact to storage and return a small key/path
    key = f"<s3://my-bucket/obj-{x}.parquet>"
    # ...write data...
    return key

@task
def consume_from_key(key: str):
    # load/process as needed
    return f"processed {key}"

@flow
def subflow_a(items):
    # returns small keys, even though the true outputs are large
    return produce_and_store.map(items)

@flow
def subflow_b(keys):
    return consume_from_key.map(keys)

@flow
def parent(items):
    keys = subflow_a(items)            # resolves to list[str], not huge data
    return subflow_b(keys)             # mapped on keys

3) Use wait_for to coordinate readiness without data movement - This can help when you want to ensure the second subflow starts only after upstream mapped work is done, while you pass only references or other small parameters.

@flow
def parent(items):
    a_keys_futs = produce_and_store.map(items)  # PrefectFutureList of keys
    # pass small parameters to subflow_b but wait on upstream completion
    return subflow_b(wait_for=a_keys_futs, keys=[f.result() for f in a_keys_futs])

Notes and references: - Futures API and types (PrefectFuture, PrefectFutureList, wait, as_completed): prefect/futures.py - Flow engine parameter resolution (futures are resolved at flow boundaries): prefect/flow_engine.py - Annotations like quote/opaque (advanced control; generally not for chaining futures across flows): prefect/utilities/annotations.py - Flows accept wait_for for dependency-only gating: see Flow.call and engine handling in the links above

If you can share which constraint is driving the subflow split (memory, isolation, scheduling, or team boundaries), I can suggest a concrete pattern and wiring (e.g., result storage config, key schemas, or a refactor to keep the mapped chain in one flow).