Great question. Since you’re on Prefect 2 (Cloud), the simplest, resilient pattern is: - Put the 5 per-item steps into a subflow. - Submit one subflow per element so they can run independently (and concurrently if you want). - Let a subflow fail on its own item, notify on that failure, but don’t let it bubble up and fail the parent flow. Here’s a concrete example you can drop in. It continues processing the rest of the items even if one fails, and it sends a notification for the failed item.

from typing import Any, Iterable
from prefect import flow, task, get_run_logger
from prefect.task_runners import ConcurrentTaskRunner
from prefect.blocks.notifications import SlackWebhook  # or another notification block

# Example notification task (use any notification block you prefer)
@task
def notify_failure(item: Any, err_msg: str):
    # Load a pre-configured block in Prefect Cloud -> Blocks
    slack = SlackWebhook.load("my-slack-webhook")
    slack.notify(f"Processing failed for item={item}: {err_msg}")

# Your 5 tasks (add retries if helpful)
@task(retries=2, retry_delay_seconds=30)
def step1(item): ...
@task(retries=2, retry_delay_seconds=30)
def step2(x): ...
@task(retries=2, retry_delay_seconds=30)
def step3(x): ...
@task(retries=2, retry_delay_seconds=30)
def step4(x): ...
@task(retries=2, retry_delay_seconds=30)
def step5(x): ...

# Subflow that handles one item end-to-end
@flow
def process_one_item(item: Any):
    logger = get_run_logger()
    try:
        a = step1.submit(item)
        b = step2.submit(a)
        c = step3.submit(b)
        d = step4.submit(c)
        e = step5.submit(d)
        # This will raise if any upstream failed, which we catch below
        return e.result()
    except Exception as exc:
        # Notify and re-raise so the subflow run is marked as Failed
        notify_failure.submit(item, repr(exc))
        raise

# Parent flow that keeps going even when some items fail
@flow(task_runner=ConcurrentTaskRunner())
def process_all(items: Iterable[Any]):
    futures = [process_one_item.submit(item) for item in items]

    results = []
    for f in futures:
        try:
            results.append(f.result())  # catch failures per-item
        except Exception:
            # We already notified inside the subflow; keep going
            continue

    return results

Notes and options: - Concurrency: With

ConcurrentTaskRunner

, each item’s subflow can run in parallel. If you’d rather process sequentially, remove the task runner and call

process_one_item(item)

directly in a loop. - Notifications via Automations: Instead of (or in addition to) the

notify_failure

task, you can set a Prefect Cloud Automation to notify when a flow run or task run enters a Failed state. That way, you don’t need to write notification code at all and each failed subflow will trigger a notification automatically. - Retries: Add `retries`/`retry_delay_seconds` to any of the 5 tasks as needed to make each item more robust without blocking others. Follow-ups: - How would you like to notify (Slack, email, Teams, webhook, GCP Pub/Sub, etc.)? I can tailor the

notify_failure

task to your setup. - Do you need to cap concurrency (e.g., only process K items at a time)? If so, we can add a semaphore or adjust your work pool/worker concurrency settings.

Thanks for the error detail — that’s on me. In Prefect 2, only tasks have

.submit()

. Flows (including subflows) are called like normal functions; there’s no

Flow.submit

. Here’s a corrected pattern that: - Fans out across your N items - Runs 5 tasks in sequence per item - Continues processing other items if one fails - Sends a notification for failed items

from typing import Any, Iterable, List
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
from prefect.blocks.notifications import SlackWebhook  # or any notifier you prefer

# Your 5 steps
@task(retries=2, retry_delay_seconds=30)
def step1(item): ...
@task
def step2(x): ...
@task
def step3(x): ...
@task
def step4(x): ...
@task
def step5(x): ...

@task
def notify_failure(item: Any, info: str = ""):
    slack = SlackWebhook.load("my-slack-webhook")  # create this Block in Prefect Cloud -> Blocks
    slack.notify(f"Processing FAILED for item={item}. {info}")

