Hi Can I somehow ignore skipped tasks of a `map` in downstre

Question

Hi Can I somehow ignore skipped tasks of a `map` in downstream tasks Let s say a mapped task skips for some inputs In this case I would still like the downstream task that runs on the return values of the mapped task to continue with the results of all successful task runs An example is in the thread slightly smiling face

Simon Gasse · Answer

```import prefect from prefect import Flow task from prefect triggers import any successful LOG = prefect context get logger task def inc arg return arg + 1 task def skip conditionally arg if arg == 1 raise prefect engine signals SKIP Skipped this argument return arg task trigger=any successful def print args args <http LOG info|LOG info> args with Flow Mapping Test as flow filtered nums = skip conditionally map range 3 increased nums = inc map filtered nums print args increased nums flow run ```

Simon Gasse · Answer

The output is ``` 2021 08 31 14 58 27+0200 INFO prefect FlowRunner | Beginning Flow run for Mapping Test 2021 08 31 14 58 27+0200 INFO prefect TaskRunner | Task skip conditionally Starting task run 2021 08 31 14 58 27+0200 INFO prefect TaskRunner | Task skip conditionally Finished task run for task with final state Mapped 2021 08 31 14 58 27+0200 INFO prefect TaskRunner | Task skip conditionally 0 Starting task run 2021 08 31 14 58 27+0200 INFO prefect TaskRunner | Task skip conditionally 0 Finished task run for task with final state Success 2021 08 31 14 58 27+0200 INFO prefect TaskRunner | Task skip conditionally 1 Starting task run 2021 08 31 14 58 27+0200 INFO prefect TaskRunner | SKIP signal raised SKIP Skipped this argument 2021 08 31 14 58 27+0200 INFO prefect TaskRunner | Task skip conditionally 1 Finished task run for task with final state Skipped 2021 08 31 14 58 28+0200 INFO prefect TaskRunner | Task skip conditionally 2 Starting task run 2021 08 31 14 58 28+0200 INFO prefect TaskRunner | Task skip conditionally 2 Finished task run for task with final state Success 2021 08 31 14 58 28+0200 INFO prefect TaskRunner | Task inc Starting task run 2021 08 31 14 58 28+0200 INFO prefect TaskRunner | Task inc Finished task run for task with final state Mapped 2021 08 31 14 58 28+0200 INFO prefect TaskRunner | Task inc 0 Starting task run 2021 08 31 14 58 28+0200 INFO prefect TaskRunner | Task inc 0 Finished task run for task with final state Success 2021 08 31 14 58 28+0200 INFO prefect TaskRunner | Task inc 1 Starting task run 2021 08 31 14 58 28+0200 INFO prefect TaskRunner | Task inc 1 Finished task run for task with final state Skipped 2021 08 31 14 58 28+0200 INFO prefect TaskRunner | Task inc 2 Starting task run 2021 08 31 14 58 28+0200 INFO prefect TaskRunner | Task inc 2 Finished task run for task with final state Success 2021 08 31 14 58 28+0200 INFO prefect TaskRunner | Task print args Starting task run 2021 08 31 14 58 28+0200 INFO prefect TaskRunner | Task print args Finished task run for task with final state Skipped 2021 08 31 14 58 28+0200 INFO prefect FlowRunner | Flow run SUCCESS all reference tasks succeeded``` As you can see three `inc` task instances are created with one being skipped and `print args` also being skipped Instead I d like one of the following Only two `inc` task instances are created the skipped input is ignored in ` map ` preferred `print args` is not skipped but only runs on the non skipped tasks I might want to abstract this to failure at some point so that failed mapped tasks are ignored in downstream tasks Any suggestions

Simon Gasse · Answer

OK so maybe `FilterTask` will be of use looking into it <https docs prefect io api latest tasks control flow html filtertask>

Simon Gasse · Answer

So this serves my purpose example with uncaught failure instead of caught except ```import prefect from prefect import Flow task from prefect triggers import any successful from prefect tasks control flow filter import FilterTask from prefect engine signals import SKIP LOG = prefect context get logger filter skip fail = FilterTask filter func=lambda x not isinstance x SKIP and not isinstance x BaseException task def inc arg return arg + 1 task def skip conditionally arg if arg == 1 raise RuntimeError raise prefect engine signals SKIP Skipped this argument return arg task trigger=any successful def print args args <http LOG info|LOG info> args with Flow Mapping Test as flow filtered nums = filter skip fail skip conditionally map range 3 increased nums = inc map filtered nums print args increased nums flow run ```

Simon Gasse · Answer

Output ``` 2021 08 31 15 33 09+0200 INFO prefect FlowRunner | Beginning Flow run for Mapping Test 2021 08 31 15 33 09+0200 INFO prefect TaskRunner | Task skip conditionally Starting task run 2021 08 31 15 33 09+0200 INFO prefect TaskRunner | Task skip conditionally Finished task run for task with final state Mapped 2021 08 31 15 33 09+0200 INFO prefect TaskRunner | Task skip conditionally 0 Starting task run 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task skip conditionally 0 Finished task run for task with final state Success 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task skip conditionally 1 Starting task run 2021 08 31 15 33 12+0200 ERROR prefect TaskRunner | Task skip conditionally 1 Exception encountered during task execution Traceback most recent call last lt redacted gt raise RuntimeError RuntimeError 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task skip conditionally 1 Finished task run for task with final state Failed 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task skip conditionally 2 Starting task run 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task skip conditionally 2 Finished task run for task with final state Success 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task FilterTask Starting task run 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task FilterTask Finished task run for task with final state Success 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task inc Starting task run 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task inc Finished task run for task with final state Mapped 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task inc 0 Starting task run 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task inc 0 Finished task run for task with final state Success 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task inc 1 Starting task run 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task inc 1 Finished task run for task with final state Success 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task print args Starting task run 2021 08 31 15 33 12+0200 INFO prefect | 1 3 2021 08 31 15 33 12+0200 INFO prefect TaskRunner | Task print args Finished task run for task with final state Success 2021 08 31 15 33 12+0200 INFO prefect FlowRunner | Flow run SUCCESS all reference tasks succeeded```

Kevin Kho · Answer

Glad you figured this out +1

Simon Gasse · Answer

< Kevin Kho> would it make sense to link this mention it in the docs on `Task map` At least this is where I was looking and could not find it

Kevin Kho · Answer

Yeah I can see that PR s would be welcome if you re interested slightly smiling face

Simon Gasse · Answer

Alright I ll take a look one of the next days wink

Simon Gasse · Answer

< Kevin Kho> I put it up yesterday night and it was already merged slightly smiling face

Kevin Kho · Answer

That was fast