Great pattern. In Prefect 3.x, you’ll want a classic fan-out/fan-in: - Fan-out: launch many append-to-staging tasks concurrently - Fan-in: once they’re all done, run a single merge/dedupe task Two clean ways to “fan-in” the merge: - Use

wait_for=[...]

on the merge task for a state-only dependency - Or call

wait(futures)

before submitting the merge Here’s a minimal, production-friendly example:

from prefect import flow, task
from prefect.futures import wait
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.runtime import flow_run

@task(retries=3, retry_delay_seconds=5)
def append_to_staging(batch, staging_table, db_url):
    # write batch to staging (append)
    # e.g., use DB driver/sqlalchemy; keep it idempotent if possible
    return len(batch)  # optional: return rows written

@task
def merge_and_dedupe(staging_table, final_table, db_url):
    # do your SQL-based dedupe and merge in a single transaction
    # e.g.:
    # - dedupe staging (CTE or DELETE w/ window function)
    # - upsert into final (MERGE/ON CONFLICT)
    # - optionally TRUNCATE or DROP staging
    return {"merged": True}

@flow(task_runner=ThreadPoolTaskRunner(max_workers=8))
def load_with_staging(batches, final_table, db_url):
    # Optional: unique staging table per run to avoid cross-run collisions
    staging_table = f"{final_table}_stg_{flow_run.id[:8]}"

    # Fan-out: concurrent appends
    futures = append_to_staging.map(
        batches,
        staging_table=[staging_table] * len(batches),
        db_url=[db_url] * len(batches),
    )

    # Option A: state-only dependency via wait_for
    # merge_future = merge_and_dedupe.submit(
    #     staging_table, final_table, db_url, wait_for=list(futures)
    # )

    # Option B: explicitly wait before submitting merge
    wait(futures)
    merge_future = merge_and_dedupe.submit(staging_table, final_table, db_url)

    return merge_future.result()

Notes and best practices: - Use

ThreadPoolTaskRunner(max_workers=...)

for I/O-bound DB writes; consider

ProcessPoolTaskRunner

if your transformation is CPU-bound before the write. - Prefer a unique staging table per flow run (e.g., suffix with

flow_run.id

) so parallel runs don’t contend. If you must share a single staging table across runs, use a run_id column and scope your merge to only the current run’s rows. - Make appends idempotent where possible, and add

retries

to handle transient DB issues. - Do the dedupe+merge in a single transaction to keep it atomic. - If your merge should only run when all appends succeed, stick with

wait(futures)

and verify their states before merging. If partial success is acceptable, filter completed futures:

done, _ = wait(futures)
  successful = [f for f in done if f.state.is_completed()]
  if len(successful) != len(futures):
      # decide whether to proceed, alert, or fail
      pass

If you share a bit more about your database (Postgres, Snowflake, BigQuery, etc.) and the dedupe key/logic, I can tailor the merge SQL/approach and connection code.