Short answer: yes. - Using

logger = get_run_logger()

and then

<http://logger.info|logger.info>(...)

inside a task will show up in the Prefect UI/Cloud even when the task runs with a

DaskTaskRunner

. -

print(...)

does not get sent to the UI by default. If you want

print

output to appear, enable

log_prints=True

on the flow (or on specific tasks). Example:

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

@task
def work():
    logger = get_run_logger()
    <http://logger.info|logger.info>("This will appear in the Prefect UI")
    print("This will only appear if log_prints=True")

@flow(task_runner=DaskTaskRunner(), log_prints=True)
def my_flow():
    work.submit()

my_flow()

Notes/caveats: - Works in distributed Dask execution because Prefect propagates run context to workers; just ensure workers can reach your Prefect API. - Very large log messages may be truncated by Prefect’s API log size limits. - Standard Python logging (e.g.,

logging.getLogger(__name__).info(...)

) also gets captured when called within a run context. If you’re on Prefect 3.x,

DaskTaskRunner

comes from the

prefect-dask

collection.

Sorry, I encountered an error while processing your request. Please try again.

Use

.result()

. - Calling

future.result()

(default

raise_on_failure=True

) will raise if that task failed, which will fail the flow. -

wait()

(either

future.wait()

or

prefect.futures.wait([...])

) only waits for completion and does NOT raise on failure. If you use

wait()

alone, your flow won’t fail unless you then call

.result()

or manually check states and raise. Recommended patterns:

# Multiple submits
futures = [task_fn.submit(x) for x in items]
results = [f.result() for f in futures]  # Flow fails if any task fails

# Using .map()
futures = task_fn.map(items)
results = futures.result()  # Fails the flow if any mapped task fails

If you need to resolve nested structures of futures and still fail on any error:

from prefect.futures import resolve_futures_to_results
resolved = resolve_futures_to_results(data_with_futures)  # Raises on any failure

If you prefer to handle failures gracefully:

from prefect.futures import wait

done, _ = wait(futures)  # no exception raised here
for f in done:
    if f.state.is_failed():
        # inspect/log, then optionally raise to fail the flow
        raise f.state.result()  # or raise RuntimeError(...)

This behavior is the same with `DaskTaskRunner`; the runner doesn’t change how

.result()

vs

.wait()

propagate failures.

Short answer: no—`wait` is intentionally non-raising in Prefect 3.x. It just waits and returns the done/not_done sets. There’s no flag to make it raise on failure. If you want “raise on failure” semantics without keeping results: - All-or-nothing check (raises if any future didn’t complete successfully):

from prefect.futures import resolve_futures_to_results

  resolve_futures_to_results(futures)  # ignore the return; raises if any failed/crashed/cancelled

Note: this raises a generic exception if any are not COMPLETED. - Raise underlying exceptions as they finish (and don’t keep results):

from prefect.futures import as_completed

  for f in as_completed(futures):
      f.result()  # raises the original task exception if it failed

Or after everything has finished:

for f in futures:
      f.result()  # raises underlying exception if failed

References: - prefect.futures.wait - resolve_futures_to_results - as_completed - PrefectFuture.result