prefect-ray
Accelerate your workflows by running tasks in parallel with Ray
Ray can run your tasks in parallel by distributing them over multiple machines.
The prefect-ray
integration makes it easy to accelerate your flow runs with Ray.
Install prefect-ray
The following command will install a version of prefect-ray
compatible with your installed version of prefect
.
If you don’t already have prefect
installed, it will install the newest version of prefect
as well.
Upgrade to the latest versions of prefect
and prefect-ray
:
Ray limitations
There are a few limitations with Ray:
- Ray support for Python 3.12 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 withconda
. See the Ray documentation for instructions. - Ray support for Windows is currently in beta.
See the Ray installation documentation for further compatibility information.
Run tasks on Ray
The RayTaskRunner
is a Prefect task runner that submits tasks to Ray for parallel execution.
By default, a temporary Ray instance is created for the duration of the flow run.
For example, this flow counts to three in parallel:
If you already have a Ray instance running, you can provide the connection URL via an address
argument.
To configure your flow to use the RayTaskRunner
:
- Make sure the
prefect-ray
collection is installed as described earlier:pip install prefect-ray
. - In your flow code, import
RayTaskRunner
fromprefect_ray.task_runners
. - Assign it as the task runner when the flow is defined using the
task_runner=RayTaskRunner
argument.
For example, this flow uses the RayTaskRunner
with a local, temporary Ray instance created by Prefect at flow run time.
This flow uses the RayTaskRunner
configured to access an existing Ray instance at ray://192.0.2.255:8786
.
RayTaskRunner
accepts the following optional parameters:
Parameter | Description |
---|---|
address | Address of a currently running Ray instance, starting with the ray:// URI. |
init_kwargs | Additional kwargs to use when calling ray.init . |
The Ray client uses the ray:// URI to indicate the address of a Ray instance.
If you don’t provide the address
of a Ray instance, Prefect creates a temporary instance automatically.
Run tasks on a remote Ray cluster
When using the RayTaskRunner
with a remote Ray cluster, you may run into issues that are not seen when using a local Ray instance.
To resolve these issues, we recommend taking the following steps when working with a remote Ray cluster:
- By default, Prefect will not persist any data to the filesystem of the remote ray worker. However, if you want to take advantage of Prefect’s caching ability, you will need to configure a remote result storage to persist results across task runs.
We recommend using the Prefect UI to configure a storage block to use for remote results storage.
Here’s an example of a flow that uses caching and remote result storage:
- If you get an error stating that the module ‘prefect’ cannot be found, ensure
prefect
is installed on the remote cluster, with:
- If you get an error with a message similar to “File system created with scheme ‘s3’ could not be created”, ensure the required Python modules are installed on both local and remote machines. For example, if using S3 for storage:
- If you are seeing timeout or other connection errors, double check the address provided to the
RayTaskRunner
. The address should look similar to:address='ray://<head_node_ip_address>:10001'
:
Specify remote options
The remote_options
context can be used to control the task’s remote options. For example, we can set the number of CPUs and GPUs to use for the process
task:
Resources
Refer to the prefect-ray
SDK documentation to explore all the capabilities of the prefect-ray
library.
For further assistance using Ray, consult the Ray documentation.