Short answer: - Tasks run in the same pod/process as their parent flow run. - Subflows are separate flow runs and, on Kubernetes, will be picked up by a worker and executed in their own Kubernetes Job/Pod (i.e., a new pod by default). Details and caveats: - In Prefect 3.x, a “task” is just a function executed by the task runner within the flow-run process. By default this does not create new pods; it runs inline in the same pod as the parent flow. - If you configure a distributed task runner (e.g., Dask or Ray), the tasks are offloaded to that cluster. Prefect still doesn’t create one pod per task; instead, the Dask/Ray cluster handles where they run (which could be separate pods if your cluster is on Kubernetes). - If you want pod-level isolation per unit of work, use subflows (each subflow run will create its own pod on a Kubernetes work pool) or make each unit its own flow/deployment and trigger them individually. If you’re on a different version (e.g., Prefect 2.x) let me know, but the behavior above is the same in practice: tasks run in the flow’s pod; subflows are separate runs that result in separate pods on Kubernetes.

You’re right: there’s no flow.submit in Prefect. In Prefect 3.x: - tasks.submit runs concurrently in the same pod - to get 1-pod-per-unit, you need separate flow runs (i.e., deployments), which a Kubernetes worker will execute in separate pods Native pattern to fan out N pods from a sync parent flow - Make the “task” you want isolated its own flow with a deployment - From the parent flow, trigger N child flow runs via run_deployment - Do the fan-out and wait concurrently using small wrapper tasks (so the parent doesn’t block serially) Example:

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run

# tiny task to launch a child flow run and return its id
@task
def launch_child(deployment_name: str, params: dict) -> str:
    # as_subflow=True links runs in the UI; each run will still execute in its own pod
    fr = run_deployment(
        name=deployment_name,
        parameters=params,
        as_subflow=True,
        timeout=0,  # do not wait here
        # work_queue_name="k8s-queue",           # optional routing
        # job_variables={"cpu": "1", "memory": "2Gi"}  # optional pod overrides
    )
    return str(fr.id)

# tiny task to wait on a child run to finish
@task
def wait_child(flow_run_id: str):
    return wait_for_flow_run(flow_run_id).state

@flow(task_runner=ThreadPoolTaskRunner(max_workers=32))
def parent(items: list[int]):
    # 1) fan out: trigger N child flows (N pods)
    launches = [launch_child.submit("child-flow/my-deployment", {"item": i}) for i in items]
    child_run_ids = [f.result() for f in launches]

    # 2) wait for all children concurrently (optional)
    waits = [wait_child.submit(fr_id) for fr_id in child_run_ids]
    results = [w.result() for w in waits]
    return results

Why this works for 1 pod per unit - Each run_deployment call creates a new flow run routed to your Kubernetes work pool; the worker spins a new Job/Pod for it - The parent remains in its own pod; the child runs each get their own pod - Using small @task wrappers lets you launch and wait concurrently in a sync flow Notes and tips - Ensure the child flow has a deployment and is routed to your Kubernetes work pool - Tune concurrency in the parent with ThreadPoolTaskRunner(max_workers=...) to match your desired fan-out - Make sure your Kubernetes work pool/worker can run that many runs at once (work pool concurrency, cluster autoscaling, image pull time, etc.) - Use job_variables in run_deployment to tweak per-pod resources, env, image, etc. - Use idempotency_key if you want to dedupe launches on retries Docs - Run work concurrently - Task runners - Trigger deployments (run_deployment) If you share how many concurrent “units” you need and your work pool name, I can tailor the max_workers, work queue, and job_variables example for your setup.