Short answer: yes — with a few caveats. Your pattern will work in Prefect 3.x with DaskTaskRunner if: -

download

is a Prefect task - You import

unmapped

from Prefect - You call

.map(...)

from inside a

@flow

(mapping requires a flow context) - You keep the returned futures and optionally wait on them before the flow exits Here’s a complete example adapted to your design:

from dataclasses import dataclass
from typing import Dict, List

from prefect import flow, task, unmapped
from prefect_dask import DaskTaskRunner

@dataclass
class CollectionData:
    provider: str
    download_links: List[str]

# Example tasks
@task
def get_new_items(end_date, n_days, collection, granules, aoi_filter):
    # return provider, list_of_links
    ...

@task
def download(provider: str, link: str) -> str:
    # do the download and return a path or record
    ...

@flow(task_runner=DaskTaskRunner())  # or DaskTaskRunner(address="<tcp://scheduler:8786>")
def run_downloads(collections, end_date, n_days, granules, aoi_filter):
    # This runs in the flow process (outside the Dask workers)
    all_items: Dict[str, CollectionData] = {}

    for coll in collections:
        # If you want this strictly outside the Dask cluster, keep it as a plain function
        # or call get_new_items.fn(...) to run synchronously in the flow process.
        provider, links = get_new_items.fn(
            end_date=end_date,
            n_days=n_days,
            collection=coll,
            granules=granules,
            aoi_filter=aoi_filter,
        )
        all_items[coll] = CollectionData(provider=provider, download_links=links)

    # Map downloads per collection; provider is constant, links are mapped
    all_futures = []
    for collection, data in all_items.items():
        futures = download.map(
            provider=unmapped(data.provider),
            link=data.download_links,
        )
        all_futures.extend(futures)

    # Optionally wait for completion and collect results
    results = [f.result() for f in all_futures]
    return results

Notes and tips: - Mapping semantics: at least one argument must be an iterable (

link

in your case). Anything wrapped with

unmapped(...)

is treated as a constant for every mapped run. - Flow context:

.map(...)

must be called inside a

@flow

. If you truly need to map outside a flow, there’s a

deferred=True

mode, but that’s not typical for this pattern. - Concurrency: with DaskTaskRunner, mapped tasks will execute in parallel on Dask workers. Your per-collection loops will still run concurrently across collections because the futures are independent. - Remote clusters: if you connect to an external Dask cluster, make sure your task code and dependencies are available on the workers (e.g., same environment or packaged code). - Alternative flattening pattern: you can build two flat lists (providers repeated per link, and links) and call a single

download.map(provider=providers, link=links)

if you prefer one large mapped call. If you share how you plan to run Dask (local vs. external cluster) and whether

get_new_items

should be a task or a plain function, I can tailor the example further.

Great question — let me break down exactly what you get from each future and what’s inside

done

. Assumptions: - Prefect 3.x -

wait

is imported from

prefect.futures

(important — not

concurrent.futures

) What each mapped future returns: -

out_qvf_paths = process_hdf.map(hdf_path=all_hdfs)

- Type:

PrefectFutureList[list[str]]

(one future per HDF) - Each

future.result()

here returns a list (e.g., a list of QV file paths for that HDF) -

out_qvf_names = save_qvf_files.map(qvfiles=out_qvf_paths, scriptname="modisimport_flow.py")

- Type:

PrefectFutureList[list[str]]

- Each

future.result()

returns a list (e.g., normalized/saved QV filenames for that HDF) - Your flattening is correct: it produces a single list of all QV filenames

qvfs_created = [name for future in out_qvf_names for name in future.result()]

-

futures = update_db_table.map(qvf_fns=out_qvf_names)

- Type:

PrefectFutureList[None]

(if

update_db_table

returns

None

) - Each

future.result()

will be

None

on success What

wait(futures)

returns: -

done, not_done = wait(futures)

returns two sets of Prefect futures: - `done`: a set of

PrefectFuture

objects for the mapped

update_db_table

runs that reached a terminal state (Completed, Failed, or Cancelled) - `not_done`: still running/pending - In your loop:

for future in done:
      if future.state.is_completed() and not future.state.is_failed():
          outputs.append(future.result())  # This will append None for each success
          successes += 1

- You’re correctly counting successful

update_db_table

runs - Note:

is_completed()

already excludes failures, so

and not future.state.is_failed()

is redundant Minor note on counts: - This line works but is a bit surprising:

save_failures = len(all_hdfs * 2) - save_successes

