Short answer: yes—at two different levels. Which one you want depends on whether you mean “many flow runs at once” or “one flow run that does lots of work at once.” Here are your options with an ECS worker + work pool: 1) Parallel flow runs (fan-out across ECS tasks) - Each flow run becomes its own ECS Task. To run flows in parallel, just create/trigger multiple flow runs; your ECS worker will launch multiple ECS tasks concurrently as long as: - Your work pool doesn’t have a restrictive concurrency limit - Your worker is allowed to claim multiple runs - Your ECS cluster has capacity - Scale by running multiple workers and/or increasing pool concurrency. This is the simplest way to “distribute.” 2) Parallelism within a single flow run (task-level concurrency) - Use a task runner and submit tasks concurrently. Example:

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

@task
def do_work(i):
    # your work here
    return i * 2

@flow(task_runner=ConcurrentTaskRunner(max_workers=16))
def my_flow():
    futures = [do_work.submit(i) for i in range(100)]
    results = [f.result() for f in futures]
    return results

- This runs tasks concurrently inside the single ECS task that’s executing the flow run. - Good for IO-bound work. For heavy CPU, consider distributing (see next). 3) Distributed execution for one “parent” flow (many machines) - Fan out into many child flow runs (each a separate ECS task) from inside your flow by calling

run_deployment

repeatedly for a “child” deployment. That gives you real horizontal scale across ECS tasks. - Alternatively, use a distributed task runner (Dask/Ray) via the Prefect collections if you already have a Dask/Ray cluster available. This keeps it one flow run but distributes work to the cluster. Notes and sanity checks - If you only run one flow at a time, you’ll only get one ECS task at a time. Parallel flow runs require multiple triggers (or a parent flow that fans out). - Bound your blast radius with pool concurrency limits if needed; then increase limits as you scale out. - Ensure your AWS side (cluster capacity, task definitions, service quotas) can handle the parallelism you ask for. If you can share which flavor you want—parallel flow runs, parallel tasks within one run, or true distributed execution—I’ll give you the exact wiring. Related reading (high level but useful): - Work Pools and Workers - Scaling parallel work (conceptual; in modern Prefect you’ll use

.submit

instead of 1.x-style mapping) - Prefect 3 GA overview

Yep. Here’s a minimal flow that clearly shows parallel branches in the UI. The key is: use a task runner and submit tasks without introducing dependencies between them, then have a single downstream task depend on all of them.

from time import sleep
from random import randint
from prefect import flow, task, get_run_logger
from prefect.task_runners import ConcurrentTaskRunner

@task
def fetch(i: int) -> int:
    logger = get_run_logger()
    d = randint(1, 3)
    <http://logger.info|logger.info>(f"fetch {i}: sleeping {d}s")
    sleep(d)
    <http://logger.info|logger.info>(f"fetch {i}: done")
    return i

@task
def combine(values: list[int]) -> int:
    return sum(values)

@flow(name="parallel-ui-demo", task_runner=ConcurrentTaskRunner(max_workers=8))
def parallel_flow(n: int = 6) -> int:
    # Fan-out: these run concurrently
    futures = [fetch.submit(i) for i in range(n)]

    # IMPORTANT for a clear graph: pass futures directly to the downstream task
    # (don’t call .result() here, or the UI loses the explicit dependency edges)
    total = combine.submit(futures)

    # If you want a Python value in the driver at the end:
    return total.result()

if __name__ == "__main__":
    parallel_flow()

What you’ll see in the UI: - A “parallel-ui-demo” flow run with N sibling

fetch

tasks at the same level, running at the same time. - One downstream

combine

task that depends on all of them. Notes: -

ConcurrentTaskRunner(max_workers=8)

controls intra-flow parallelism. Bump it up (within reason) to increase concurrency. - This works the same on ECS: each flow run is one ECS task; the parallelism happens inside that task. Light reading (if you insist): - Scaling parallel work - Work pools and workers

Yes. Use the Dask task runner from the Prefect-Dask integration. Here’s a minimal example that shows clear parallel branches in the UI. Install first: - pip:

