Dask and Ray Task Runners

Task runners provide an execution environment for tasks. In a flow decorator, you can specify a task runner to run the tasks called in that flow.

The default task runner is the ConcurrentTaskRunner.

Use .submit to run your tasks asynchronously

To run tasks asynchronously use the .submit method when you call them. If you call a task as you would normally in Python code it will run synchronously, even if you are calling the task within a flow that uses the ConcurrentTaskRunner, DaskTaskRunner, or RayTaskRunner.

Many real-world data workflows benefit from true parallel, distributed task execution. For these use cases, the following Prefect-developed task runners for parallel task execution may be installed as Prefect Integrations.

• DaskTaskRunner runs tasks requiring parallel execution using dask.distributed.
• RayTaskRunner runs tasks requiring parallel execution using Ray.

These task runners can spin up a local Dask cluster or Ray instance on the fly, or let you connect with a Dask or Ray environment you've set up separately. Then you can take advantage of massively parallel computing environments.

Use Dask or Ray in your flows to choose the execution environment that fits your particular needs.

To show you how they work, let's start small.

Remote storage

We recommend configuring remote file storage for task execution with DaskTaskRunner or RayTaskRunner. This ensures tasks executing in Dask or Ray have access to task result storage, particularly when accessing a Dask or Ray instance outside of your execution environment.

Configure a task runner

You may have seen this briefly in a previous tutorial, but let's look a bit more closely at how you can configure a specific task runner for a flow.

Let's start with the SequentialTaskRunner. This task runner runs all tasks synchronously and may be useful when used as a debugging tool in conjunction with async code.

Let's start with this simple flow. We import the SequentialTaskRunner, specify a task_runner on the flow, and call the tasks with .submit().

from prefect import flow, task
from prefect.task_runners import SequentialTaskRunner


@task
def say_hello(name):
    print(f"hello {name}")


@task
def say_goodbye(name):
    print(f"goodbye {name}")


@flow(task_runner=SequentialTaskRunner())
def greetings(names):
    for name in names:
        say_hello.submit(name)
        say_goodbye.submit(name)

if __name__ == "__main__":
    greetings(["arthur", "trillian", "ford", "marvin"])

Save this code as sequential_flow.py and run it.

python sequential_flow.py

If you remove the log messages from the output and just look at the printed output from the task runs, you see they're executed sequentially:

hello arthur
goodbye arthur
hello trillian
goodbye trillian
hello ford
goodbye ford
hello marvin
goodbye marvin

Run tasks in parallel with Dask

This basic flow won't benefit from parallel execution, but let's proceed so you can see just how simple it is to use the DaskTaskRunner for more complex flows.

Configure your flow to use the DaskTaskRunner:

1. Make sure the prefect-dask collection is installed by running pip install -U prefect-dask.
2. In your flow code, import DaskTaskRunner from prefect_dask.task_runners.
3. Assign it as the task runner when the flow is defined using the task_runner=DaskTaskRunner argument.
4. Use the .submit method when calling task-decorated functions.

Example code:

from prefect import flow, task
from prefect_dask.task_runners import DaskTaskRunner


@task
def say_hello(name):
    print(f"hello {name}")


@task
def say_goodbye(name):
    print(f"goodbye {name}")


@flow(task_runner=DaskTaskRunner())
def greetings(names):
    for name in names:
        say_hello.submit(name)
        say_goodbye.submit(name)


if __name__ == "__main__":
    greetings(["arthur", "trillian", "ford", "marvin"])

Note that, because you're using DaskTaskRunner in a script, you must use if __name__ == "__main__": or you'll see warnings and errors.

Run dask_flow.py. If you get a warning about accepting incoming network connections, that's okay - everything is local in this example.

python dask_flow.py

DaskTaskRunner automatically creates a local Dask cluster, then starts executing all of the task runs in parallel. The results do not return in the same order as the sequential code above.

Abbreviated output:

goodbye marvin
hello arthur
goodbye ford
hello trillian

Notice what happens if you do not use the submit method when calling tasks:

from prefect import flow, task
from prefect_dask.task_runners import DaskTaskRunner


@task
def say_hello(name):
    print(f"hello {name}")


@task
def say_goodbye(name):
    print(f"goodbye {name}")


@flow(task_runner=DaskTaskRunner())
def greetings(names):
    for name in names:
        say_hello(name)
        say_goodbye(name)


if __name__ == "__main__":
    greetings(["arthur", "trillian", "ford", "marvin"])

Run the script:

python dask_flow.py

Once again, the tasks run sequentially. Here's the output with logs removed.

hello arthur
goodbye arthur
hello trillian
goodbye trillian
hello ford
goodbye ford
hello marvin
goodbye marvin

The task runs are not submitted to the DaskTaskRunner; instead, they run sequentially.

Run tasks in parallel with Ray

You can easily switch to Ray as another parallel task runner option. Use the RayTaskRunner instead of DaskTaskRunner.

To configure your flow to use the RayTaskRunner:

1. Install prefect-ray into your environment with pip install -U prefect-ray.
2. In your flow code, import RayTaskRunner from prefect_ray.task_runners.
3. Specify the task runner when the flow is defined using the task_runner=RayTaskRunner argument.

Ray environment limitations

While we're excited about parallel task execution via Ray, there are a few limitations with Ray you should be aware of:

• Support for Python 3.11 is experimental.
• Ray support for non-x86/64 architectures such as ARM/M1 processors with installation from pip alone and will be skipped during installation of Prefect. It is possible to manually install the blocking component with conda. See the Ray documentation for instructions.
• Ray's Windows support is currently in beta.

See the Ray installation documentation for further compatibility information.

Save this code in ray_flow.py.

from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner

@task
def say_hello(name):
    print(f"hello {name}")

@task
def say_goodbye(name):
    print(f"goodbye {name}")

@flow(task_runner=RayTaskRunner())
def greetings(names):
    for name in names:
        say_hello.submit(name)
        say_goodbye.submit(name)

if __name__ == "__main__":
    greetings(["arthur", "trillian", "ford", "marvin"])

Now run ray_flow.py RayTaskRunner automatically creates a local Ray instance, then immediately starts executing all of the tasks in parallel. If you have an existing Ray instance, you can provide the address as a parameter to run tasks in the instance. See Running tasks on Ray for details.

Using multiple task runners

Many workflows include a variety of tasks, and not all of them benefit from parallel execution. You'll most likely want to use the Dask or Ray task runners and spin up their respective resources only for those tasks that need them.

Because task runners are specified on flows, you can assign different task runners to tasks by using subflows to organize those tasks.

This example uses the same tasks as the previous examples, but on the parent flow greetings() we use the default ConcurrentTaskRunner. Then we call a ray_greetings() subflow that uses the RayTaskRunner to execute the same tasks in a Ray instance.

from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner

@task
def say_hello(name):
    print(f"hello {name}")

@task
def say_goodbye(name):
    print(f"goodbye {name}")

@flow(task_runner=RayTaskRunner())
def ray_greetings(names):
    for name in names:
        say_hello.submit(name)
        say_goodbye.submit(name)

@flow()
def greetings(names):
    for name in names:
        say_hello.submit(name)
        say_goodbye.submit(name)
    ray_greetings(names)

if __name__ == "__main__":
    greetings(["arthur", "trillian", "ford", "marvin"])

If you save this as ray_subflow.py and run it, you'll see that the flow greetings runs as you'd expect for a concurrent flow, then flow ray-greetings spins up a Ray instance to run the tasks again.