Short answer: there’s no hard cap in Prefect 3.x. 20k tasks in a single flow run is doable, but there are trade‑offs. Each

task.submit(...)

creates an individual task run that the API/UI tracks, so very large fan‑outs can make the flow run detail page feel heavy and generate a lot of API traffic. If you value per‑case observability and retries, one case per task is fine; if not, batching multiple cases per task will keep things lighter. What to expect and recommended patterns - No hard limit: Prefect doesn’t enforce a maximum number of tasks per flow run. - Practical limits: UI pages with tens of thousands of task runs can be slower to render; large fan‑outs also increase API chatter. The engine batches state updates to reduce overhead, but you’ll still see increased load with 20k tasks. - Concurrency: Your actual concurrency is controlled by the task runner (e.g.,

ThreadPoolTaskRunner(max_workers=25)

), so you can safely submit 20k tasks and only 25 will run at once. Two good approaches for ~20k cases at ~25 concurrent 1) One case per task (maximize observability and retries) - Pros: granular retries/visibility/metrics/logs per case. - Cons: large single flow run UI; lots of task runs. - How to throttle:

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

  @task
  def process_one(case):
      # your logic
      ...

  @flow(task_runner=ThreadPoolTaskRunner(max_workers=25))
  def process_all(cases: list):
      futures = [process_one.submit(c) for c in cases]
      # Optionally wait on results to surface exceptions here:
      # for f in futures: f.result()

- If the UI feels heavy, split into subflows by batch to keep each flow run’s task count smaller:

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

  @task
  def process_one(case):
      ...

  @flow(task_runner=ThreadPoolTaskRunner(max_workers=25))
  def process_batch(batch):
      futures = [process_one.submit(c) for c in batch]
      # Optionally: for f in futures: f.result()

  @flow
  def parent(cases: list, batch_size: int = 1000):
      for i in range(0, len(cases), batch_size):
          process_batch(cases[i:i+batch_size])

This yields ~20 subflow runs of 1,000 tasks each instead of 20k tasks in a single run. 2) Mini‑batch multiple cases per task (minimize task count) - Pros: much fewer task runs; faster UI pages; fewer API events. - Cons: you lose per‑case retries/visibility; one failed case could fail the whole batch task unless you handle errors inside the task. - Example:

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

  @task
  def process_batch_cases(cases):
      results = []
      for c in cases:
          try:
              results.append(("ok", c, ...))  # your result
          except Exception as e:
              results.append(("err", c, str(e)))
      return results

  @flow(task_runner=ThreadPoolTaskRunner(max_workers=25))
  def process_all_batched(cases: list, batch_size: int = 100):
      batches = [cases[i:i+batch_size] for i in range(0, len(cases), batch_size)]
      futures = [process_batch_cases.submit(b) for b in batches]
      # Optionally: for f in futures: f.result()

With batch_size=100 you’d have ~200 task runs instead of 20k. Additional tips - Logging: Be mindful of per‑task log volume; heavy logging across 20k tasks can be a bottleneck. - Results: If outputs are large, persist results to external storage rather than keeping everything in memory or large state payloads. - Distributed runners: If you outgrow a single machine, consider Dask/Ray task runners to distribute work while keeping in‑flow submission patterns the same. - Observability vs. load: Use subflows or batching to find a sweet spot—e.g., subflows of 500–1,000 tasks balances UI responsiveness and per‑case detail. Bottom line

- 20k tasks at 25 concurrency will work. If you need per‑case lineage/retries, keep one case per task but consider subflow batching so each flow run stays in the low thousands of tasks. If you don’t need that granularity, process multiple cases per task to reduce task count. If you can share whether you’re on Prefect Cloud or self‑hosted and whether per‑case retries/observability are required, I can recommend a concrete shape (batch sizes, runner selection, and any result/logging tweaks) tailored to your setup.

Great question. With subflows you can return any serializable value from each child flow and aggregate in the parent. Here are a few proven patterns, from simplest to more scalable. 1) Simple: sequential subflows that return summaries - Each child flow processes a batch and returns a small summary (counts, URIs to results, etc.). - The parent calls each child and combines the summaries.

