Yes—there are two good ways to run them in parallel in Prefect 2.x. Which one you choose depends on whether you want parallelism inside a single flow run or many independent flow runs. Option A: Parallelize work inside one flow run (use tasks + .submit) - Put the “unit process” logic in a task. - Use a concurrent task runner and call

.submit()

in your loop. - Prefect will run those task runs concurrently within the same parent flow run.

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

@task
def run_unit(item):
    # do the unit work for this item
    return f"done {item}"

@flow(task_runner=ConcurrentTaskRunner())
def global_flow(data):
    futures = [run_unit.submit(x) for x in data]  # launch in parallel
    results = [f.result() for f in futures]       # wait/collect
    return results

When to use: - You want all work to share one flow run for observability and logging. - Work is I/O-bound or moderate CPU and you’re fine scaling concurrency within a single worker. - You can also swap in Dask or Ray task runners if you need real parallel CPU or a cluster. Option B: Launch many separate flow runs (use run_deployment) - Create a deployment for the unit flow. - From the global/orchestrator flow, trigger that deployment multiple times with

timeout=0

to submit and continue; Prefect Cloud will schedule them concurrently across your work pool capacity (e.g., your GCP workers). - Optional: wait for completion by giving each call a

timeout

or by polling their states.

from prefect import flow
from prefect.deployments import run_deployment

@flow
def global_orchestrator(data):
    # fire-and-forget: each is its own flow run, scheduled concurrently
    runs = [
        run_deployment(
            name="unit_flow/unit_deploy",
            parameters={"item": x},
            timeout=0,          # don't block
            as_subflow=True     # link as child runs in UI (set False for independent)
        )
        for x in data
    ]
    return [r.id for r in runs]

When to use: - You want each unit to be its own isolated flow run for retries, scaling, or per-run infra (e.g., GKE/Cloud Run jobs). - You want Prefect Cloud to distribute work across many workers/pods in your GCP work pool. Quick guidance for GCP: - If you’re scaling out across many machines, Option B + a Kubernetes or Cloud Run work pool is usually best. - If you’re staying on a single worker/machine, Option A with

ConcurrentTaskRunner

(or Dask/Ray) is simplest. A couple of clarifying questions to make sure we pick the right pattern: - Do you want each “unit process” to be its own flow run for isolation and scaling, or is it fine to keep everything inside one parent flow run? - How are you running workers today on GCP (Kubernetes, Cloud Run, VM)?

Great—this is possible in Prefect 2.x without

run_deployment

, with one important constraint: to run multiple subflows concurrently when calling them “like functions,” your parent flow and the unit subflow must be async and you should use asyncio to schedule them together. Pattern: async parent + async subflows + asyncio.gather - Define both flows as async. - In the global flow, create the subflow coroutines and pass them to

asyncio.gather(...)

to run them concurrently. Example:

import asyncio
from prefect import flow

@flow
async def unit_flow(item):
    # do your unit work here
    return f"done {item}"

@flow
async def global_flow(data):
    # launch unit subflows concurrently
    coros = [unit_flow(x) for x in data]   # don't await yet
    results = await asyncio.gather(*coros) # run in parallel
    return results

Notes and caveats: - Direct subflow calls are blocking by default; concurrency requires async + gather/task groups. - Task runners affect tasks, not subflows. Putting a subflow call inside a task and submitting it won’t parallelize the subflows. - If you need independent cancellation/infra scaling per unit, consider

run_deployment(as_subflow=True, timeout=0)

. But since you asked to avoid it, the async pattern above is the way to go. Given you’re on Prefect Cloud (2.x) and GCP: - This async subflow pattern keeps everything in a single parent run, which sounds like your goal. - If later you need to fan out across many GCP workers/pods, switching to deployments is the path to scale out-of-process. Want me to help adapt your current code into the async-gather pattern? If you can share the current global flow and unit flow signatures, I can propose the exact changes.