Short answer: yes—this is a common and recommended pattern in Prefect 2. An “orchestrator” flow that fans out to other flows conditionally and in parallel is a great way to separate concerns and scale work. How to choose between subflows and deployments - Subflows (call the flow function directly) - Best when all work can run on the same infrastructure and you want simple data sharing. - Returns values directly (like a normal function call), which makes passing data between flows easy. - Runs in the same process/worker as the orchestrator, so it’s tightly coupled and resource-shared. - Good for small/medium result payloads. - Appears as nested runs in the UI automatically. - Deployments (trigger with

run_deployment

) - Best when you want execution isolation, different infrastructure (e.g., different pools, regions, or machine types), or to scale fan-out beyond a single worker. - Parent calls child via `run_deployment(...)`; each child runs on its own worker. - Treat data hand-offs like services: prefer writing to storage (e.g., GCS/BigQuery) and passing references/IDs, not large in-memory objects. - Can be triggered in parallel by wrapping

run_deployment

in a task and calling

.submit()

many times. Running the child flows in parallel - Subflows: wrap the subflow call in a task and

.submit()

those tasks to get concurrency with the default ConcurrentTaskRunner. - Deployments: create a small task that calls

run_deployment(...)

and

.submit()

many of them. Each child will be picked up by its own worker, enabling wide parallelism. Examples 1) Orchestrator with parallel subflows (easy data sharing)

from prefect import flow, task, get_run_logger

@flow
def child_flow(x: int) -> int:
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"child got {x}")
    return x * 2

@task
def invoke_child(x: int):
    # Call the flow to create a subflow run; returns the child’s result
    return child_flow(x)

@flow
def orchestrator(numbers: list[int]):
    # fan out in parallel via tasks
    futures = [invoke_child.submit(n) for n in numbers]
    results = [f.result() for f in futures]
    return sum(results)

2) Orchestrator that fans out to deployments (execution isolation + scale)

from prefect import flow, task
from prefect.deployments.flow_runs import run_deployment

@task
def trigger_child(i: int):
    # name is "<flow_name>/<deployment_name>"
    return run_deployment(
        name="child-flow/production",
        parameters={"x": i},
        as_subflow=True,               # shows as a subflow in the UI
        flow_run_name=f"child-{i}",
        tags=["orchestrated"],
    )

@flow
def orchestrator(items: list[int]):
    runs = [trigger_child.submit(i) for i in items]
    flow_runs = [r.result() for r in runs]  # FlowRun objects
    # For large results, write to GCS/BigQuery in the child and pass references

Additional guidance - Parallelism limits: The default task runner is concurrent and great for IO-bound fan-out. For CPU-heavy work, consider a DaskTaskRunner or RayTaskRunner. Constrain concurrency with tags and global limits if needed. - Data handoffs on GCP: Prefer storing large data in GCS or BigQuery in child flows and passing URIs/IDs back to the orchestrator. Prefect’s GCP collection can help: prefect-gcp docs. - UI: Subflows are nested automatically. For deployments triggered from a parent,

as_subflow=True

helps you visualize lineage. Good references - Patterns overview: Workflow design patterns - Subflows and flows: Flows and subflows - Task runners and concurrency: Task runners - Deployments (incl. triggering from flows): Deployments tutorial If you share a bit about: - the size/type of data you’re passing between flows,

- how many parallel children you expect, and - whether the children need different GCP infrastructure, I can recommend a concrete pattern (e.g., subflows only vs. deployments + GCS/BigQuery handoffs) and code tailored to your setup.

Great question. In Prefect 2.x: - Subflows block: If your orchestrator calls another flow as a subflow (i.e., calls the flow function), the parent will wait for it to finish. You can catch errors to prevent them from failing the parent, but you can’t “fire-and-forget” a subflow — it runs in the same flow run and must finish. - To not wait, use deployments (separate flow runs): Trigger child flows as deployments. The orchestrator can schedule child runs and return immediately, so errors in children won’t block or fail the parent. Two practical patterns 1) Fire-and-forget via the API (recommended for true decoupling) - This schedules child flow runs and returns immediately; the parent flow is not affected by child errors.

from prefect import flow
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import DeploymentFlowRunCreate

@flow
async def orchestrator(items: list[int]):
    async with get_client() as client:
        dep = await client.read_deployment_by_name("child-flow/production")
        for i in items:
            await client.create_flow_run_from_deployment(
                dep.id,
                DeploymentFlowRunCreate(
                    parameters={"x": i},
                    tags=["orchestrated"],
                    # Optional: flow_run_name=f"child-{i}"
                )
            )
    # Done scheduling; returns without waiting for child completion

Sync alternative:

from prefect import flow
from prefect.client.orchestration import SyncPrefectClient
from prefect.client.schemas.actions import DeploymentFlowRunCreate

@flow
def orchestrator(items: list[int]):
    with SyncPrefectClient() as client:
        dep = client.read_deployment_by_name("child-flow", "production")
        for i in items:
            client.create_flow_run_from_deployment(
                deployment_id=dep.id,
                flow_run=DeploymentFlowRunCreate(parameters={"x": i}, tags=["orchestrated"])
            )
    # Returns immediately

2) Using run_deployment without waiting -

run_deployment(...)

schedules a flow run and can poll for completion. If you don’t want to wait, call it inside a task and just don’t poll or resolve.

from prefect import flow, task
from prefect.deployments.flow_runs import run_deployment

@task
def trigger(i: int):
    # Schedules the child run and returns a FlowRun object immediately
    return run_deployment(
        name="child-flow/production",
        parameters={"x": i},
        as_subflow=False  # optional: don’t show as nested UI subflow if you want full decoupling
        # You can also set a very short timeout if you’ve used waiting/polling elsewhere
    )

@flow
def orchestrator(items: list[int]):
    # Fan out scheduling tasks; parent does not wait for child completion
    _ = [trigger.submit(i) for i in items]
    # Do not resolve these futures; return now

Notes and implications - Subflows cannot be fire-and-forget. They run in the same process and the parent waits. - True non-blocking orchestration requires separate flow runs (deployments). - If you don’t want the parent to even “track” children in the UI, use

as_subflow=False

or the client API. If you want visibility, leave

as_subflow=True

and optionally query child states later. - Error handling: With fire-and-forget, child failures won’t fail the parent. Use Prefect Cloud Automations to alert on failed child runs or add a separate “monitor” flow that queries child run states. Helpful references - Flows and subflows: https://docs-2.prefect.io/latest/concepts/flows/ - Deployments (incl. triggering from flows): https://docs-2.prefect.io/latest/tutorial/deployments/ If you share whether you prefer async or sync flows and how you’d like to monitor child failures (e.g., Slack alerts, retries), I can tailor the snippet for your GCP/Cloud setup.