pip install "prefect-dask dask[distributed]"

- Or if you insist on being tidy:

uv run --with 'prefect-dask' --with 'dask[distributed]' python your_script.py

Example with a local Dask cluster (parallel on the ECS task/container that’s running the flow):

from time import sleep
from random import randint

from prefect import flow, task, get_run_logger
from prefect_dask import DaskTaskRunner
from dask.distributed import LocalCluster

@task
def fetch(i: int) -> int:
    logger = get_run_logger()
    d = randint(1, 3)
    <http://logger.info|logger.info>(f"fetch {i}: sleeping {d}s")
    sleep(d)
    <http://logger.info|logger.info>(f"fetch {i}: done")
    return i

@task
def combine(values: list[int]) -> int:
    return sum(values)

@flow(
    name="dask-parallel-ui-demo",
    task_runner=DaskTaskRunner(
        cluster_class=LocalCluster,
        cluster_kwargs={"n_workers": 4, "threads_per_worker": 1},  # adjust as needed
    ),
)
def parallel_flow(n: int = 6) -> int:
    # Fan-out: these run concurrently on Dask workers
    futures = [fetch.submit(i) for i in range(n)]

    # Keep the UI graph parallel by passing futures, not resolved results
    total = combine.submit(futures)
    return total.result()

if __name__ == "__main__":
    parallel_flow()

What you’ll see in the UI: - N sibling fetch tasks running at the same time (parallel branches) - One downstream combine task depending on all of them Want true distributed across multiple machines? Point the runner to an existing Dask scheduler:

@flow(task_runner=DaskTaskRunner(address="<tcp://dask-scheduler:8786>"))
def parallel_flow(n: int = 6):
    futures = [fetch.submit(i) for i in range(n)]
    return combine.submit(futures).result()

Notes: - On ECS, the LocalCluster runs inside the single ECS task for that flow run. For multi-machine scale, use an external Dask cluster (ECS, EKS, EC2, etc.) and pass its scheduler address. - Keep the futures wired into downstream tasks to preserve the parallel graph in the UI. Docs, if you like reading more than watching things run: - prefect-dask integration - Dask distributed

