[Draft] [flytekit] Polish - Map Tasks

[wild-endeavor]

Map Task Polish

This is a series of tickets to improve the flytekit authoring experience. If any changes are not possible to make in a backwards-compatible way, split it out into a separate ticket.

Rename map task

Rename map_task to just map This will interfere with the native Python map but it's okay, as we now recommend users import flytekit as fl

Failure Toleration

The failure toleration parameters for map_task are very powerful but too verbose. Let's mark min_successes and min_success_ratio as deprecated and make a new argument tolerance that is type float | int.

Parallelism

Deprecate the max_parallelism argument of workflow and LaunchPlan (is there one?) make a new one called concurrency to match that of map_task.

Are you sure this issue hasn't been raised already?

• Yes

Have you read the Code of Conduct?

• Yes

[granthamtaylor]

Can we also consider ways to include fixed arguments? Currently, defining partial functions seems unnecessarily verbose.

Consider the following map_task:

@fl.task
def my_task(greeting: str, name: str):
    print(f"{greeting}, {name}")

@fl.workflow
def my_workflow(greeting: str="hello", names: list[str]=["bob", "bill"]):

    fl.map_task(partial(my_task, greeting=greeting) min_success_ratio=0.5))(name=names)

I see two strategies here:

1. Leverage the type hints to determine which arguments need to be fixed, and which are dynamic.

@fl.workflow
def my_new_workflow(greeting: str="hello", names: list[str]=["bob", "bill"]):

    fl.map(my_task, greeting=greeting, name=names, tolerance=0.5)

    # rename `map_task` to `apply`
    # tolerance is of type `float|int`
    # greeting and name are both **kwargs, and are either used as dynamic or static inputs depending on the type hints used at registration time

2. map could instead be a class class map: ... that has some methods (__init__, fix, run)

@fl.workflow
def my_new_workflow(greeting: str="hello", names: list[str]=["bob", "bill"]):

    fl.map(my_task, tolerance=0.5).fix(greeting=greeting).run(name=names)

3. Use a context manager

@fl.workflow
def my_new_workflow(greeting: str="hello", names: list[str]=["bob", "bill"]):

    with fl.map(my_task, tolerance=0.5) as mapper:
        mapper(greeting=greeting, name=names)

The downside of option 1 is that there are reserved argument names to map (tolerance, concurrency) which means that the task cannot have arguments of the same name.

2 and 3 are a little bit strange.