from prefect import flow

@flow
def process_batch(batch):
    # Do the work for this batch (looping internally to avoid per-case tasks)
    successes, failures, uris = 0, 0, []
    for case in batch:
        try:
            # ... your processing ...
            # optionally write large outputs to storage and collect a URI
            # uris.append(write_to_storage(case_id, result))
            successes += 1
        except Exception:
            failures += 1

    return {"successes": successes, "failures": failures, "uris": uris}

@flow
def parent_flow(cases, batch_size=1000):
    summaries = []
    for i in range(0, len(cases), batch_size):
        summary = process_batch(cases[i:i+batch_size])
        summaries.append(summary)

    total_success = sum(s["successes"] for s in summaries)
    total_fail = sum(s["failures"] for s in summaries)
    all_uris = [u for s in summaries for u in s["uris"]]
    return {"successes": total_success, "failures": total_fail, "uris": all_uris}

2) Concurrency across subflows (no per-case tasks) with an async parent - Run up to N subflows concurrently to achieve your global throttle (e.g., 25). - Each child still loops internally, so you avoid creating 20k tasks.

import asyncio
from prefect import flow

@flow
async def process_batch(batch):
    successes, failures = 0, 0
    for case in batch:
        try:
            # ... your processing ...
            successes += 1
        except Exception:
            failures += 1
    return {"successes": successes, "failures": failures}

@flow
async def parent_flow(cases, batch_size=800, max_concurrent_batches=25):
    # 20k cases, 25 concurrent batches -> ~800 per batch by default
    batches = [cases[i:i+batch_size] for i in range(0, len(cases), batch_size)]
    sem = asyncio.Semaphore(max_concurrent_batches)

    async def run_one(batch):
        async with sem:
            return await process_batch(batch)

    tasks = [asyncio.create_task(run_one(b)) for b in batches]
    summaries = await asyncio.gather(*tasks, return_exceptions=False)

    total_success = sum(s["successes"] for s in summaries)
    total_fail = sum(s["failures"] for s in summaries)
    return {"successes": total_success, "failures": total_fail}

Why this is nice for your use case: - You get exactly ~25-way concurrency across batches. - You create only one flow run per batch (no per-case tasks if you loop inside the child), keeping the UI responsive. - You still get retry/alerting at the batch (flow) level. If you need per-case retries, reintroduce a small number of tasks inside the child. 3) If you do want 25-way concurrency inside each batch (tasks), but still keep UI light - Use a subflow per batch and a capped task runner in that subflow; return only a summary so the parent aggregates tiny data. ``` from prefect import flow, task from prefect.task_runners import ThreadPoolTaskRunner @task def process_case(case): # ... work ... return {"ok": True} # keep it small @flow(task_runner=ThreadPoolTaskRunner(max_workers=25)) def process_batch_with_tasks(batch): futures = [process_case.submit(c) for c in batch] successes = 0 failures = 0 for f in futures: try: _ = f.result() successes += 1 except Exception: failures += 1 return {"successes": successes, "failures": failures} @flow def parent_flow(cases, batch_size=1000): summaries = [] for i in range(0, len(cases), batch_size): summaries.append(process_batch_with_tasks(cases[i:i+batch_size])) return {

"successes": sum(s["successes"] for s in summaries), "failures": sum(s["failures"] for s in summaries), } ``` Practical tips - Return summaries/references, not large payloads. If results are big, write them to storage (S3, GCS, filesystem) inside the child flow and return URIs. The parent can collect URIs and (optionally) post-process. - If you want concurrency only at the “25 total” level and to avoid per-case tasks entirely, pattern #2 (async parent + sequential child loops) is usually the cleanest. - If you need per-case retries/observability, use a small amount of task parallelism inside the child and keep each child’s task count modest (e.g., a few hundred), then aggregate summaries in the parent. If you share: - whether you want batch-level vs per-case retries/visibility, and - roughly how large each case’s output is, I can tailor one of these patterns (batch sizing, concurrency cap, and result handling) to your exact setup.