Advanced Topics
Contents
Advanced Topics#
Skipping Checkpoints#
Ray Workflows provides strong fault tolerance and exactly-once execution semantics by checkpointing. However, checkpointing could be time consuming, especially when you have large inputs and outputs for workflow tasks. When exactly-once execution semantics is not required, you can skip some checkpoints to speed up your workflow.
Checkpoints can be skipped by specifying checkpoint=False
:
data = read_data.options(**workflow.options(checkpoint=False)).bind(10)
This example skips checkpointing the output of read_data
. During recovery, read_data
would be executed again if recovery requires its output.
If the output of a task is another task (i.e., for dynamic workflows), we skip checkpointing the entire task.
Use Workflows with Ray Client#
Ray Workflows supports Ray Client API, so you can submit workflows to a remote
Ray cluster. This requires starting the Ray cluster with the --storage=<storage_uri>
option
for specifying the workflow storage.
To submit a workflow to a remote cluster, all you need is connect Ray to the cluster before submitting a workflow. No code changes are required. For example:
import subprocess
import ray
from ray import workflow
@ray.remote
def hello(count):
return ["hello world"] * count
try:
subprocess.check_call(
["ray", "start", "--head", "--ray-client-server-port=10001", "--storage=file:///tmp/ray/workflow_data"])
ray.init("ray://127.0.0.1:10001")
assert workflow.run(hello.bind(3)) == ["hello world"] * 3
finally:
subprocess.check_call(["ray", "stop"])
Warning
Ray client support is still experimental and has some limitations. One known limitation is that
workflows will not work properly with ObjectRefs as workflow task inputs. For example,
workflow.run(task.bind(ray.put(123)))
.