ray
Ray is a distributed computing framework that scales Python from a laptop to a cluster with two core primitives: tasks (stateless functions you decorate with @ray.remote and call with .remote()) and actors (stateful worker classes). On top of that core sit five AI libraries: Ray Data for scalable datasets, Ray Train for distributed training, Ray Tune for hyperparameter search, Ray Serve for model serving, and RLlib for reinforcement learning. It runs on bare machines, clouds, and Kubernetes (via KubeRay), and is the execution layer under vLLM deployments and many large training setups. It comes from UC Berkeley's RISELab and is commercially backed by Anyscale.
The strongest general-purpose choice for scaling Python ML workloads, with primitives simple enough to learn in an afternoon. The operational depth behind those primitives is where the cost lives, so do not adopt a cluster framework for a one-machine problem.
Use it if
- You have Python code that is embarrassingly parallel or needs to span machines, and you want to scale it without rewriting into Spark or MPI
- You are doing distributed ML work: multi-GPU training, hyperparameter sweeps with early stopping, or batch inference over datasets that do not fit one node
- You need stateful distributed services, like sharded model replicas or simulators, which the actor model expresses directly
- You are serving models with autoscaling and composition needs that a single FastAPI process cannot meet; Ray Serve handles replicas, batching, and multi-model pipelines
- Your workload fits on one machine; multiprocessing, joblib, or plain asyncio give you parallelism without a head node, a dashboard, and a distributed object store to operate
- You mainly do SQL-ish dataframe analytics; Spark, Dask, or DuckDB fit that shape better, and Ray Data is aimed at ML ingest, not BI
- You are not ready to operate a distributed system: debugging serialization errors, object store memory pressure, worker OOMs, and cluster autoscaling has a real learning curve, and the 2843 open issues reflect that surface area
- You just need background jobs with retries and scheduling; Celery or a hosted queue is simpler to run and reason about
Setup reality
pip install ray gets you the core, but the useful extras are separate: ray[data], ray[train], ray[tune], ray[serve], or ray[default] for the dashboard. The wheel is large (hundreds of MB installed) and pulls grpcio and protobuf, which can clash with your existing pins. Locally, ray.init() just works; a real cluster means head and worker nodes, matching Ray and Python versions on every machine, and understanding the object store (default 30% of RAM) before the first out-of-memory surprise. Everything you ship to a task must be picklable, and the error messages when it is not are famously unhelpful.
Patterns
Run a function as a distributed taskremote-task
import ray
ray.init()
@ray.remote
def square(x: int) -> int:
return x * x
ref = square.remote(4) # returns an ObjectRef immediately
print(ray.get(ref)) # 16.remote() is non-blocking and returns a future (ObjectRef); ray.get() blocks until the result is ready. Calling the function directly without .remote() raises an error.
Fan out many tasks and gather resultsparallel-map
import ray
@ray.remote
def process(item):
return item * 2
refs = [process.remote(i) for i in range(100)]
results = ray.get(refs) # one ray.get on the whole listLaunch all tasks first, then call ray.get once on the list. Calling ray.get inside the loop serializes everything and is the classic Ray anti-pattern.
Keep state in a distributed actorstateful-actor
import ray
@ray.remote
class Counter:
def __init__(self):
self.n = 0
def increment(self) -> int:
self.n += 1
return self.n
counter = Counter.remote()
print(ray.get(counter.increment.remote())) # 1
print(ray.get(counter.increment.remote())) # 2Each actor is a dedicated worker process and its methods run one at a time by default, so a single actor is also a serialization point; shard across several actors for throughput.
Put large data in the object store onceshare-large-objects
import ray
import numpy as np
big = np.zeros((10_000, 10_000))
big_ref = ray.put(big)
@ray.remote
def use(arr):
return arr.sum()
refs = [use.remote(big_ref) for _ in range(8)]
ray.get(refs)Passing big_ref sends the array to the shared object store once; passing big directly would re-serialize it into every task. Workers on the same node read it zero-copy.
Reserve CPUs or GPUs for a taskrequest-gpus-and-cpus
import ray
@ray.remote(num_gpus=1)
def train_shard(shard):
import torch
device = torch.device("cuda")
...
@ray.remote(num_cpus=4)
def heavy_cpu(work):
...Resource requests are logical, used for scheduling; Ray sets CUDA_VISIBLE_DEVICES but does not enforce actual usage. Tasks needing more GPUs than any node has will hang pending forever.
Process results as they completewait-for-first-results
import ray
refs = [work.remote(i) for i in range(20)]
while refs:
done, refs = ray.wait(refs, num_returns=1)
result = ray.get(done[0])
handle(result)ray.wait returns as soon as num_returns tasks finish, which keeps slow stragglers from blocking the pipeline the way a single ray.get over the whole list would.
Connect a script to an existing clusterconnect-to-cluster
import ray
# on the head node machine: ray start --head
# on workers: ray start --address='<head-ip>:6379'
ray.init(address="auto") # from a script on any cluster node
print(ray.cluster_resources())ray.init() with no address starts a throwaway local instance; address="auto" joins the running cluster. Python and Ray versions must match across all nodes or workers fail to start.
Batch transform a dataset with Ray Dataray-data-batch-inference
import ray
ds = ray.data.read_parquet("s3://bucket/data/")
def add_score(batch):
batch["score"] = batch["value"] * 2
return batch
ds = ds.map_batches(add_score, batch_format="pandas")
ds.write_parquet("/tmp/out/")Requires pip install 'ray[data]'. Execution is lazy: nothing runs until a consuming call like write_parquet, take, or iter_batches.
Hyperparameter search with Ray Tuneray-tune-sweep
from ray import tune
def objective(config):
score = (config["lr"] - 0.01) ** 2
tune.report({"loss": score})
tuner = tune.Tuner(
objective,
param_space={"lr": tune.grid_search([0.001, 0.01, 0.1])},
)
results = tuner.fit()
print(results.get_best_result(metric="loss", mode="min").config)Requires pip install 'ray[tune]'. Each trial runs as its own task or actor, so the sweep parallelizes across whatever resources the cluster has.
Serve a model over HTTP with Ray Serveray-serve-deployment
from ray import serve
from starlette.requests import Request
@serve.deployment(num_replicas=2)
class Model:
def __init__(self):
self.factor = 2
async def __call__(self, request: Request) -> dict:
data = await request.json()
return {"result": data["x"] * self.factor}
serve.run(Model.bind(), route_prefix="/predict")Requires pip install 'ray[serve]'. Each replica is an actor; scale with num_replicas or an autoscaling config, and compose deployments by passing bound handles into constructors.
Retry flaky tasks and bound waitstask-retries-and-timeouts
import ray
@ray.remote(max_retries=3, retry_exceptions=True)
def flaky(url):
return fetch(url)
ref = flaky.remote("https://example.com")
try:
result = ray.get(ref, timeout=30)
except ray.exceptions.GetTimeoutError:
handle_slow_task()By default Ray retries tasks only on worker crashes; retry_exceptions=True extends that to Python exceptions. The timeout on ray.get does not cancel the task, it just stops waiting.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| dask | PyPI | You want distributed numpy/pandas-style computation that feels closer to the PyData stack |
| pyspark | PyPI | Your organization runs Spark and the work is large-scale ETL and SQL analytics |
| celery | PyPI | You need distributed background task queues with retries and scheduling, not ML compute |