It duplicates the list then counts. Equivalent and clearer:

save_failures = 2 * len(all_hdfs) - save_successes

References: - Prefect futures/wait implementation: prefect/src/prefect/futures.py - Task mapping (Task.map/unmapped): prefect/src/prefect/tasks.py If you want, I can suggest a version that gathers success/failure counts directly from the states without building

outputs

(since they’re

None

).

You’re on the right track. The key is that you’re mixing two different “units”: - Number of futures: one per HDF at each map stage - Number of created files: potentially multiple per HDF (you expect 2) So: - Counting DB updates: count states of

futures

(one per HDF) - Counting created QV files: flatten results from

out_qvf_names

(list-of-lists), but guard against failures A couple of fixes I recommend: - Use

out_qvf_names.result(raise_on_failure=False)

so your summary doesn’t blow up if any save task failed (those entries will be

None

) - Compute DB successes/failures from states after

wait(futures)

(no timeout =

not_done

should be empty) - Keep the “2 per HDF” expectation clear and computed from

len(all_hdfs)

Here’s a tightened-up summary section:

from itertools import chain
from prefect.futures import wait
from prefect import get_run_context
from prefect.settings import get_current_settings

# ... your maps ...
out_qvf_paths = process_hdf.map(hdf_path=all_hdfs)
out_qvf_names = save_qvf_files.map(qvfiles=out_qvf_paths, scriptname="my.py")
futures = update_db_table.map(qvf_fns=out_qvf_names)

# Ensure DB updates are finished (blocks until all are terminal)
done, not_done = wait(futures)

# Safely collect created QV filenames even if some saves failed
# (failed entries will be None)
qvf_lists = out_qvf_names.result(raise_on_failure=False)
qvfs_created = [name for lst in qvf_lists if lst is not None for name in lst]

save_successes = len(qvfs_created)
expected_qvfs = 2 * len(all_hdfs)
save_failures = max(0, expected_qvfs - save_successes)

# Count DB update results
db_successes = sum(1 for f in done if f.state.is_completed())
db_failures = len(done) - db_successes  # failed/cancelled

context = get_run_context()
flow_run_name = context.flow_run.name
settings = get_current_settings()

msg_body = (
    f"flow run Summary: {flow_run_name}"
    f"\n\nsave failed: {save_failures}"
    f"\nsave succeeded: {save_successes}"
    f"\nDB updates failed: {db_failures}"
    f"\nDB updates succeeded: {db_successes}"
)

# Optionally include links for failed DB updates
for f in done:
    if not f.state.is_completed():
        task_id = f.task_run_id
        url = f"{settings.ui_url}/runs/task-run/{task_id}"
        msg_body += f"\nDB update task {task_id} incomplete: {f.state} {url}"

<http://logger.info|logger.info>(f"save succeeded: {save_successes}")
<http://logger.info|logger.info>(f"save failed: {save_failures}")
<http://logger.info|logger.info>(f"DB updates succeeded: {db_successes}")
<http://logger.info|logger.info>(f"DB updates failed: {db_failures}")

my_package.sendMsg(["recipient1"], "my subject", msg_body)

Why this helps: -

out_qvf_names.result(raise_on_failure=False)

won’t raise if a save failed; you just skip

None

entries when flattening -

done

is a set of the

update_db_table

futures that have reached a terminal state; with no timeout,

not_done

should be empty. Counting

is_completed()

gives you DB successes; everything else in

done

is a failure/cancel - The counts now reflect: - save_successes/save_failures = file-level outputs - db_successes/db_failures = per-HDF DB updates Links for reference: - Prefect futures and wait: prefect/src/prefect/futures.py - Task mapping/unmapped: prefect/src/prefect/tasks.py If you want per-HDF detail (e.g., which HDF saved how many files and its DB update state), I can show a small zip over

(out_qvf_paths, out_qvf_names, futures)

that builds a per-item summary.

