Proposal: safe_map function for tasks

[evamaxfield]

Introduction

Hey all I have been using Prefect for a bit and have started generalizing many of my Task.map operations to allow for more operations to occur on error but that don't stop the entire workflow and am looking for feedback and other peoples ideas as to how they are handling things like this. To do so I have been doing the following:

General Form

class TaskResult(NamedTuple):
    key: str
    result: Any

class TaskError(NamedTuple):
    key: str
    error: Exception

@task
def mapped_task(key: str, some, other, params) -> Union[TaskResult, TaskError]:
    try:
        # do some work
        return TaskResult(key, result)
    except Exception as e:
        return TaskError(key, e)

@task
def filter_types(objs, dtype):
    # This could also just be the `FilterTask`
    # https://docs.prefect.io/api/latest/tasks/control_flow.html#filtertask
    return [obj for obj in objs if isinstance(obj, dtype)

# assume we have defined:
# keys = [...]
# other params
with Flow("example") as flow:
     # process
     results = mapped_task.map(keys, unmapped(some), unmapped(other), unmapped(params))

     # split results and errors
     errors = filter_types(results, TaskError)
     results = filter_types(results, TaskResult)

    # do work with errors
    # do work with results

Proposal

I think this could potentially be done for me by adding a safe_map or similarly named function to the Task object. Example API:

@task
def mapped_task(key: str, some, other, params) -> Any
    # do some work
    return result

with Flow("example") as flow:
    results, errors = mapped_task.safe_map(
        keys,
        unmapped(some),
        unmapped(other),
        unmapped(params),
        unique_parameters=["key"]
    )

    # do work with errors
    # do work with results

Note: I added the unique_parameters parameter to the function because it is incredibly valuable to me to be returned enough information about which set of parameters failed the Task. This parameter could be removed in favor of simply returning the entire parameter set that failed however.

Use Cases

The most common use case for me would be to log the error in some non-Prefect-Server controlled database while continuing on with the rest of the flow. But a direct example where I think people would see direct value would be something like taking the errors and sending a Slack message alerting of which items failed to process during the map.

Directly for me this is how I process thousands of images, and in general if we lose a couple images in the pipeline, we don't care, we really just want to be alerted of what percentage / how many we lost. (I.E. a slack message that says "image_processing failed on 20 images")

Interplay with Triggers

Given the above use cases, this adds to the interplay with Task Triggers. Expanding the above example, we don't care about how many images fail the pipeline, up-to-a-point. The end goal for this feature would be to work tightly with Triggers such as some_failed or some_successful. So that we can log the 20 images that fail, but if there are 1000 images, it does indeed fail the whole workflow run.

---

Looking forward to feedback and ideas. 🙂

[jlowin]

Hey @JacksonMaxfield - I understand your use case but I'm not sure if this yet rises to a level that we would want to promote as first-class in the Prefect framework. My recommendation at this time would be to take your first example code and bundle it up as a custom function safe_map which applies your logic in a reusable way.

If you look through the Prefect codebase, we very rarely apply semantic meaning (or even binary splits) to task states; we view the objective of the library as providing first-class tools for working with a multitude of states (success and failed of course, but also retries, skips, scheduled, and a variety we haven't even released yet!). We've only taken an "opinion" when we see a clear and unambiguous pattern emerge. Consequently I think it's a little too early to automatically have a map operator that splits the world into success and failure, though I'd welcome seeing similar patterns from the community in case a consensus has emerged around this. I suspect, however, that people handle this through a mix of filter tasks (when the processing happens inline) and triggers (when the processing happens post-map).

This is just my gut design sense at this time - to reiterate, I do understand your use case and objective, but I'd want to see clear consensus emerge before we codify it, as there are many ways to achieve this.

[evamaxfield]

Hmmm I hadn't thought about the other task states at all so excellent point!

I still can see a future where there is a split_map or something that does split out the results into a dictionary or similar of all the states and then a list of the values for each state, potentially. But, in general, I think I largely agree with you that there isn't a 100% clear pattern present yet.