Short answer: yes—configure your three subflows to persist results and use a cache policy keyed on their inputs. When you re-run the parent deployment with the same parameters, Prefect will return the cached results for the subflows that already completed and only execute the one(s) that failed. Why this works - Prefect 3.x flows support result persistence and caching (not just tasks). If a subflow has a persisted Completed result and its cache key matches, the engine will reuse that result instead of re-executing the subflow body. - This makes a subsequent re-run of the parent flow effectively “rerun failed only”. What you need 1) Durable result storage accessible to your workers (S3/GCS/Azure/… or a shared filesystem) 2) Subflows decorated with: -

persist_result=True

-

result_storage=...

pointing to your durable store - A cache policy keyed on inputs (and optionally code) so only identical work is reused Example

from datetime import timedelta

from prefect import flow
from prefect.filesystems import RemoteFileSystem
from prefect.cache_policies import Inputs, TaskSource, CompoundCachePolicy

# Result store (use your own bucket/path)
result_store = RemoteFileSystem(basepath="<s3://my-bucket/prefect-results/>")

# Cache on inputs + code so changing code invalidates old results
cache_policy = CompoundCachePolicy(Inputs(), TaskSource())

@flow(
    name="subflow-a",
    persist_result=True,
    result_storage=result_store,
    cache_policy=cache_policy,
    cache_expiration=timedelta(days=30),
)
def subflow_a(x: int) -> int:
    # expensive work...
    return x + 1

@flow(
    name="subflow-b",
    persist_result=True,
    result_storage=result_store,
    cache_policy=cache_policy,
    cache_expiration=timedelta(days=30),
)
def subflow_b(y: int) -> int:
    # expensive work...
    return y * 2

@flow(
    name="subflow-c",
    persist_result=True,
    result_storage=result_store,
    cache_policy=cache_policy,
    cache_expiration=timedelta(days=30),
)
def subflow_c(z: int) -> int:
    # expensive work...
    return z ** 2

@flow(name="parent")
def parent_flow(x: int, y: int, z: int):
    a = subflow_a(x)  # Will be skipped on re-run if cached result exists
    b = subflow_b(y)  # Ditto
    c = subflow_c(z)  # Only this runs if the others are cached
    return a, b, c

How to use it - First run of the deployment runs all three subflows. - If, say,

subflow_b

fails, fix the issue and re-run the deployment with the same inputs. Prefect will: - Detect cached Completed results for

subflow_a

and

subflow_c

and skip them - Execute

subflow_b

again (no cache because it previously failed or no result persisted) Notes and tips - If you want to reuse results even when code changes, use

Inputs()

alone for

cache_policy

. If you want code changes to invalidate the cache (recommended), combine

Inputs()

with

TaskSource()

as shown. - Ensure your result storage is truly durable and shared across runs (e.g., S3/GCS/Azure). Local storage is fine only if the same host/container is reused. - Retries are still useful for transient failures within a single run (

@flow(retries=...)

), but caching/persisted results is what enables “rerun failed only” across separate runs. - If your subflows depend on side effects or non-deterministic inputs, carefully choose your cache policy or add a custom

cache_key_fn

. Docs - Results and result storage: Prefect 3 docs: Results - Caching concepts and policies: Prefect 3 docs: Caching Questions to confirm - Are your subflows pure functions of their inputs (or at least idempotent for the same parameters)? - What execution environment are you using (local, Docker, Kubernetes) and where should results be stored (S3/GCS/etc.)? I can tailor the

result_storage

setup for your stack.

Short answer: no. In Prefect 3.x there is no

flow.submit

. Ways to run flows concurrently or “fire-and-forget”: - Wrap subflow calls in tasks and use

task.submit

(leverages your flow’s task runner concurrency):

from prefect import flow, task

  @flow
  def subflow_a(x): ...
  @flow
  def subflow_b(y): ...

  @task
  def run_a(x): return subflow_a(x)
  @task
  def run_b(y): return subflow_b(y)

  @flow
  def parent():
      f1 = run_a.submit(1)
      f2 = run_b.submit(2)
      return f1.result(), f2.result()

- Trigger separate flow runs via deployments using

run_deployment

(each becomes a child flow run and can proceed in parallel via your workers/agents):

from prefect import flow
  from prefect.deployments import run_deployment

  @flow
  def parent():
      fr1 = run_deployment("my-project/subflow-a", parameters={"x": 1}, as_subflow=True)
      fr2 = run_deployment("my-project/subflow-b", parameters={"y": 2}, as_subflow=True)
      # fr1/fr2 are child flow runs; they execute independently on workers

Helpful docs: - Task runners and concurrency: Task runners - Deployments and running them programmatically: Deployments