Great question! In Prefect 3.4.x you control task-level concurrency with a flow’s task_runner. Use threads for I/O-bound work and processes for CPU-bound work. Quick mapping from 2.x to 3.x - 2.x

ConcurrentTaskRunner

→ 3.x

ThreadPoolTaskRunner

(I/O concurrency) - 3.x also includes

ProcessPoolTaskRunner

(true CPU parallelism) Multi-threading (I/O-bound) - Good for API calls, DB queries, file I/O - Concurrency is limited by max_workers; the GIL still applies Example:

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

@task
def fetch(i: int) -> int:
    import time, httpx
    time.sleep(0.5)
    return httpx.get("<https://httpbin.org/get>").status_code

@flow(task_runner=ThreadPoolTaskRunner(max_workers=10))
def threaded_flow():
    futs = fetch.map(range(20))     # or [fetch.submit(i) for i in range(20)]
    wait(futs)                      # optional; .result() will also block
    return futs.result()

if __name__ == "__main__":
    threaded_flow()

Notes - Ensure your task code is thread-safe (avoid unsynchronized writes to shared globals). - Changing Prefect context/tags inside tight loops can be expensive; set them outside loops when possible. Multi-processing (CPU-bound) - True parallelism across cores - Each task runs in a separate process; arguments/returns must be picklable - Define tasks at module top-level; avoid capturing non-picklable objects Example:

from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner
from prefect.futures import wait

@task
def heavy(n: int) -> int:
    return sum(i*i for i in range(n))

@flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
def multiproc_flow():
    futs = heavy.map([5_000_000, 6_000_000, 7_000_000, 8_000_000])
    wait(futs)
    return futs.result()

if __name__ == "__main__":
    multiproc_flow()

Notes - Use the if name == "__main__": guard when running locally (especially on Windows or from a script). Prefer running flows from a module or through a worker rather than inside interactive notebooks for process pools. - Processes add serialization overhead; use them for heavier, CPU-bound tasks. How to pick - I/O-bound (network, DB, disk) → ThreadPoolTaskRunner - CPU-bound (compute/ML) → ProcessPoolTaskRunner - Need distributed compute? Consider Dask or Ray task runners. Collecting results - Either block explicitly or return futures:

from prefect.futures import wait
futs = my_task.map(items)
wait(futs)
results = futs.result()  # or [f.result() for f in futs]

Docs - Concepts: Task runners Task runners (3.x) - How-to: Run work concurrently Run work concurrently - API reference: task_runners prefect.task_runners If you can share what your workload looks like (I/O vs CPU, approx task durations, where you run flows), I can recommend concrete max_workers values and patterns.

Short answer: In Prefect 3.4.x, your code runs the mapped tasks concurrently using threads, not processes. Why - You didn’t specify a task runner on the flow, so Prefect 3.x uses the default ThreadPoolTaskRunner. -

.map(...)

submits multiple task runs which the thread pool executes concurrently in the same Python process. - In your task you call

time.sleep(2)

. Sleep releases the GIL, so the threads can overlap and you’ll see ~2s total for all 5 items (plus small overhead), not ~10s. How to prove it (PID is the same, thread IDs differ)

from prefect import task, flow
import time, os, threading

array = [5, 3, 8, 6, 2]

@task
def square_root(num):
    print(f"PID={os.getpid()} thread={threading.get_ident()} num={num}")
    time.sleep(2)
    return num ** 0.5

@flow  # default = ThreadPoolTaskRunner -> multithreading
def square_root_flow(arr):
    return square_root.map(arr).result()

if __name__ == "__main__":
    square_root_flow(array)

If you want multiprocessing (true CPU parallelism), set a ProcessPoolTaskRunner

from prefect import task, flow
from prefect.task_runners import ProcessPoolTaskRunner
import time, os

array = [5, 3, 8, 6, 2]

@task
def square_root(num):
    print(f"PID={os.getpid()} num={num}")
    time.sleep(2)  # or CPU-heavy work
    return num ** 0.5

@flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
def square_root_flow(arr):
    return square_root.map(arr).result()

if __name__ == "__main__":
    square_root_flow(array)

- With processes, you’ll see different PIDs for different tasks. - Arguments/returns must be pickle-able, and there’s extra overhead—use processes for CPU-bound work. Docs - Task runners (3.x): Concepts: Task runners - Run work concurrently: How-to: Run work concurrently

That ImportError means your installed Prefect version doesn’t include ProcessPoolTaskRunner (or you’re on Prefect 2.x). How to confirm - What version are you on?

python -c "import prefect, sys; print('prefect:', prefect.__version__); print('python:', sys.version)"

- What task runners are available?

python - <<'PY'
import prefect, prefect.task_runners as tr
print('prefect:', prefect.__version__)
print([name for name in dir(tr) if 'TaskRunner' in name])
PY

How to fix - Upgrade to a Prefect 3.x version that includes ProcessPoolTaskRunner:

pip install -U prefect
# or
uv pip install -U prefect
# or (conda)
conda install -c conda-forge prefect

- Then import using:

from prefect.task_runners import ProcessPoolTaskRunner

Working example (processes)

from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner
import os, time