Absolutely. Here’s a larger “mini-ETL” example that runs clearly in parallel and looks good in the UI. It shows both: - A flow using ConcurrentTaskRunner (threads in a single process) - A flow using DaskTaskRunner (local Dask cluster; swap to a remote scheduler for real distribution) Install extras for Dask: - pip: pip install "prefect prefect-dask dask[distributed]" ``` import random import time import math from statistics import mean, stdev from typing import List, Dict from prefect import flow, task, get_run_logger from prefect.task_runners import ConcurrentTaskRunner # Dask runner (optional) try: from prefect_dask import DaskTaskRunner from dask.distributed import LocalCluster DASK_AVAILABLE = True except Exception: DASK_AVAILABLE = False # ------------------------- # Meaningful-ish ETL tasks # ------------------------- @task def extract_batch(batch_id: int, batch_size: int = 20_000) -> List[float]: """ Simulate extracting a batch of numeric records (IO-ish + a bit of jitter). """ logger = get_run_logger() # Simulate IO latency so parallelism is obvious in the UI time.sleep(0.2 + random.random() * 0.3) # Generate some data with mild structure rng = random.Random(batch_id) data = [rng.random() * (1 + 0.1 * math.sin(i / 500)) for i in range(batch_size)] logger.info(f"Batch {batch_id}: extracted {len(data)} records") return data @task def transform_features(data: List[float]) -> Dict[str, float]: """ Compute features: mean, stdev, percentiles, and a simple non-linear transform. Sorting makes this a bit CPU-bound so parallel workers are visible. """ logger = get_run_logger() n = len(data) if n < 2: return {"n": n, "mean": data[0] if n == 1 else 0.0, "stdev": 0.0, "p50": 0.0, "p90": 0.0, "p99": 0.0, "energy": 0.0} data_sorted = sorted(data) p = lambda q: data_sorted[int(q * (n - 1))] features = { "n": n, "mean": mean(data), "stdev": stdev(data), "p50": p(0.50), "p90": p(0.90), "p99": p(0.99), # "Energy" is an arbitrary non-linear summary to add some CPU work "energy": sum(x * x for x in data) / n, } logger.info(f"Transformed: mean={features['mean']:.4f}, p99={features['p99']:.4f}") return features @task def score_batch(features: Dict[str, float]) -> Dict[str, float]: """ Score a batch with a toy model combining features. """ # A pretend scoring function score = ( 0.5 * features["mean"] + 0.3 * (features["p99"] - features["p50"]) + 0.2 * features["stdev"] + 0.1 * features["energy"] ) return {"n": features["n"], "score": score} @task def aggregate(scores: List[Dict[str, float]]) -> Dict[str, float]: """ Aggregate batch scores into a global result. """ total_n = sum(s["n"] for s in scores) weighted = sum(s["score"] * s["n"] for s in scores) / max(total_n, 1) return {"total_records": total_n, "weighted_score": weighted} # ------------------------------------------------- # Flow 1: ConcurrentTaskRunner (simple, zero extras) # ------------------------------------------------- @flow(name="etl-concurrent-demo", task_runner=ConcurrentTaskRunner(max_workers=16)) def etl_concurrent(n_batches: int = 8, batch_size: int = 20_000) -> Dict[str, float]: logger = get_run_logger() logger.info("Starting concurrent ETL") # Fan-out: extract in parallel extracted_futures = [extract_batch.submit(i, batch_size) for i in range(n_batches)] # Per-batch transform in parallel feature_futures = [transform_features.submit(f) for f in extracted_futures] # Per-batch score in parallel score_futures = [score_batch.submit(f) for f in feature_futures] # Important for a clear parallel graph in the UI: pass futures, not resolved values aggregated = aggregate.submit(score_futures) result = aggregated.result()

logger.info(f"Concurrent ETL complete: {result}") return result # -------------------------------------------- # Flow 2: DaskTaskRunner (local Dask cluster) # -------------------------------------------- if DASK_AVAILABLE: @flow( name="etl-dask-demo", task_runner=DaskTaskRunner( cluster_class=LocalCluster, cluster_kwargs={"n_workers": 4, "threads_per_worker": 1}, # tune as needed ), ) def etl_dask(n_batches: int = 8, batch_size: int = 20_000) -> Dict[str, float]: logger = get_run_logger() logger.info("Starting Dask ETL") extracted_futures = [extract_batch.submit(i, batch_size) for i in range(n_batches)] feature_futures = [transform_features.submit(f) for f in extracted_futures] score_futures = [score_batch.submit(f) for f in feature_futures] aggregated = aggregate.submit(score_futures) result = aggregated.result() logger.info(f"Dask ETL complete: {result}") return result else: def etl_dask(*args, **kwargs): raise RuntimeError("prefect-dask and dask[distributed] are not installed") if name == "__main__": # Run both for comparison (comment out one if you like) etl_concurrent() if DASK_AVAILABLE: etl_dask() ``` What you’ll see in the Prefect UI: - A wide “fan-out” of sibling extract tasks, then a parallel transform layer, then a parallel score layer, and finally a single aggregate node. - With ConcurrentTaskRunner, everything runs inside one process (threaded). - With DaskTaskRunner(LocalCluster), tasks run on multiple Dask workers inside the same ECS container. To scale across machines, point at an external scheduler: - Replace the Dask runner with: DaskTaskRunner(address="tcp://your-dask-scheduler:8786") Notes for ECS: - Each flow run is one ECS task. The parallelism above happens inside that container. - For true multi-machine distribution, run a Dask scheduler/workers separately (ECS service, EKS, EC2, etc.) and give the flow the scheduler address. - Keep passing futures into downstream tasks to preserve the parallel branches in the UI. Helpful links: - prefect-dask integration - Dask distributed If you want, I can adapt this to: - Write results to S3 or a DB - Use an external Dask cluster you already have - Fan out into child flow runs for horizontal scaling across ECS tasks