# Build one item's pipeline by chaining task futures
def launch_one_item_pipeline(item: Any):
    a = step1.submit(item)
    b = step2.submit(a)
    c = step3.submit(b)
    d = step4.submit(c)
    e = step5.submit(d)
    return item, e  # return the terminal future for this item

@flow(task_runner=ConcurrentTaskRunner())
def process_all(items: Iterable[Any]) -> List[Any]:
    # fan out
    terminals = [launch_one_item_pipeline(item) for item in items]

    results = []
    for item, fut in terminals:
        state = fut.wait()  # does not raise; just returns the final State
        if state.is_failed():
            # notify and continue to the next item
            notify_failure.submit(item)
            continue
        results.append(fut.result())  # safe: only called on completed futures
    return results

Notes: - Concurrency:

ConcurrentTaskRunner

lets different items run at the same time; each item’s 5 steps still run in order because they’re chained via futures. - Notifications: The example uses a Slack Webhook Block. Alternatively, you can skip the

notify_failure

task entirely and set up a Prefect Cloud Automation to notify on task/flow failures: - Automations overview - Futures, states, and task runners docs for more detail: - Task runners and .submit - States and futures If you prefer to keep a subflow per item, call it directly (no

.submit

). Subflows run synchronously, so to get per-item concurrency with subflows you’d either: - Wrap the subflow call in a small task and submit those tasks, or - Use

run_deployment

to trigger each item as its own flow run on your workers (good when you want isolation and autoscaling). Want me to show the

run_deployment

version as well, or tailor the notifier to Slack, email, or GCP Pub/Sub?

Got it — the parent flow is failing because an exception is bubbling up when one item’s task chain fails. In Prefect 2, calling

future.result()

on a failed task raises, which marks the flow as Failed. Two fixes: 1) Don’t raise on failed futures; inspect their state instead 2) Catch exceptions in per-item chains Here’s a pattern that keeps the flow Completed even when some items fail, while still notifying:

from typing import Any, Iterable, List, Tuple
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
from prefect.blocks.notifications import SlackWebhook

@task
def step1(item): ...
@task
def step2(x): ...
@task
def step3(x): ...
@task
def step4(x): ...
@task
def step5(x): ...

@task
def notify_failure(item: Any, message: str = ""):
    SlackWebhook.load("my-slack-webhook").notify(f"FAILED for item={item}: {message}")

def launch_chain(item: Any):
    a = step1.submit(item)
    b = step2.submit(a)
    c = step3.submit(b)
    d = step4.submit(c)
    e = step5.submit(d)
    return item, e  # terminal future for this item

@flow(task_runner=ConcurrentTaskRunner())
def process_all(items: Iterable[Any]) -> dict:
    terminals: List[Tuple[Any, "PrefectFuture"]] = [launch_chain(item) for item in items]

    successes, failures = [], []
    for item, fut in terminals:
        state = fut.wait()  # does not raise
        if state.is_failed():
            # notify and continue; do NOT call fut.result() here
            notify_failure.submit(item, state.message or "Task chain failed")
            failures.append(item)
            continue
        # safe to resolve because it's completed
        successes.append(fut.result())
    return {"successes": successes, "failures": failures}

Key points: - Use

fut.wait()

(or

fut.result(raise_on_failure=False)

) to avoid raising on failures. - Don’t call

.result()

on a failed future unless you pass

raise_on_failure=False

or you wrap it in try/except. - As long as your flow function does not raise, the flow will finish in a Completed state even if some task runs failed. If you prefer subflows per item, call the subflow directly (no

.submit

) and catch exceptions inside the subflow so they don’t propagate to the parent; or trigger separate runs via

run_deployment

and just poll states. Docs for reference: - Task runners, futures, and state handling: https://docs-2.prefect.io/latest/concepts/task-runners/ - States and how exceptions propagate: https://docs-2.prefect.io/latest/concepts/states/ - Automations (alternative to in-code notifications): https://docs-2.prefect.io/latest/concepts/automations/ If your flow still shows Failed, can you share the loop where you collect results? It usually means a

.result()

call is still raising somewhere.