@task
def square_root(n: int) -> float:
    print(f"PID={os.getpid()} n={n}")
    time.sleep(2)
    return n ** 0.5

@flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
def square_root_flow(arr):
    return square_root.map(arr).result()

if __name__ == "__main__":
    square_root_flow([5, 3, 8, 6, 2])

If you can’t upgrade right now - Use threads (I/O-bound only):

from prefect.task_runners import ThreadPoolTaskRunner
@flow(task_runner=ThreadPoolTaskRunner(max_workers=10))
def square_root_flow(arr):
    return square_root.map(arr).result()

Docs - Task runners overview: Task runners (3.x) - Run work concurrently: Run work concurrently If you share your

prefect

and Python versions, I can suggest the exact upgrade path.

That sentence is about using multiprocessing (and distributed runners) where your task code runs in separate Python processes. - “All task parameters and return values must be serializable with cloudpickle.” - When Prefect runs tasks in other processes, it has to send the inputs to that process and bring the result back. That transfer happens by serializing (pickling) the data. - cloudpickle can handle many Python objects beyond the stdlib pickle, but not everything. - Practical rule: only pass/return plain data or objects that cloudpickle can serialize. Examples - Usually OK: ints/floats/strings/bools, lists/dicts/tuples of those, dataclasses with simple fields, numpy arrays, pandas DataFrames, top-level functions/classes. - Not OK: open file handles, sockets, database connections/clients (psycopg2, SQLAlchemy sessions, boto3 clients), thread/process locks, generators/coroutines, loggers. Quick self-check:

import cloudpickle
  cloudpickle.dumps(obj)  # succeeds if serializable

Patterns - Don’t pass a DB connection into a task; instead pass a DSN/creds and create the connection inside the task. - Don’t return an open file handle; return the bytes or a path. - “The runner automatically handles context propagation and environment variable passing to subprocess workers.” - Prefect’s runtime context (e.g.,

prefect.runtime.flow_run.id

, tags, settings) is made available inside the subprocess so your task code can use it as if it were running in the main process. - The subprocesses inherit the environment variables, so things like

AWS_*

,

PREFECT_*

, etc. are available without you manually copying them. Notes - Set any required environment variables before the process pool is created (ideally before the flow starts). Changing

os.environ

later won’t necessarily update already-spawned workers. - This applies to ProcessPoolTaskRunner and similarly to distributed runners (Dask/Ray). Thread-based execution does not need serialization because it stays in the same process. Small example showing both concepts:

from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner
from prefect.runtime import flow_run
import os

@task
def show_context_and_do_work(n: int) -> int:
    # Context is available in the subprocess
    print("Flow run ID:", flow_run.id)
    # Env is inherited
    print("ENV FOO:", os.environ.get("FOO"))
    # Return plain data (serializable)
    return n * n

@flow(task_runner=ProcessPoolTaskRunner(max_workers=2))
def my_flow():
    return show_context_and_do_work.map([1, 2, 3]).result()

if __name__ == "__main__":
    os.environ["FOO"] = "bar"
    my_flow()

Docs - Task runners (3.x): Concepts: Task runners - Run work concurrently: How-to: Run work concurrently

Yes — in Prefect you can run tasks in multiple processes and then concatenate their outputs. The general pattern is: - Submit tasks concurrently (e.g., with ProcessPoolTaskRunner or .map) - Collect their futures - Resolve to results and combine however you like (string join, list extend, pandas.concat, etc.) Here are a few compact examples in Prefect 3.x: 1) Lists from multiple processes

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

@task
def make_chunk(i):
    return [i, i + 1, i + 2]

@flow(task_runner=ProcessPoolTaskRunner)
def main():
    futures = [make_chunk.submit(i) for i in range(4)]
    chunks = [f.result() for f in futures]
    all_items = [x for chunk in chunks for x in chunk]  # concat
    return all_items

2) Map + bulk result

from prefect import flow, task

@task
def to_upper(s):
    return s.upper()

@flow
def main():
    futures = to_upper.map(["a", "b", "c"])
    parts = futures.result()        # ["A", "B", "C"]
    joined = "".join(parts)         # "ABC"
    return joined

3) DataFrame concat

import pandas as pd
from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner

@task
def load_part(i):
    return pd.DataFrame({"id": [i, i + 1]})

@flow(task_runner=ProcessPoolTaskRunner)
def main():
    futures = load_part.map(range(4))
    dfs = futures.result()
    df = pd.concat(dfs, ignore_index=True)
    return df

4) Stream results as they finish (optional)

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

@task
def make_chunk(i):
    return [i, i+1, i+2]

@flow
def main():
    futures = [make_chunk.submit(i) for i in range(4)]
    out = []
    for f in as_completed(futures):
        out += f.result()  # concatenate as each completes
    return out

Notes: - Passing futures into a downstream task will auto-resolve them, so you can also do

combine_task(futures)

and Prefect will resolve the list of futures for you inside that task. - Works the same for “two or more than two” processes — just collect however many futures you need. Quick check: what do you want to “concat” specifically (strings, lists, pandas DataFrames, files)? And are these results coming from Prefect tasks in a ProcessPoolTaskRunner, or from separate flows/external processes? I can tailor an example to match your exact case.