✅ thought for 61.0 seconds

Absolutely. Below is a “grouped analysis” mini-pipeline that: - Loads a synthetic transaction dataset - Groups it (by category by default) - Fans out per-group analysis in parallel (stats + top customers per group) - Aggregates the results You get two flows: - etl_grouped_concurrent: uses ConcurrentTaskRunner (threads in one process) - etl_grouped_dask: uses DaskTaskRunner (a local Dask cluster; swap to a remote scheduler for true distribution) Install for Dask: - pip install "prefect prefect-dask dask[distributed]" ``` import math import random import time from collections import defaultdict from typing import Dict, List, Tuple from prefect import flow, task, get_run_logger from prefect.task_runners import ConcurrentTaskRunner import statistics as stats # Optional Dask runner try: from prefect_dask import DaskTaskRunner from dask.distributed import LocalCluster DASK_AVAILABLE = True except Exception: DASK_AVAILABLE = False # ------------------------- # Synthetic dataset + tasks # ------------------------- @task def fetch_transactions( n_customers: int = 200, n_txn: int = 60_000, n_categories: int = 8, seed: int = 42, ) -> List[Dict]: """ Simulate a transaction dataset with mild structure. Each record: {customer_id, category, amount}. """ logger = get_run_logger() logger.info("Fetching synthetic transactions...") time.sleep(0.3) # simulate IO latency rng = random.Random(seed) data = [] for _ in range(n_txn): cust = rng.randrange(n_customers) cat = rng.randrange(n_categories) base = 20 + 10 * math.sin(cat) + 5 * math.log1p(cust % 7) amount = abs(rng.lognormvariate(1 + 0.05 * cat, 0.75)) + base data.append({"customer_id": cust, "category": cat, "amount": amount}) logger.info(f"Generated {len(data)} transactions across {n_categories} categories") return data @task def group_by_key( transactions: List[Dict], key: str = "category", ) -> List[Tuple[int, List[Dict]]]: """ Group records by the given key and return a list of (group_key, records). """ groups = defaultdict(list) for tx in transactions: groups[tx[key]].append(tx) # Deterministic ordering for nicer logs grouped = sorted(groups.items(), key=lambda kv: kv[0]) return grouped @task def compute_group_stats(group_key: int, records: List[Dict]) -> Dict: """ Compute descriptive stats for a group. """ logger = get_run_logger() vals = [r["amount"] for r in records] n = len(vals) if n == 0: return {"group": group_key, "n": 0, "mean": 0.0, "stdev": 0.0, "p90": 0.0, "max": 0.0, "anomaly": 0.0} vals_sorted = sorted(vals) p90 = vals_sorted[int(0.9 * (n - 1))] mean = stats.fmean(vals) stdev = stats.pstdev(vals) if n > 1 else 0.0 max_val = vals_sorted[-1] anomaly = (max_val - mean) / (stdev + 1e-6) if n > 1 else 0.0 # Add a tiny pause to make parallelism obvious time.sleep(0.05) logger.info(f"Group {group_key}: n={n}, mean={mean:.2f}, p90={p90:.2f}, anomaly={anomaly:.2f}") return { "group": group_key, "n": n, "mean": mean, "stdev": stdev, "p90": p90, "max": max_val, "anomaly": anomaly, } @task def top_k_customers(group_key: int, records: List[Dict], k: int = 5) -> Dict: """ Compute top-k customers by spend for a group. """ spend_by_customer = defaultdict(float) for r in records: spend_by_customer[r["customer_id"]] += r["amount"] top = sorted(spend_by_customer.items(), key=lambda x: x[1], reverse=True)[:k] return { "group": group_key, "top_customers": [{"customer_id": c, "spend": s} for c, s in top], } @task def aggregate_group_results(stats_list: List[Dict], tops_list: List[Dict]) -> Dict: """ Merge per-group outputs into a global summary. """ # Index by group for safe merging

stats_by_group = {s["group"]: s for s in stats_list} tops_by_group = {t["group"]: t for t in tops_list} total_n = sum(s["n"] for s in stats_by_group.values()) or 1 weighted_mean = sum(s["mean"] * s["n"] for s in stats_by_group.values()) / total_n # Most anomalous groups (top 3) anomalies = sorted(stats_by_group.values(), key=lambda s: s["anomaly"], reverse=True)[:3] merged = [] for g, s in stats_by_group.items(): merged.append({ "group": g, **s, **tops_by_group.get(g, {"top_customers": []}), }) return { "summary": { "total_records": total_n, "weighted_mean": weighted_mean, "top_anomalies": [{"group": a["group"], "anomaly": a["anomaly"]} for a in anomalies], }, "per_group": merged, } # ----------------------------------------------- # Flow A: ConcurrentTaskRunner (threads, one host) # ----------------------------------------------- @flow(name="grouped-analysis-concurrent", task_runner=ConcurrentTaskRunner(max_workers=16)) def etl_grouped_concurrent( n_customers: int = 200, n_txn: int = 60_000, n_categories: int = 8, ) -> Dict: logger = get_run_logger() logger.info("Starting grouped analysis with ConcurrentTaskRunner") data_f = fetch_transactions.submit(n_customers=n_customers, n_txn=n_txn, n_categories=n_categories) groups_f = group_by_key.submit(data_f, key="category") # We need the group list to know how many tasks to spawn groups = groups_f.result() # [(group_key, records), ...] # Fan-out per group: parallel stats and top-k per group stats_futures = [] tops_futures = [] for g, recs in groups: stats_futures.append(compute_group_stats.submit(g, recs)) tops_futures.append(top_k_customers.submit(g, recs, k=5)) # Aggregate; pass futures to keep parallel edges in the UI aggregated = aggregate_group_results.submit(stats_futures, tops_futures) result = aggregated.result() logger.info(f"Concurrent flow done. Total records: {result['summary']['total_records']}") return result # ----------------------------------------------- # Flow B: DaskTaskRunner (local Dask cluster) # ----------------------------------------------- if DASK_AVAILABLE: @flow( name="grouped-analysis-dask", task_runner=DaskTaskRunner( cluster_class=LocalCluster, cluster_kwargs={"n_workers": 4, "threads_per_worker": 1}, # tune to taste ), ) def etl_grouped_dask( n_customers: int = 200, n_txn: int = 60_000, n_categories: int = 8, ) -> Dict: logger = get_run_logger() logger.info("Starting grouped analysis with DaskTaskRunner") data_f = fetch_transactions.submit(n_customers=n_customers, n_txn=n_txn, n_categories=n_categories) groups_f = group_by_key.submit(data_f, key="category") groups = groups_f.result() stats_futures = [] tops_futures = [] for g, recs in groups: stats_futures.append(compute_group_stats.submit(g, recs)) tops_futures.append(top_k_customers.submit(g, recs, k=5)) aggregated = aggregate_group_results.submit(stats_futures, tops_futures) result = aggregated.result() logger.info(f"Dask flow done. Total records: {result['summary']['total_records']}") return result else: def etl_grouped_dask(*args, **kwargs): raise RuntimeError("prefect-dask and dask[distributed] are not installed") if name == "__main__": # Run both for comparison etl_grouped_concurrent() if DASK_AVAILABLE: etl_grouped_dask() ``` What you’ll see in the UI: - One grouping stage, then a wide fan-out of per-group analysis tasks (compute_group_stats and top_k_customers), followed by a single aggregate node. - With ConcurrentTaskRunner, everything runs inside a single ECS task/container using threads.

- With DaskTaskRunner(LocalCluster), the per-group tasks are distributed across Dask workers inside that container. For multi-machine scale, point to an external scheduler: - Replace the task runner with: DaskTaskRunner(address="tcp://your-dask-scheduler:8786") Docs you might actually want: - prefect-dask: prefect-dask integration - Dask distributed: Dask docs If you want this writing results to S3 or pointing at your existing Dask cluster, say the word and your wish shall be my command. Begrudgingly.