Workflows: Fast, Durable Application Flows¶
Workflows is available as alpha in Ray 1.7+. Expect rough corners and for its APIs and storage format to change. Please file feature requests and bug reports on GitHub Issues or join the discussion on the Ray Slack.
Ray Workflows provides high-performance, durable application workflows using Ray tasks as the underlying execution engine. It is intended to support both large-scale workflows (e.g., ML and data pipelines) and long-running business workflows (when used together with Ray Serve).
Flexibility: Combine the flexibility of Ray’s dynamic task graphs with strong durability guarantees. Branch or loop conditionally based on runtime data. Use Ray distributed libraries seamlessly within workflow steps.
Performance: Workflows offers sub-second overheads for task launch and supports workflows with hundreds of thousands of steps. Take advantage of the Ray object store to pass distributed datasets between steps with zero-copy overhead.
Dependency management: Workflows leverages Ray’s runtime environment feature to snapshot the code dependencies of a workflow. This enables management of workflows and virtual actors as code is upgraded over time.
You might find that workflows is lower level compared to engines such as AirFlow (which can also run on Ray). This is because workflows focuses more on core workflow primitives as opposed to tools and integrations.
Workflows provides the step and virtual actor durable primitives, which are analogous to Ray’s non-durable tasks and actors.
Steps are functions annotated with the
@workflow.step decorator. Steps are retried on failure, but once a step finishes successfully it will never be run again. Similar to Ray tasks, steps can take other step futures as arguments. Unlike Ray tasks, you are not allowed to call
ray.wait() on step futures, which ensures recoverability.
from ray import workflow @workflow.step def one() -> int: return 1 @workflow.step def add(a: int, b: int) -> int: return a + b output: Workflow[int] = add.step(100, one.step())
A workflow is an execution graph of steps created with
Workflow.run_async(). Once started, a workflow’s execution is durably logged to storage. On system failure, workflows can be resumed on any Ray cluster with access to the storage.
workflow.init(storage="/tmp/data") assert output.run(workflow_id="run_1") == 101 assert workflow.get_status("run_1") == workflow.WorkflowStatus.SUCCESSFUL assert workflow.get_output("run_1") == 101
Large data objects can be stored in the Ray object store. References to these objects can be passed into and returned from steps. Objects are checkpointed when initially returned from a step. After checkpointing, the object can be shared among any number of workflow steps at memory-speed via the Ray object store.
@ray.remote def hello(): return "hello" @workflow.step def words() -> List[ray.ObjectRef]: return [hello.remote(), ray.put("world")] @workflow.step def concat(words: List[ray.ObjectRef]) -> str: return " ".join([ray.get(w) for w in words]) workflow.init() assert concat.step(words.step()).run() == "hello world"
Workflows can generate new steps at runtime. When a step returns a step future as its output, that DAG of steps is dynamically inserted into the workflow DAG following the original step. This feature enables nesting, looping, and recursion within workflows.
@workflow.step def add(a: int, b: int) -> int: return a + b @workflow.step def fib(n: int) -> int: if n <= 1: return n return add.step(fib.step(n - 1), fib.step(n - 2)) assert fib.step(10).run() == 55
Virtual actors have their state durably logged to workflow storage. This enables the management of long-running business workflows. Virtual actors can launch sub-workflows from method calls and receive timer-based and externally triggered events. [Events support is under development.]
@workflow.virtual_actor class Counter: def __init__(self): self.count = 0 def incr(self): self.count += 1 return self.count workflow.init(storage="/tmp/data") c1 = Counter.get_or_create("counter_1") assert c1.incr.run() == 1 assert c1.incr.run() == 2