Awesome — here’s a per-HDF summary pattern that zips your inputs with each mapped stage and produces both per-item details and aggregates. It relies on state checks (so it’s robust to failures) and only pulls results with raise_on_failure=False where needed. ``` from prefect.futures import wait from prefect import get_run_context from prefect.settings import get_current_settings # After your maps: # out_qvf_paths = process_hdf.map(hdf_path=all_hdfs) # out_qvf_names = save_qvf_files.map(qvfiles=out_qvf_paths, scriptname="my.py") # futures = update_db_table.map(qvf_fns=out_qvf_names) # Ensure downstream is terminal; upstreams will be terminal due to dependencies done, _ = wait(futures) settings = get_current_settings() per_item = [] for i, (hdf, paths_fut, names_fut, db_fut) in enumerate(zip(all_hdfs, out_qvf_paths, out_qvf_names, futures)): # Current states (terminal after wait on db_fut) paths_state = paths_fut.state names_state = names_fut.state db_state = db_fut.state # Safely get saved names (list[str] or None if failed) saved_names = names_fut.result(raise_on_failure=False) if saved_names is None: saved_names = [] saved_count = len(saved_names) # Optional: safely get generated QV paths too (if you want to include them) qv_paths = paths_fut.result(raise_on_failure=False) or [] qv_paths_count = len(qv_paths) per_item.append({ "index": i, "hdf": hdf, # States "paths_state": paths_state.type.value, # e.g. "COMPLETED", "FAILED", "CANCELLED" "save_state": names_state.type.value, "db_state": db_state.type.value, "paths_ok": paths_state.is_completed(), "save_ok": names_state.is_completed(), "db_ok": db_state.is_completed(), # Outputs "qv_paths_count": qv_paths_count, "qv_paths": qv_paths, # optional, may be large "saved_count": saved_count, "saved_names": saved_names, # optional, may be large # UI links "paths_task_url": f"{settings.ui_url}/runs/task-run/{paths_fut.task_run_id}", "save_task_url": f"{settings.ui_url}/runs/task-run/{names_fut.task_run_id}", "db_task_url": f"{settings.ui_url}/runs/task-run/{db_fut.task_run_id}", # Optional: include messages for quick diagnostics "paths_message": getattr(paths_state, "message", None), "save_message": getattr(names_state, "message", None), "db_message": getattr(db_state, "message", None), }) # Aggregates total_hdfs = len(all_hdfs) expected_saved = 2 * total_hdfs total_saved = sum(item["saved_count"] for item in per_item) save_failures = max(0, expected_saved - total_saved) db_successes = sum(1 for item in per_item if item["db_ok"]) db_failures = total_hdfs - db_successes # Compose a summary message context = get_run_context() flow_run_name = context.flow_run.name lines = [ f"flow run Summary: {flow_run_name}", f"Save succeeded (files): {total_saved}", f"Save failed (files): {save_failures}", f"DB updates succeeded (HDFs): {db_successes}", f"DB updates failed (HDFs): {db_failures}", "", "Per-HDF details:", ] for item in per_item: lines.append( f"- [{item['index']}] {item['hdf']}" f" | paths={item['paths_state']} ({item['qv_paths_count']})" f" | save={item['save_state']} ({item['saved_count']})" f" | db={item['db_state']}" ) if not item["db_ok"]: lines.append(f" DB task: {item['db_task_url']} msg={item['db_message']}") if not item["save_ok"]: lines.append(f" Save task: {item['save_task_url']} msg={item['save_message']}") if not item["paths_ok"]: lines.append(f" Paths task: {item['paths_task_url']} msg={item['paths_message']}") msg_body = "\n".join(lines) logger.info(f"Save succeeded (files): {total_saved}") logger.info(f"Save failed (files): {save_failures}") logger.info(f"DB updates succeeded (HDFs): {db_successes}")

logger.info(f"DB updates failed (HDFs): {db_failures}") # Send report my_package.sendMsg(["recipient1"], "my subject", msg_body) ``` Why this works well: - It zips the inputs and each mapped stage by index, so every HDF gets its own row. - It uses

.state

to classify outcomes (Completed/Failed/Cancelled), which is robust even if a task failed upstream and the downstream was cancelled. - It uses

.result(raise_on_failure=False)

only to gather lists for counting, avoiding exceptions. - You get clickable UI links for any problematic task run. If you want to persist

per_item

as a JSON artifact (for later inspection), we can add a small helper to write it to storage or log it as a Prefect artifact. For reference on futures and mapping: futures and tasks.

Short answer: yes. If you want an accurate “save failures” number, base your expected count on the number of successful upstream items (downloads or processed HDFs), not just the number of futures you created. A couple concrete improvements: 1) Fix result handling for mapped saves Right now, a failed mapped task will give you an Exception object when you call

.result(raise_on_failure=False)

, not

None

. Your comprehension will break when it tries to iterate an Exception. Handle exceptions explicitly:

# Collect save results safely
qvf_lists = out_qvf_names.result(raise_on_failure=False)

# Files actually created
qvfs_created = [
    name
    for lst in qvf_lists
    if isinstance(lst, list)  # only flatten successful list-returning tasks
    for name in lst
]

# Optional: capture save errors for reporting
save_errors = [e for e in qvf_lists if isinstance(e, Exception)]

2) Compute “expected” from successful upstreams If each successful download should produce 2 QVFs, count only the downloads that actually succeeded:

# Determine which downloads succeeded
download_results = [f.result(raise_on_failure=False) for f in all_hdfs]
successful_downloads = [r for r in download_results if not isinstance(r, Exception)]

expected_qvfs = 2 * len(successful_downloads)
save_successes = len(qvfs_created)
save_failures = max(0, expected_qvfs - save_successes)

Alternatively, if “expected” should reflect the last successful stage before saving, base it on successful process_hdf results instead:

proc_results = out_qvf_paths.result(raise_on_failure=False)
successful_proc = [r for r in proc_results if not isinstance(r, Exception)]

expected_qvfs = 2 * len(successful_proc)
save_failures = max(0, expected_qvfs - save_successes)

3) Optional: capture download failures explicitly This gives you a clean funnel report (downloads -> processed -> saved):

download_errors = [
    r for r in download_results if isinstance(r, Exception)
]

proc_errors = [
    r for r in proc_results if isinstance(r, Exception)
]

# save_errors already computed above

This way you can distinguish: - Save failures (save task errored when inputs were good) - Skips/upstream failures (no save because download/process failed) If you share what each task returns on success/failure, I can tailor the checks (e.g., list vs path vs None) so your counts are perfectly aligned.

This looks solid — you’re chaining the maps correctly and using

wait(futures)

to block on the DB updates. A few refinements to make your reporting accurate and robust: - When you collect mapped results, handle exceptions explicitly (failed tasks return an Exception object when you use

raise_on_failure=False

, not None). - Base “expected” counts on the last successful upstream stage that guarantees the 1→2 QVF fan-out (typically

process_hdf

, or

download_hdfs

if that’s your contract). - Optionally, capture errors at each stage so you can report a clean funnel (downloads → processed → saved → DB updates). - If

all_items

ends up empty, consider short-circuiting before you call

wait

on an empty list. Here’s a compact pattern you can drop in after your current code:

# Guard: no DB updates to wait on
if not futures:
    <http://logger.info|logger.info>("No DB updates to apply.")
else:
    done, not_done = wait(futures)

# 1) Downloads
download_results = [f.result(raise_on_failure=False) for f in all_hdfs]
dl_ok = [r for r in download_results if not isinstance(r, Exception)]
dl_errs = [e for e in download_results if isinstance(e, Exception)]

# 2) Process HDFs
proc_results = out_save_files.result(raise_on_failure=False)
proc_ok = [r for r in proc_results if not isinstance(r, Exception)]
proc_errs = [e for e in proc_results if isinstance(e, Exception)]

# 3) Save files (returns list[str] per item)
save_results = out_qvf_names.result(raise_on_failure=False)
qvfs_created = [name for lst in save_results if isinstance(lst, list) for name in lst]
save_errs = [e for e in save_results if isinstance(e, Exception)]

# 4) DB updates
update_results = [f.result(raise_on_failure=False) for f in futures] if futures else []
update_errs = [e for e in update_results if isinstance(e, Exception)]

# Expected vs actual (choose the correct upstream for your contract)
expected_qvfs = 2 * len(proc_ok)   # or: 2 * len(dl_ok)
save_failures = max(0, expected_qvfs - len(qvfs_created))

<http://logger.info|logger.info>(
    "Summary: downloads ok=%d err=%d | processed ok=%d err=%d | "
    "saved files=%d (expected=%d, missing=%d) | db_update_err=%d | not_done=%d",
    len(dl_ok), len(dl_errs),
    len(proc_ok), len(proc_errs),
    len(qvfs_created), expected_qvfs, save_failures,
    len(update_errs), len(not_done) if futures else 0,
)

Optional enhancements: - If you want to stream progress and start DB updates ASAP, iterate with

as_completed(out_qvf_names)

and map updates as each save finishes. - If you truly aren’t using Dask, the default

ConcurrentTaskRunner

will still execute these maps concurrently; your comment mentions Dask — confirm which runner you’ve configured so expectations match. If you share what each task returns on success (e.g., paths vs dataclasses) I can tailor the type checks further.