The Complete Ray Guide

This is part one of three. It covers everything you need to do real work with Ray, not a teaser. By the end you can turn an ordinary Python function into work that runs on every core of your machine, keep state in a long-lived worker, share large data without copying it, start a small cluster, submit a script to it, read the dashboard, and recognise the dozen errors that account for most beginner pain. The guide is written against Ray 2.58, the stable release at the time of writing. Mid-level and Senior take the same topics further; nothing here is thrown away.

Each section ends with a Try it task. Do them as you go. They take a few minutes each, and distributed computing only becomes intuitive once you have watched your own tasks fan out, finish out of order, and come back.

What Ray is, and the problem it solves

Ray is an open-source framework for running Python code on many processors at once, on one machine or across a cluster of machines, while keeping the programming model close to ordinary Python. You mark a function or a class as something that may run elsewhere, call it in a slightly different way, and Ray decides where it runs, moves the data it needs, and hands you the result.

To see why that is useful, start with what a Python program does by default: it runs one thread of work at a time. Python's global interpreter lock means that even if you start several threads, only one of them executes Python bytecode at any instant. If your laptop has ten cores, a plain Python loop uses roughly one of them. For a quick script this does not matter. For the workloads that machine learning produces it matters enormously: tuning forty hyperparameter combinations, resizing two million images, running a model over a night's worth of documents, or simulating ten thousand game episodes for reinforcement learning. All of those are collections of independent pieces of work, and the natural thing is to run the pieces side by side.

Before Ray, you had a few options, none of them comfortable. Python's built-in multiprocessing module spreads work over cores on one machine, but it stops at the machine boundary and gives you little help sharing large data. Cluster frameworks such as Apache Spark (see the Spark guide) scale across machines, but they ask you to express your program in their data-processing vocabulary, which fits table-shaped transformations well and fits "run this Python class on a GPU and keep it alive" poorly. Message queues and job schedulers work across machines but leave you writing the plumbing by hand: serialising arguments, tracking which worker has what, retrying failures. Ray's pitch is to give you the plumbing, with a small API that feels like Python.

The diagram is the whole core idea, and the rest of this guide fills it in.

Ray has two layers, and keeping them apart will save you confusion. Ray Core is the low-level runtime: tasks, actors and objects, which you will meet in the next section. On top of Core sit five libraries aimed at machine-learning work: Ray Data for loading and transforming data and running batch inference, Ray Train for distributed model training, Ray Tune for hyperparameter search, Ray Serve for online model serving (it has its own guide, Ray Serve), and RLlib for reinforcement learning. The libraries are built from Core primitives; understanding Core means you understand what the libraries are doing when something goes wrong.

What people use it for:

You need little to follow along: Python 3, a terminal, and a machine with a few cores. Nothing here requires a GPU or a cloud account.

The mental model: tasks, actors and objects

Ray Core has three nouns. If you can explain these to a colleague, you understand more of Ray than most people who have copied a tutorial.

A task is a function that runs somewhere else. You decorate an ordinary Python function with @ray.remote, and from then on you can call it with .remote() instead of directly. A task is stateless: it receives arguments, computes, returns a value, and is forgotten. Because tasks share nothing, Ray can run many at once and run each wherever there is room.

An actor is a class whose instance lives in its own worker process. You decorate a class with @ray.remote, create an instance with .remote(), and call its methods with .remote() too. Unlike a task, an actor remembers: its attributes persist between calls. Actors are the answer to "I need something that loads a large model once and then answers many requests" or "I need one place that keeps a running count". By default an actor handles one method call at a time, in the order received, so you do not need locks around its state.

An object is a value stored in Ray's distributed object store. Every .remote() call returns immediately with an ObjectRef, which is a handle to a result that may not exist yet. Think of it as a receipt, or in programming terms a future or promise. You turn the receipt into the actual value by calling ray.get(ref), which waits until the work is finished. Objects are immutable: once stored, a value never changes, which is what lets Ray share them freely between processes.

Surrounding these three nouns is the machinery that runs them.

A cluster is one head node plus any number of worker nodes. The head runs the Global Control Service (GCS), which stores cluster metadata, and the dashboard. Every node, the head included, runs a raylet, which decides what runs on that node and manages its slice of the object store. The driver is simply your own program, the process where you call ray.init(). Workers are the processes that actually execute tasks and host actors. On your laptop all of this lives on one machine; Ray starts it for you when you call ray.init(), which is why you can learn the whole model without a cluster.

One more concept deserves attention early: resources. Each node advertises logical quantities of CPUs, GPUs and memory, and each task or actor declares how much it needs (one CPU by default for a task). The scheduler places work where the declared resources are free. These are scheduling bookkeeping, not enforcement. If you say a task needs one CPU and it spins up eight threads, Ray will not stop it. Ray uses the numbers to avoid overbooking, so the honest numbers are your side of the bargain.

Installing Ray and checking the setup

Ray is installed with pip. Always do it inside a virtual environment so that Ray's dependencies stay separate from your system Python.

python3 -m venv .venv
source .venv/bin/activate
pip install -U "ray[default]"

The bare pip install -U "ray" gives you Core only. The default extra adds the dashboard, the Jobs system and the cluster launcher, which you will want from the second half of this guide onwards. If you plan to use the libraries, extras exist for them, such as ray[data,train,tune,serve]. To pin an exact release, which is good practice because Ray ships a new minor version every few weeks, write pip install "ray[default]==2.58.0".

On Windows, the commands are slightly different:

py -m venv .venv
.venv\Scripts\activate
pip install -U "ray[default]"

Support differs by platform. Linux on x86_64 and aarch64 and macOS on Apple silicon are the well-supported targets. Windows is marked beta in the documentation: multi-node clusters on Windows are untested and file opening is slower. If you are on Windows, running Ray inside WSL2 is a practical way to get Linux behaviour. Check the installation page for the range of Python versions the current release supports before you create your virtual environment, because the supported range moves between releases.

If you prefer conda, the conda-forge channel carries a ray-default package, and Docker images are published as rayproject/ray. When running Ray inside a Docker container, give the container shared memory, for example --shm-size=2gb. Without it, Ray warns that the object store is using /tmp instead of /dev/shm and runs noticeably slower.

Now verify the install two ways, once from the shell and once from Python:

ray --version
python -c "import ray; ray.init(); print(ray.cluster_resources())"

The second command starts a private one-machine Ray instance, prints the resources it found, and exits. The output is a dictionary along these lines, with numbers that depend on your hardware:

{'CPU': 10.0, 'memory': 14000000000.0, 'object_store_memory': 6000000000.0, 'node:127.0.0.1': 1.0}

Read it as a list of what Ray believes it has to spend: CPU is the number of logical cores it detected, memory and object_store_memory are byte counts, and the node: entry is a special per-node resource Ray uses internally. If you have a supported NVIDIA GPU and the right drivers, a GPU key appears too. The exact numbers will differ on your machine; what matters is that the CPU count matches your core count, because that number caps how many one-CPU tasks run at once.

Your first project: from a slow loop to parallel tasks

Create a file called first.py. The program below pretends to do slow work, first serially and then with Ray, and times both. The time.sleep stands in for real computation such as a feature calculation or an API call.

import time
import ray

ray.init()

def slow_square(x):
    time.sleep(1)          # pretend this is real work
    return x * x

@ray.remote
def slow_square_remote(x):
    time.sleep(1)
    return x * x

# Serial: ordinary Python, one at a time
start = time.time()
serial = [slow_square(i) for i in range(8)]
print("serial  :", serial, f"{time.time() - start:.1f}s")

# Parallel: submit everything, then collect
start = time.time()
refs = [slow_square_remote.remote(i) for i in range(8)]
parallel = ray.get(refs)
print("parallel:", parallel, f"{time.time() - start:.1f}s")

ray.shutdown()

Run it with python first.py. On a machine with at least eight cores, the output looks like this, with the serial run near eight seconds and the parallel one a little over one:

serial  : [0, 1, 4, 9, 16, 25, 36, 49] 8.0s
parallel: [0, 1, 4, 9, 16, 25, 36, 49] 1.3s

The extra third of a second is Ray's start-up cost: launching worker processes the first time they are needed. If your machine has four cores, expect the parallel run to take about two seconds, since only four one-CPU tasks run at a time and eight tasks need two rounds. That is the resource system doing its job.

Walk through the changes, because there are only four. First, ray.init() starts the runtime. Second, @ray.remote on the function turns it into a remote function. Third, calling it as slow_square_remote.remote(i) returns an ObjectRef at once, without waiting, so the list comprehension finishes in milliseconds having merely submitted eight tasks. Fourth, ray.get(refs) accepts a list of references and blocks until all of them are ready, returning the values in the same order as the list.

That ordering is worth saying out loud. The tasks may finish in any order, but ray.get on a list always returns results in the order you submitted them, so you can pair results with inputs without extra bookkeeping.

If you call the remote function directly, slow_square_remote(3), Ray raises an error telling you to use .remote() instead. That is the most common first mistake and it is a helpful one: the error is the framework preventing you from accidentally running something locally that you meant to distribute.

Tasks in depth

Now that the basic loop works, the details of tasks are what separate a script that happens to run from one you understand.

Tasks can call tasks. Inside a remote function you may call other remote functions and even pass their ObjectRefs along. If you pass an ObjectRef as an argument to another task, Ray waits for it to be ready and substitutes the real value before your function runs. This lets you build pipelines without ever calling ray.get in the middle:

import ray
ray.init()

@ray.remote
def load(n):
    return list(range(n))

@ray.remote
def total(numbers):
    return sum(numbers)

data_ref = load.remote(1000)
result_ref = total.remote(data_ref)   # pass the ref, not the data
print(ray.get(result_ref))            # 499500

Here the list produced by load never travels to your driver at all if the two tasks land on the same node; Ray resolves the reference where it is needed. That matters when the intermediate values are large.

Getting results as they finish. ray.get on a list waits for every task. When the tasks take uneven time and you would like to act on each result as soon as it arrives, use ray.wait. It takes a list of references and returns two lists: those that are ready and those that are not.

import time, random
import ray
ray.init()

@ray.remote
def work(i):
    time.sleep(random.random() * 3)
    return i

pending = [work.remote(i) for i in range(6)]
while pending:
    ready, pending = ray.wait(pending, num_returns=1)
    print("finished:", ray.get(ready[0]))

Each loop turn prints whichever task finished first, so the numbers appear out of order. ray.wait also accepts a timeout in seconds, which returns whatever is ready when time runs out. This is the right tool for processing a stream of results, or for giving up on stragglers.

Declaring what a task needs. By default a task asks for one CPU. You can say otherwise in the decorator or per call:

import ray
ray.init()

@ray.remote(num_cpus=2)
def heavy(x):
    return x * 2

@ray.remote
def light(x):
    return x + 1

print(ray.get(heavy.remote(5)))                        # uses the decorator's 2 CPUs
print(ray.get(light.options(num_cpus=0.5).remote(5)))  # override for this call

num_cpus, num_gpus and memory are the three you will use most. Fractional values are allowed, so num_gpus=0.5 lets two tasks share one GPU. Remember that these are promises to the scheduler. A task declared num_gpus=1 is simply given a GPU assignment; with PyTorch or TensorFlow you still have to put your tensors on that device. A task that needs more of a resource than any single node has can never run, and Ray reports it as infeasible; the errors section shows how that looks.

Failures. If your task raises an exception, Ray catches it, stores it, and re-raises it on the driver when you call ray.get, wrapped in a RayTaskError that includes the remote traceback. This means a bug in a remote function shows up in the place you are looking, with the line number from the worker. Separately, if a worker process dies, for example because the machine ran out of memory, Ray retries the task automatically a few times by default. You can control this with max_retries in the decorator. Retry on application exceptions is off by default, and you turn it on with retry_exceptions=True.

Idempotence is your responsibility. Because tasks may be retried, a task should be safe to run twice. Appending a row to a database inside a task, then having the worker crash after the insert but before returning, produces a duplicate on retry. Design tasks as "compute a value from inputs", and write side effects deliberately.

Actors: remembering things between calls

Tasks forget everything. Sometimes that is exactly wrong. Loading a 2 GB model for each of a thousand predictions would be absurd; you want to load it once and keep it. That is what an actor is for.

import ray
ray.init()

@ray.remote
class Counter:
    def __init__(self):
        self.n = 0

    def inc(self):
        self.n += 1
        return self.n

    def value(self):
        return self.n

c = Counter.remote()                       # starts a worker process, runs __init__ there
refs = [c.inc.remote() for _ in range(3)]
print(ray.get(refs))                       # [1, 2, 3]
print(ray.get(c.value.remote()))           # 3

Three things happen that are worth spelling out. Counter.remote() creates a dedicated worker process and runs __init__ inside it; this takes a moment, which is why you create actors up front, not per request. c.inc.remote() sends a message to that process and returns an ObjectRef for the result. And because the actor handles one call at a time in arrival order, the three increments return [1, 2, 3] without any locking.

The state lives in the actor's process, not in your driver. You cannot read c.n directly; attribute access on the handle does not reach across processes. You must define a method that returns it, as value does above, and call it with .remote().

Actors are the right shape for several common jobs:

• Holding a loaded model so that each call only pays for inference.
• Holding a connection or client, such as a database session, that is expensive to create and unsafe to pickle (serialise) across processes.
• Accumulating results, such as counters, metrics or a shared best-score tracker.
• Wrapping a stateful service, which is exactly how Ray Serve builds its replicas.

Actors also take resource declarations, and they hold their resources for as long as they live. This is the cause of a classic puzzle: you create four actors that each ask for one CPU on a four-core machine, and then your tasks never run, because every CPU is claimed by an idle actor. The fix is to size actors deliberately or give them num_cpus=0 when they are only coordinating.

Several actors can run side by side, which is how you get parallelism with state: create a handful of actors that each hold a model copy, and spread requests across them.

import ray
ray.init()

@ray.remote
class Scorer:
    def __init__(self, scale):
        self.scale = scale             # imagine loading a model here

    def score(self, x):
        return x * self.scale

scorers = [Scorer.remote(10) for _ in range(3)]
refs = [scorers[i % 3].score.remote(i) for i in range(9)]
print(ray.get(refs))                   # [0, 10, 20, 30, 40, 50, 60, 70, 80]

When you are finished with an actor, it goes away automatically once all handles to it are out of scope and the driver exits. To stop one explicitly, call ray.kill(actor_handle). You can also give an actor a name, Counter.options(name="counter").remote(), and fetch it from another script with ray.get_actor("counter"). A named actor with lifetime="detached" outlives the script that created it, which is useful and also a way to leak resources, so use it with intent.

Objects and the object store

Every value that flows between Ray tasks and actors lives in the object store, a region of shared memory on each node. You normally never touch it directly, but understanding it explains the speed, the memory errors and one habit that makes programs much faster.

When a task returns a value, Ray writes it into the object store of the node where the task ran and gives you an ObjectRef. When another task needs it, Ray copies it across the network if the task is on another node, or lets the task read it straight from shared memory if it is on the same node. For NumPy arrays this reading is zero-copy: the array in your worker points into the shared memory rather than copying it, which is why passing large arrays between tasks on one machine is cheap.

You can place a value in the store yourself with ray.put, which returns a reference. The reason to do this is that a big argument passed directly to many tasks is serialised again for every call, whereas a reference is tiny:

import numpy as np
import ray
ray.init()

big = np.random.rand(10_000_000)          # about 80 MB

@ray.remote
def head_sum(arr, k):
    return arr[:k].sum()

big_ref = ray.put(big)                    # store once
refs = [head_sum.remote(big_ref, k) for k in (10, 100, 1000)]
print(ray.get(refs))

Pass big_ref and the array is stored once and shared; pass big and every call serialises 80 MB again. Ray will usually warn you when a remote function captures a very large value in its closure, which is the same mistake in disguise.

The store has a fixed size per node. When it fills up, Ray spills older objects to disk, which keeps things working but slows them down, and if the store is full and nothing can be spilled you receive an ObjectStoreFullError. Objects are also reference-counted: when no ObjectRef points at a value any more, it is freed. This is why holding on to a giant list of references you no longer need keeps memory busy. Delete the list, or let it go out of scope, to release the data.

Two exceptions to know by name. ObjectLostError means an object disappeared, for example because the node that held it died and Ray could not rebuild it. OwnerDiedError means the process that created the reference, its owner, has died, so Ray no longer knows how to find the value. Both are rare on a laptop and common in clusters with unreliable machines; the senior material covers lineage reconstruction.

The cluster: starting nodes, the dashboard, and the CLI

Until now ray.init() quietly created a private one-machine instance that died with your script. Real work happens on a cluster that outlives scripts. A cluster is started with the ray command line tool, and you can build one entirely on your laptop to learn the shape.

Start a head node:

ray start --head

The command prints the address of the cluster, a command to add more nodes, a snippet to connect from Python, and the URL of the dashboard, which by default is http://127.0.0.1:8265. To join another machine as a worker you run ray start --address=<head_ip>:6379 on it, where 6379 is the default port of the Global Control Service. That is the whole procedure: one head, then each worker pointed at it. The machines must be able to reach each other over the network, and every node must run the same Ray version and the same Python version; a mismatch is refused on connect.

Now connect your script to the running cluster instead of creating a private one:

import ray

ray.init(address="auto")      # find the cluster started on this machine
print(ray.cluster_resources())

address="auto" looks for a running Ray instance on this machine. If none exists, you get a ConnectionError, which is the signal to run ray start --head first. You can also set the RAY_ADDRESS environment variable so that both ray.init() and the CLI find the cluster without any argument.

Open the dashboard in a browser. Its pages list nodes, jobs, actors and tasks with their state, CPU and memory use, and logs. During your experiments it is the fastest answer to "what is Ray doing right now?". The same information is available in the terminal through the State API commands:

ray status            # resources in use and available, plus autoscaler state
ray list actors       # every actor and whether it is alive
ray list tasks        # recent tasks and their states
ray list nodes        # the nodes in the cluster
ray summary tasks     # tasks grouped by function name and state
ray logs              # browse and fetch logs
ray memory            # which objects are using the object store
ray stop              # stop the Ray processes on this machine

ray status deserves a habit of its own. It shows the usage as used / total for each resource, and pending demands that cannot be met. When a program seems stuck, this single command usually tells you whether Ray is waiting for resources that do not exist.

On a laptop the head and a worker can both run locally for practice. After ray start --head, running ray start --address=127.0.0.1:6379 in a second terminal adds a second node on the same machine, which is a perfectly valid way to see two nodes listed in the dashboard. Finish with ray stop in each terminal.

On managed platforms, the head and workers are started for you. On Kubernetes the standard route is the KubeRay operator, which has its own place in the Kubernetes guide's ecosystem, and the commercial managed platform from Ray's creators is called Anyscale. You do not need either to complete this guide.

Submitting work as a job, and controlling the environment

Running python script.py on your laptop and connecting with ray.init(address=...) works while you are developing. For anything repeated, Ray offers Ray Jobs: you submit a script to the cluster, the cluster runs it, and you can check its status and logs afterwards, even after your terminal is closed. Jobs go through the dashboard's address, port 8265, so install ray[default] to have them.

Save a small script:

import ray

ray.init()

@ray.remote
def square(x):
    return x * x

print(ray.get([square.remote(i) for i in range(10)]))

Submit it from the directory that contains it:

ray job submit --address http://127.0.0.1:8265 --working-dir . -- python job_script.py

Everything after the double dash is the command the cluster will run. --working-dir . uploads the current directory so that the cluster can find your files, which matters when the cluster is on other machines that do not have your code. The command streams the job's output and prints a job identifier. Afterwards you can inspect it:

ray job status <job_id>
ray job logs <job_id>
ray job stop <job_id>

Jobs solve a real problem. If your driver runs on a laptop that sleeps, the work dies with it; a job lives on the cluster.

The same submission can be made from Python through JobSubmissionClient, which is the way a scheduler or a web application starts Ray work, but the command line is enough to start.

Submitting code to remote machines raises an obvious question: how do the right libraries get there? The answer is runtime_env, a description of the environment that a job, task or actor needs. You can set it for the whole program in ray.init:

import ray

ray.init(
    runtime_env={
        "pip": ["requests"],
        "env_vars": {"GREETING": "hello"},
        "working_dir": ".",
    }
)

@ray.remote
def read_env():
    import os
    return os.environ.get("GREETING")

print(ray.get(read_env.remote()))   # hello

pip lists packages to install for the workers, env_vars sets environment variables, and working_dir ships a folder of your code. Ray builds the environment on the workers on demand and caches it. This is far more convenient than logging into each machine to install packages, and it is why mismatched dependencies are less common with Ray than with hand-managed servers. It is not free: building a large environment takes time on first use, so for big dependencies like deep-learning frameworks you normally bake them into the machine image or container image instead.

The libraries on top of Core

You now know the foundation. This section is a guided tour of the five libraries, enough to recognise what each is for and to run a small example, not a complete treatment. All of them are installed with extras, such as pip install -U "ray[data,train,tune,serve]".

Ray Data loads, transforms and writes datasets that are bigger than one machine's memory, and is most often used for batch inference and preprocessing. A dataset is split into blocks that are processed in a streaming fashion, so memory stays bounded. Here is a small example that needs no files:

import ray

ds = ray.data.range(1000)                       # a dataset with an "id" column

def double(batch):
    batch["id"] = batch["id"] * 2
    return batch

ds = ds.map_batches(double)
print(ds.take(3))

map_batches runs your function over batches of rows, which is faster than row by row, and is the central operation for running a model over a dataset: you pass a class that loads the model once and use concurrency to say how many copies run. Real datasets come from ray.data.read_parquet("s3://...") and friends, and results are written with write_parquet. Some older Ray Data functions are deprecated in recent releases, such as Dataset.zip, so if a tutorial uses them, check the release notes.

Ray Train runs a training function across several workers, which can be GPUs or machines, and takes care of starting the processes, wiring them together and recovering from failures. You write a function that trains a model for one worker, and hand it to a trainer such as TorchTrainer with a scaling configuration saying how many workers and whether they use GPUs. Inside your function you report metrics and save checkpoints through Ray Train's APIs. Ray Train has gone through a major redesign; many older tutorials use an earlier interface, so always read the docs for the version you installed.

Ray Tune runs hyperparameter search. You describe a search space, such as learning rates to try, and Tune launches many trials in parallel, each a Ray task or actor, and can stop weak trials early. It integrates with experiment trackers such as MLflow. The API centres on tune.Tuner, which takes your training function, a param_space and a tune.TuneConfig, and .fit() runs the search. The older tune.run function appears in many tutorials and is the previous style.

Ray Serve serves models online. You define a deployment, a class or function that Ray runs as several replicas, and Serve handles the HTTP routing and scaling. A deployment looks like this:

from ray import serve

@serve.deployment(num_replicas=2)
class Hello:
    async def __call__(self, request):
        return {"message": "hello from Ray Serve"}

app = Hello.bind()
serve.run(app, route_prefix="/")

After serve.run, an HTTP server on port 8000 answers requests at /. The Serve guide, Ray Serve, covers deployment configuration, autoscaling and production use in full. For serving large language models, see also vLLM, which Ray Serve integrates with.

RLlib is the reinforcement learning library. It uses Core's actors to run many simulated environments in parallel and collect experience for training. It is a specialist tool, so most newcomers can skip it until they have a reinforcement learning problem.

How do you choose? The question to ask is what your work looks like:

Because all five sit on Core, they can be combined: a Tune run can launch Train jobs, and a Serve deployment can call a Data pipeline. They also coexist with the rest of the stack. Ray can read from Parquet and Delta Lake tables (Delta Lake), and training runs started from Ray report to trackers such as MLflow.

Configuration and resources you will actually touch

Ray has many settings, but a beginner needs only a handful, and most are arguments to ray.init or ray start.

ray.init() takes num_cpus and num_gpus to override what Ray detected, which is handy for testing. Setting num_cpus=4 on a ten-core laptop makes Ray behave as if it had four, so you can observe how your program behaves on a smaller machine. It also takes address to connect to an existing cluster, runtime_env as seen earlier, and namespace, which scopes named actors so that two jobs do not collide on a name. A namespace is for organisation, not security.

The same ideas apply on the command line: ray start --head --num-cpus=4 limits the head, and --dashboard-host controls which interface the dashboard listens on. The default listens on localhost only, which is the safe choice; changing it to 0.0.0.0 makes the unauthenticated dashboard reachable from the network, so do it only behind a firewall.

The environment variables you will meet first are RAY_ADDRESS for the default cluster address and RAY_DEDUP_LOGS, which controls the collapsing of repeated log lines from many workers into one. Treat any other RAY_ variable you find in a forum post as unfamiliar until you have checked the docs for your version, because many are internal tuning knobs that change between releases.

Logs live on each node in a session directory under /tmp/ray, with the newest session linked as session_latest. Output from print inside tasks and actors is streamed to the driver by default, prefixed with the function name and process ID, which is how you see (square pid=1234) in front of lines. When you debug, remember that those lines come from worker processes: a print in a task is not printed in your own process.

For memory, the two settings that matter early are the amount of object store memory and the per-task memory declaration. Ray has a memory monitor that kills workers when the node approaches full memory, so the system preserves itself rather than letting the operating system kill something at random. When that happens you see an out-of-memory error, covered next.

Common errors and how to read them

Most Ray errors fall into a few families. Learn the family and you can guess the cause before reading the message in full. Messages below are shown as they appear in current documentation where known; wording shifts between releases, so match on the exception class.

RayTaskError wrapping your exception. Not a Ray problem. It means the code in your task raised something, and the traceback beneath shows the line in your function. Read from the bottom: the last lines are your real error. Fix the function.

Calling a remote function directly. The error says remote functions cannot be called directly and must be called with .remote(). Replace f(x) with f.remote(x), and Counter() with Counter.remote().

Out-of-memory kills. An OutOfMemoryError saying a task was killed because the node was running low on memory means Ray's memory monitor killed a worker that was using too much. Reduce how many tasks run at once, declare memory= for heavy tasks so Ray schedules fewer per node, or pass references rather than copies. Use a larger machine if the work genuinely needs the memory.

ObjectStoreFullError. The object store ran out of room and spilling could not free enough. Hold fewer references at once, process results in a streaming fashion with ray.wait, and delete references you no longer need.

RayActorError. An actor's process died, so calls to it fail. Causes include a crash in your method, running out of memory, or a node loss. Look at the actor's logs. If you want automatic recovery, configure max_restarts on the actor, covered in the mid-level guide.

WorkerCrashedError. The worker running a task died unexpectedly, again typically from out-of-memory or a native library crashing. Ray retries tasks a few times, so the error appears when the retries are exhausted.

GetTimeoutError. You called ray.get(ref, timeout=...) and the deadline passed. Not a failure of the task itself, which may still be running.

A program that hangs with a warning about resources. If a task or actor asks for more than the cluster has, Ray cannot schedule it and prints a warning that the request cannot be scheduled right now. The call waits indefinitely. Run ray status in another terminal, compare the pending demand with the totals, and either reduce the request or add capacity. A common cause is actors holding all CPUs, as noted earlier.

Cannot connect to a cluster. ray.init(address="auto") raises a ConnectionError when there is no running head. Start one with ray start --head, or drop address to let Ray create a local instance. If you connect to a remote head and it hangs, a firewall is probably blocking the GCS port, 6379, or the ports Ray picks for its other services; open the ports between nodes.

Version mismatch. A worker or client started with a different Ray or Python version than the head is refused, with a message naming both sets of versions. Reinstall to align them. This is the most common failure when students join a second machine to the cluster.

Pickling errors. Ray serialises your functions and arguments with cloudpickle. Objects holding locks, open files, sockets or database connections cannot be serialised, and you see errors such as TypeError: cannot pickle '_thread.lock' object. Create such objects inside the task or actor, not outside it, and pass only plain data.

Pydantic version trouble. Ray 2.56 dropped support for Pydantic version 1. If a Ray import fails with a Pydantic complaint, upgrade to Pydantic 2.

An old tutorial's import fails. Ray has reorganised its libraries over time, for example retiring the ray.air namespace and redesigning Train. When ModuleNotFoundError points at a Ray module, check the documentation for your installed version before changing anything else.

Putting it all together

Here is one small project that uses almost everything above. It scores a batch of short texts with a pretend model. A Scorer actor holds the model, so loading happens once per actor; a task function splits the text into chunks; a shared lookup table is stored once with ray.put; and results are collected with ray.wait as they arrive.

import time
import ray

ray.init(num_cpus=4)

STOPWORDS = {"the", "a", "of", "and", "to"}
stop_ref = ray.put(STOPWORDS)            # shared once, passed by reference

@ray.remote
def clean(text, stop):
    words = [w for w in text.lower().split() if w not in stop]
    return " ".join(words)

@ray.remote
class Scorer:
    def __init__(self):
        time.sleep(1)                    # pretend to load a model
        self.calls = 0

    def score(self, cleaned):
        self.calls += 1
        return len(cleaned.split())      # pretend score: word count

    def count(self):
        return self.calls

texts = [
    "The cost of the cloud and the cost of GPUs",
    "A guide to the basics of Ray",
    "Tasks and actors and objects",
    "To scale a Python program",
    "The object store holds the data",
    "Jobs run on the cluster",
]

scorers = [Scorer.remote() for _ in range(2)]
cleaned_refs = [clean.remote(t, stop_ref) for t in texts]
score_refs = [scorers[i % 2].score.remote(ref) for i, ref in enumerate(cleaned_refs)]

pending = list(score_refs)
results = {}
while pending:
    ready, pending = ray.wait(pending, num_returns=1)
    idx = score_refs.index(ready[0])
    results[texts[idx]] = ray.get(ready[0])

for text, score in results.items():
    print(score, "|", text)

print("calls per scorer:", ray.get([s.count.remote() for s in scorers]))
ray.shutdown()

Walk through it as a checklist of ideas. The stopword set is stored once with ray.put and the reference goes to every clean call. The clean tasks run in parallel and return references. Those references are passed straight to the actors' score methods, so cleaned text never visits the driver. Two Scorer actors share the load, and each pays its start-up cost once. ray.wait lets you handle each score as it finishes. The last line asks each actor how many calls it served, and the answer, three and three, shows that state really did live in the actors.

Run it and expect six lines of the form 5 | The cost of the cloud and the cost of GPUs in the order they finished, then calls per scorer: [3, 3]. Because of the one-second pretend model load, the whole run takes a couple of seconds.

Now take it one step further as an exercise. Start a head with ray start --head, change ray.init(num_cpus=4) to ray.init(address="auto"), and submit it as a job with ray job submit --working-dir . -- python project.py. Open the dashboard and find the two actors and the tasks. This is the complete everyday Ray workflow: write locally, test small, submit to a cluster, observe.

What you can now do, and what comes next

You can explain Ray's three nouns and the head, worker and raylet structure behind them. You can parallelise independent work with tasks, avoid the loop-with-ray.get trap, process results as they arrive with ray.wait, hold state in actors, share large values by reference, start a small cluster, open its dashboard, query it with ray status and ray list, submit a script as a job with a runtime_env, and read the common errors by family. You also know what each of the five libraries is for and which one fits which kind of work.

Keep three habits. Submit all work before you collect any, pass references instead of data, and assume every task can run twice. Keep two warnings in mind: Ray has no authentication by default, so never expose its ports publicly, and Ray releases often, so pin the version and read the release notes before upgrading. If your data or your employer's rules require that it stay in a particular country, the cluster's machines must be placed in a cloud region that satisfies it; for teams in the Gulf and Egypt that means choosing in-region infrastructure for the cluster, not just for storage.

Where to go next depends on what you want to do. The mid-level guide covers placement groups, fault tolerance in depth, Ray Data and Train in practice and the Jobs API. The senior guide covers running Ray on Kubernetes, high availability for the head node, security and multi-tenancy. Related guides in this catalogue: Ray Serve for online inference, Kubernetes as the usual production home, Docker for building the images your workers run, Apache Spark for a different take on distributed data, and MLflow for tracking what your Ray jobs produce.

Sources

• Ray documentation home
• Installing Ray
• Ray Core walkthrough
• Ray Core scheduling
• Ray security
• Ray Serve
• Ray releases on GitHub

Part one of three. A twenty-five-minute review of the Beginner Detailed track, then the questions interviewers ask. Mid-level reviews this plus its own material; Senior reviews all three.

The thirty-second answer

Ray is an open-source framework for running Python in parallel across the cores of one machine or the machines of a cluster. You mark a function as a remote task or a class as an actor, call it with .remote(), and get back an object reference that you resolve with ray.get. Ray schedules the work, moves the data through a shared object store, and retries what fails. On top of that core sit libraries for data processing, training, tuning, serving and reinforcement learning.

Then the line that shows you have used it: "submit all your tasks first and call ray.get once on the list, because a ray.get inside the loop turns parallel work back into serial work."

The model

Say the one-line summary of each library, because interviewers ask: Ray Data loads and transforms datasets and runs batch inference, Ray Train distributes training, Ray Tune searches hyperparameters, Ray Serve serves models online, RLlib does reinforcement learning. All of them are built from tasks, actors and objects.

Tasks versus actors

The interview answer is a rule of thumb: if the work needs to remember something between calls, use an actor; otherwise use a task. Add the cost: an actor holds its resources while it lives and processes calls in order, so a single busy actor becomes a bottleneck.

The commands to have ready

pip install -U "ray[default]"      # core plus dashboard, jobs, cluster launcher
ray --version
ray start --head                   # start a head node; prints the dashboard URL
ray start --address=<head_ip>:6379 # join a worker to the head
ray status                         # resource usage and pending demand
ray list actors                    # state API: actors, tasks, nodes
ray summary tasks                  # tasks grouped by function and state
ray logs                           # fetch logs
ray memory                         # what is in the object store
ray job submit --address http://127.0.0.1:8265 --working-dir . -- python script.py
ray job logs <job_id>
ray stop

And the Python you can write from memory:

import ray
ray.init()                                   # local instance, or ray.init(address="auto")

@ray.remote
def square(x):
    return x * x

refs = [square.remote(i) for i in range(4)]  # submit everything
print(ray.get(refs))                         # [0, 1, 4, 9]

@ray.remote
class Counter:
    def __init__(self):
        self.n = 0
    def inc(self):
        self.n += 1
        return self.n

c = Counter.remote()
print(ray.get([c.inc.remote() for _ in range(3)]))   # [1, 2, 3]

ref = ray.put(list(range(1000)))             # store once, share by reference
ready, pending = ray.wait(refs, num_returns=1)
ray.shutdown()

Know the default ports as well: 6379 for the GCS that workers connect to, and 8265 for the dashboard and the Jobs API.

The classic traps and why they happen

ray.get inside the submission loop. ray.get(f.remote(i)) waits for each task before submitting the next, so you pay Ray's overhead and get no parallelism. Submit all, then get the list.

Passing a large object to many tasks by value. Each call serialises the value again. ray.put it once and pass the reference.

Pulling intermediate results to the driver. ray.get on a task's output followed by passing it into the next task sends the data through your process. Pass the ObjectRef and Ray resolves it where the next task runs.

Capturing something unpicklable. Ray serialises functions and arguments with cloudpickle. A lock, an open file or a database connection captured in a closure fails with an error like TypeError: cannot pickle '_thread.lock' object. Create those objects inside the task or actor.

Idle actors holding every CPU. An actor keeps its declared resources for its whole life. Four actors, each asking for one CPU, on a four-core machine leave no CPU for tasks, and the program waits. ray status shows it.

Treating resource numbers as limits. num_cpus=1 is a scheduling declaration. It does not stop the code from using more. It only stops Ray from placing more work than the declared total on a node.

Mixed versions. Head, workers and clients must run the same Ray and Python versions, or the connection is refused.

Exposing the dashboard. Ray has no authentication by default, and anyone who can reach the dashboard or the cluster can run arbitrary code. Keep it on a private network.

Common interview questions

What problem does Ray solve, and how is it different from multiprocessing?

Python's global interpreter lock stops threads from running Python code in parallel, so parallelism needs multiple processes. The multiprocessing module does that on one machine. Ray gives you the same idea with a higher-level interface that also works across machines: it schedules tasks against declared resources, shares data through a distributed object store without you copying it by hand, handles worker failures with retries, and supports stateful actors. The same code runs on a laptop and on a cluster.

What is the difference between a task and an actor?

A task is a stateless function: arguments in, value out, then it is forgotten, so Ray can run many in parallel and retry them freely. An actor is a class instance living in its own process, whose attributes persist between method calls. Actors process calls one at a time in order, which removes the need for locks but also means one slow call blocks the others. Use tasks for independent work, actors for loaded models, clients and shared counters.

What does .remote() return, and what does ray.get do?

.remote() returns immediately with an ObjectRef, a handle to a result that may still be computing. ray.get(ref) blocks until the value is ready and returns it. Given a list of references, it returns the values in the order of the list, not the order they finished.

How do you process results as they finish rather than waiting for all?

With ray.wait(refs, num_returns=1), which returns a list of ready references and a list of pending ones. Loop on it, handling each ready result, until nothing is pending. It also accepts a timeout.

Why might your Ray program be slower than the plain Python version?

Common causes: calling ray.get inside the loop so nothing runs concurrently; tasks so small that scheduling overhead exceeds the work, which you fix by batching many items into one task; copying large arguments for every call instead of using ray.put; and asking for more resources than exist so tasks queue. Start by running ray status and looking at the dashboard.

What is the object store?

A region of shared memory on each node that holds the values passed between tasks and actors. Values are immutable. On one node, workers can read NumPy arrays from it without copying. Across nodes, Ray transfers the data. When it fills up, Ray spills objects to disk, and if it cannot free room you get ObjectStoreFullError. Objects are freed when no references remain.

What are the head node and the GCS?

A cluster has one head node and any number of workers. The head runs the Global Control Service, which stores cluster metadata such as the actor directory, and the dashboard. Every node, including the head, runs a raylet that schedules work locally and manages that node's object store. In the default setup the head is a single point of failure, which the senior track covers.

How do you give a task a GPU?

Declare it: @ray.remote(num_gpus=1) or .options(num_gpus=1). Ray then assigns the task a GPU and sets the visible devices for that worker. The framework inside, such as PyTorch, still has to place tensors on the device. Fractions like num_gpus=0.5 let tasks share a GPU. If no node has enough GPUs, the task never runs.

How do you run a Ray program on a cluster?

Start a head with ray start --head, join workers with ray start --address=<head_ip>:6379, then either connect a script with ray.init(address="auto") or submit it as a job with ray job submit --working-dir . -- python script.py. A job runs on the cluster and survives your terminal closing, and you can fetch its logs and status later.

What is runtime_env?

A description of the environment a job, task or actor needs: pip packages, environment variables, a working directory. Ray builds it on the workers and caches it. It avoids installing packages by hand on every machine. Heavy dependencies such as deep-learning frameworks are better baked into the image, because building them on demand is slow.

What happens if a task fails or its worker dies?

An exception in your code is stored and re-raised when you call ray.get, wrapped in a RayTaskError that includes the remote traceback. If the worker process dies, Ray retries the task, three times by default for tasks, controlled by max_retries. That means tasks should be safe to run more than once.

Name the Ray libraries and what each does.

Ray Data for loading, transforming and batch-inferring over big datasets; Ray Train for distributed training with PyTorch, Lightning, XGBoost and others; Ray Tune for hyperparameter search with early stopping; Ray Serve for online serving with replicas and autoscaling; RLlib for reinforcement learning. They are all built on tasks, actors and objects.

When would you not use Ray?

When the work is a single sequential computation with nothing to parallelise; when a table-shaped transformation is better expressed in a data engine such as Spark; when a built-in tool like concurrent.futures is enough for a few cores and a short script; or when you cannot secure the cluster's network, since Ray has no default authentication.

How does Ray relate to Kubernetes?

Ray can run anywhere you can start the processes. On Kubernetes the usual route is the KubeRay operator, which defines resources for clusters, jobs and services. Kubernetes handles the machines and containers; Ray handles scheduling Python work across them. See the Kubernetes guide.

What is the relationship between Ray and Ray Serve?

Ray Serve is the serving library built on Ray Core. A Serve deployment is a class or function that Ray runs as several replicas, which are actors, behind an HTTP proxy. Understanding actors therefore explains most of Serve's behaviour. See the dedicated Ray Serve guide.

Why do versions matter so much with Ray?

Ray ships a new minor release every few weeks and does not promise compatibility between minor versions across nodes, so the head, workers and client must match, and so must the Python version. APIs also move: Ray Train and the old ray.air namespace changed substantially, so old tutorials break. In practice you pin the version and read the release notes before upgrading.

How would you share a large dataset with many tasks?

Put it in the object store once with ray.put, then pass the returned reference to each task. Each task then reads the same stored copy, which on one node is a zero-copy read for NumPy arrays, rather than receiving its own serialised copy per call.

Follow-up questions interviewers like

You said tasks are retried. When would that hurt you?

When a task has side effects. If a worker crashes after the task wrote a row to a database but before it returned, the retry writes the row again. The remedy is to make tasks idempotent: derive an output path or key from the input so a repeat overwrites, or check for an existing result first. Mentioning this shows you think about failure, not just the happy path.

How would you decide how big a task should be?

Large enough that the work outweighs the fixed cost of scheduling, serialising arguments and storing the result, and small enough that there are many more tasks than CPUs, so that the scheduler can balance uneven durations. In practice you measure one task, then group items into chunks until a task runs for at least tens of milliseconds. If a hundred thousand tiny tasks are slower than the loop, batching is the fix.

Your job works on a laptop but hangs on the cluster. What do you check first?

Run ray status and compare pending demand with available resources, because a task that asks for a GPU on a CPU-only cluster waits forever. Then check that the Ray and Python versions match across nodes, that the code and dependencies reached the workers through --working-dir and runtime_env, and that no firewall blocks the GCS port between machines. The dashboard's task view shows which tasks are pending and why.

What would you tell a teammate who wants to put the dashboard on a public address?

Not to. Ray does not authenticate by default, so anyone reaching it can run code on the cluster. Keep it on a private network, reach it through an SSH tunnel or an authenticating proxy, and treat token authentication as an additional layer only.

What a strong answer covers that a weak one misses

A weak answer says "Ray is for distributed computing" and stops. A strong answer names tasks, actors and objects, explains why the first and last are stateless and immutable, and mentions the specific mistake of calling ray.get in a loop. A weak answer lists the libraries; a strong answer says they are built on the same core and chooses between them by workload. A weak answer ignores security; a strong answer says that Ray needs network-level protection because there is no default authentication. A weak answer says "it's faster"; a strong answer says "faster when the work is parallel and large enough to outweigh scheduling overhead".

Sixty-second self-test

Answer aloud, then check against the sections above.

1. What does .remote() return, and how do you get the value?
2. Which is stateful, a task or an actor?
3. Why is [ray.get(f.remote(i)) for i in range(8)] slow?
4. What do ray.put and ray.wait do?
5. Which port do workers join, and which serves the dashboard?
6. What does num_cpus=2 promise, and what does it not enforce?
7. Name the command that shows pending resource demand.
8. What must match across head, workers and client?
9. How do you submit a script so it survives your terminal closing?
10. Why is it dangerous to expose the dashboard?

Part one of three. Most Ray beginner problems come from four sources: waiting too early, copying too much, asking for resources that do not exist, and following a tutorial written for a different version. Start with the table, then read the entries you recognise, then adopt the habits.

Common errors at this level

The mistakes in detail

Mistake 1: waiting inside the loop

Symptom. You converted a loop to Ray and it runs in the same time as before, or slower.

Cause. ray.get(f.remote(i)) blocks until that one task finishes. The next iteration starts only afterwards, so at most one task runs at a time, and you have added scheduling overhead.

Fix. Separate submission from collection:

refs = [f.remote(i) for i in range(100)]   # returns instantly
results = ray.get(refs)                    # waits once

A good test of your own code: the loop that calls .remote() should finish almost instantly. If it takes as long as the work, something in it is waiting.

Mistake 2: tasks that are too small

Symptom. A hundred thousand tiny tasks take longer than the plain loop.

Cause. Every task has a fixed cost: it is scheduled, its arguments are serialised, and its result is stored. If the work in a task takes microseconds, the overhead dominates.

Fix. Batch. Instead of one task per item, give each task a chunk of a few hundred or thousand items and have it loop internally:

@ray.remote
def process_chunk(items):
    return [expensive(i) for i in items]

chunks = [data[i:i + 1000] for i in range(0, len(data), 1000)]
results = ray.get([process_chunk.remote(c) for c in chunks])

As a rough guide, aim for tasks that run for at least tens of milliseconds, and often much longer. Measure rather than guess.

Mistake 3: sending the same big object over and over

Symptom. Memory climbs, or calls that should be fast are slow, and Ray warns that a function is very large.

Cause. A large array or model passed directly into many calls is serialised each time. A large object captured in a function's closure is serialised with the function.

Fix. Store it once and pass the reference:

model_ref = ray.put(big_model)
refs = [predict.remote(model_ref, batch) for batch in batches]

Inside the task, Ray replaces the reference with the real object before your code runs.

Mistake 4: calling ray.get on intermediate results

Symptom. A pipeline of two steps is slow, and the driver's memory grows.

Cause. The first step's output travels to your driver through ray.get, then travels out again as the second step's input.

Fix. Pass the reference:

a_ref = step_one.remote(x)
b_ref = step_two.remote(a_ref)    # Ray resolves a_ref where step_two runs
final = ray.get(b_ref)

Mistake 5: actors that starve the tasks

Symptom. You create some actors, then your tasks never start. No error, only a pending warning.

Cause. Each actor reserves its declared resources for as long as it lives, by default a CPU. Idle actors still hold them.

Fix. Run ray status, read the used and total columns, and size actors on purpose. If an actor only coordinates and does little computing, give it num_cpus=0. If you have many actors, count how many CPUs they take before launching tasks.

Mistake 6: one actor serving everyone

Symptom. Throughput does not rise when you add more clients.

Cause. An actor runs one method at a time. All calls queue behind one another in one process.

Fix. Create several actors and spread calls among them, as a pool. For a service, use Ray Serve, which manages replicas for you. See Ray Serve.

Mistake 7: relying on side effects in tasks

Symptom. After a crash, you find duplicate rows, duplicate files or doubled counts.

Cause. Ray retries tasks when a worker dies. A task that writes to an external system and then crashes before returning will run again and write again.

Fix. Make tasks idempotent: write to a path derived from the input so a repeat overwrites rather than appends, or check whether the result already exists. Return data and write it from a single place when you can.

Mistake 8: unpicklable captures

Symptom. TypeError: cannot pickle ... pointing at a lock, a socket or a client object.

Cause. Functions and arguments are serialised with cloudpickle so they can travel to other processes. Some objects, such as thread locks, open file handles and many database connections, cannot be serialised.

Fix. Create the object where it is used:

@ray.remote
def query(sql):
    import sqlite3                       # open the connection on the worker
    conn = sqlite3.connect("data.db")
    try:
        return conn.execute(sql).fetchall()
    finally:
        conn.close()

For an expensive connection, make the task an actor method and open it in __init__.

Mistake 9: trusting the resource numbers too much

Symptom. The machine freezes or swaps even though Ray says only four tasks run.

Cause. num_cpus is a scheduling promise. If each task starts a multi-threaded library such as a BLAS-backed NumPy or a data-loading pool, it can use many threads regardless of the declaration.

Fix. Limit threading inside tasks, with the library's own setting or environment variables such as OMP_NUM_THREADS, and declare num_cpus honestly to match. Declare memory= for tasks with a large footprint so that Ray schedules fewer of them per node.

Mistake 10: mixing versions

Symptom. A second machine will not join, or a client cannot connect, and the message names two sets of versions.

Cause. Ray requires the same Ray version and the same Python version across the head, the workers and any client.

Fix. Use a requirements file with a pinned version, ray[default]==2.58.0, and build every machine or container from it. Check ray --version and python --version on each node. Docker images with a pinned tag make this simple.

Mistake 11: following an old tutorial

Symptom. An import fails, or a function does not exist, in code copied from a blog.

Cause. Ray changes quickly. Ray Train was redesigned, the ray.air namespace was retired, and some Ray Data functions were deprecated and then removed. The older Serve start-up calls were replaced by serve.run and serve.deployment.

Fix. Look at the date and the Ray version in the tutorial. Compare with the documentation page for your installed version. When in doubt, write the example from the current docs, and keep the version pinned so that your own code does not rot between lessons.

Mistake 12: exposing the cluster

Symptom. Nothing visibly wrong, which is the danger.

Cause. Ray has no authentication by default. Anyone who can reach the dashboard on 8265 or the cluster's ports can submit code that runs on your machines. Cloud virtual machines with a public IP and a wide-open security group are the usual way this happens.

Fix. Keep Ray on a private network. Use an SSH tunnel to reach the dashboard from your laptop, for example ssh -L 8265:127.0.0.1:8265 user@head-host. Restrict security groups to known addresses. Token authentication, available since Ray 2.52, adds a layer but does not replace network isolation.

Habits worth adopting on day one

Pin the version. Write ray[default]==2.58.0 into your requirements and set the same in any container. Ray has a new minor release every few weeks, and a pin turns a surprise into a decision. Read the release notes before moving the pin.

Develop on small resources. Use ray.init(num_cpus=2) while writing code. If a program works with two CPUs, it will work with two hundred, and scheduling behaviour is easier to follow when there is little of it.

Look at ray status when anything stalls. It is the fastest way to find out whether Ray is waiting for resources. Make it a reflex before searching the web.

Keep the dashboard open. The task and actor views, the logs and the node charts answer most "what is it doing?" questions. Learn where they are while nothing is broken.

Measure one task before you scale. Time a single call locally, then predict the parallel time: tasks times duration divided by CPUs. If reality is far off, you have found overhead, a bottleneck or a hidden serial step. Do the calculation before you start a big run.

Make tasks pure where you can. A task that returns a value computed only from its inputs can be retried, reordered and run anywhere. Put side effects in a small number of well-understood places.

Hold references deliberately. A reference keeps its object alive in the store. After you have used a large result, drop the reference, with del, or let it leave scope, to release the memory.

Handle large work as a stream. When results are large or numerous, process them with ray.wait as they arrive instead of collecting everything in one list. Memory stays bounded and you see the first results sooner.

Put the environment in a file. For anything beyond a laptop, describe dependencies in a runtime_env or, better, in a container image: Docker is the usual tool. A cluster that depends on hand-installed packages becomes irreproducible quickly.

Test against a failure. Once your program works, kill a worker process on purpose, or add max_retries=0 and see what breaks, to learn how it behaves when something goes wrong. It is far easier to find a non-idempotent task in a rehearsal than in a production incident.

Write results to durable storage. Local disk on a cluster node vanishes when the node does. In clusters, save outputs to object storage, a database or a shared file system, not to the worker's own filesystem.

Practice cards

Each card is a small exercise. Do not look at the answer until you have tried.

Card 1: predict the time. Eight tasks each sleep two seconds. The machine has four CPUs and each task needs one. How long does ray.get on all eight take, roughly? Then run it with ray.init(num_cpus=4).

Card 2: find the bug. This code is meant to run in parallel and does not:

@ray.remote
def work(i):
    return i * 2

out = []
for i in range(10):
    out.append(ray.get(work.remote(i)))

Card 3: why does it hang? On a four-CPU machine you create four actors with default settings and then submit a task. Nothing happens. Why, and what is the one-command diagnosis?

Card 4: share it properly. You have a 500 MB array that fifty tasks read. Rewrite the call so that the array is stored once.

Card 5: stateful or not? You need to keep a loaded tokenizer and apply it to many texts. Task or actor, and why?

Card 6: read the error. ray.init(address="auto") fails with a connection error. Give two fixes.

Recipes worth keeping

Run a script on a cluster so it survives your laptop closing.

ray job submit --address http://127.0.0.1:8265 --working-dir . -- python script.py
ray job logs <job_id>

Reach a remote dashboard safely.

ssh -L 8265:127.0.0.1:8265 user@head-host
# then open http://127.0.0.1:8265 in your browser

Check what is happening right now.

ray status
ray list tasks
ray summary tasks

Time a parallel section.

import time
start = time.time()
results = ray.get([f.remote(i) for i in range(100)])
print(f"{time.time() - start:.2f}s")

Reset a messy local state.

ray stop

Then start again. If the command reports processes that refuse to stop, check for leftover Python scripts holding Ray sessions.

Pin the install.

pip install "ray[default]==2.58.0"
pip freeze | grep -i ray

Record the output in the project's requirements file.

Reading a stuck program in three steps

When a Ray program stops making progress, resist the urge to restart it. A restart destroys the evidence. Work through three steps instead, each answering one question.

First, ask whether Ray is waiting for resources. Run ray status in another terminal. If a resource shows as fully used and there is pending demand listed beneath it, the problem is scheduling: something is holding the CPUs or GPUs, or a request is larger than any node. Fix the request or free the resources.

Second, ask which work is running. ray list tasks and ray list actors show states such as running, pending and failed, and ray summary tasks groups them by function name, so a function with thousands of pending entries stands out immediately. The dashboard shows the same information visually.

Third, ask what the workers are saying. Open the logs for the stuck task or actor from the dashboard or with ray logs. A worker that is silent is probably busy or blocked; a worker with a traceback tells you the cause directly. Only after these three steps should you decide whether to stop the job, change the code or add machines.

A short checklist

Before you trust a Ray program, go through this list.

• Is Ray pinned to a version, and does every node have the same Ray and Python?
• Does the submission loop call .remote() only, with a single ray.get after it?
• Are tasks large enough that overhead does not dominate?
• Are large shared values stored once with ray.put?
• Are intermediate results passed as references instead of through ray.get?
• Are actors sized deliberately, and not holding CPUs the tasks need?
• Are tasks safe to run twice?
• Do locks, files and connections get created inside tasks and actors?
• Is output written to durable storage rather than a worker's local disk?
• Is the dashboard on a private network, behind a tunnel or a proxy, and not on the internet?
• Have you read the release notes for the version you are using?

If you can tick each of those, you have avoided most of the problems that bring beginners to forums. The next step is the mid-level track, which covers placement, fault tolerance and the libraries in more depth, together with Ray Serve for taking a model online.

This is part two of three. It picks up where Beginner ended and takes every topic from there further, then adds the machinery you have not met yet. It is written against Ray 2.58, and Ray releases every few weeks, so treat the version number as part of every claim and check the release notes before you copy a pattern into production.

Where this picks up

Each section starts with the problem it solves and ends with a Try it you can run in a few minutes. Where the official documentation names something differently in an older tutorial, the current name is used and the old one is flagged.

How Ray runs your code, precisely

Beginner gave you three nouns: tasks, actors and objects. To predict Ray's behaviour you need the machinery that connects them. A Ray cluster has one head node and any number of worker nodes. Every node runs a raylet, which is the per-node scheduler and the manager of that node's object store. The head node additionally runs the GCS (Global Control Service), which holds cluster-wide metadata: which nodes exist, which actors live where, and which placement groups have been reserved.

Your script is the driver. When it calls f.remote(x), the driver does not run f. It creates a task description, registers a future (an ObjectRef), and hands the task to the local raylet. The raylet looks for a node with the declared resources free, preferring a node that already holds the task's input objects, and leases a worker process there. The worker runs the function and writes the return value into its node's object store. The driver learns the value is ready, and ray.get fetches it.

Two properties of this design explain most surprises. First, .remote() returns immediately, so a loop of a million calls queues a million tasks without waiting for any of them. Second, the process that submitted a task owns its result. If that owner dies, the object is lost even if the data still sits in a store somewhere, which is where OwnerDiedError comes from. Keep ownership in mind whenever a long-lived value is created inside a short-lived task.

Actors are processes with a mailbox

An actor is a dedicated worker process that keeps state between calls. Method calls on one actor run one at a time in submission order by default, which is why a Counter needs no lock and also why one slow method blocks every other caller. If you need concurrency inside an actor, you opt in with async def methods or the max_concurrency option. If you need more throughput than one process gives, you create several actors and spread calls across them, which is exactly what Ray Serve replicas and Ray Data actor pools do for you.

An actor reserves its resources for its whole lifetime. That is the usual cause of the warning that a request cannot be scheduled because all cluster resources are claimed by actors: a few idle actors holding CPUs can starve the tasks that would use them. The fix is to give actors only what they need, and to use num_cpus=0 for actors that mostly wait.

The anti-pattern that costs the most

The single most common performance bug is calling ray.get inside the submission loop, which turns a parallel program into a sequential one with network latency added to every step.

import ray
ray.init()

@ray.remote
def work(i):
    return i * i

# Sequential in disguise: each get waits for one task before the next is submitted
slow = [ray.get(work.remote(i)) for i in range(100)]

# Parallel: submit everything, then wait once
refs = [work.remote(i) for i in range(100)]
fast = ray.get(refs)

When results arrive at uneven speeds and you want to process them as they finish, use ray.wait instead of waiting for the whole list:

pending = [work.remote(i) for i in range(100)]
while pending:
    ready, pending = ray.wait(pending, num_returns=1, timeout=5)
    for ref in ready:
        handle(ray.get(ref))

The timeout matters in production: without it a single stuck task blocks the loop forever. With it you can log, cancel the straggler with ray.cancel(ref), and move on.

Objects, the object store and memory

Every node reserves a region of shared memory for the object store, so that many worker processes on one machine can read the same value without copying it. When a task returns a large value or you call ray.put, the value is serialised once into that store. Other tasks on the same node then read it through zero-copy access where the type allows it, with NumPy arrays being the classic case, and the result arrives as a read-only view. Tasks on other nodes pull a copy over the network on demand.

This is why you pass large data by reference. Passing an ObjectRef as an argument sends a small handle, and Ray resolves it to the real value on whichever worker runs the task. Capturing the same large array in a closure instead makes Ray serialise it into the function definition, which can be slow and wasteful, and Ray warns when a remote function is very large.

import numpy as np
import ray

ray.init()
big = np.zeros((10_000, 10_000))     # about 800 MB
big_ref = ray.put(big)               # stored once

@ray.remote
def total(arr):
    return float(arr.sum())          # arr is a read-only view over shared memory

print(ray.get([total.remote(big_ref) for _ in range(8)]))

When the store fills

The store has a fixed size per node. When it fills, Ray spills objects to disk to make room, and restores them when someone needs them. Spilling keeps jobs alive but is slow, and a job that spills continuously is telling you it holds too many objects at once. If spilling cannot keep up you see ray.exceptions.ObjectStoreFullError. The remedies, in order of effectiveness, are:

1. Hold fewer references. An object stays alive while any ObjectRef to it exists, so a long list of refs you never release pins memory.
2. Reduce how much runs at once, so fewer results exist simultaneously.
3. Use Ray Data, which streams blocks through the pipeline with backpressure rather than materialising everything.
4. Give the nodes more memory.

Seeing what holds memory

ray memory          # object references in the cluster and who holds them
ray status          # resource usage and autoscaler state
ray list actors     # State API: every actor and its state
ray summary tasks   # task counts grouped by function and state

ray memory is the tool to reach for when the store is full. It lists the objects, their sizes, and the reference type, so you can tell a driver variable that is pinning data from a task that is still running.

Resources, placement groups and scheduling

Ray's resources are logical. When you declare num_cpus=2, you are telling the scheduler to count two CPUs against the node's total while this task runs. Ray does not enforce the limit: a task declared with one CPU can start eight threads and use eight cores. The declaration exists so that the scheduler does not over-commit a node, which means your honesty about requirements is what makes scheduling work. Under-declare and nodes thrash; over-declare and cores sit idle.

@ray.remote(num_cpus=2, memory=4 * 1024**3)
def preprocess(path): ...

@ray.remote(num_gpus=0.5)
def small_model(batch): ...

# Per-call override without redefining the function
preprocess.options(num_cpus=4).remote("s3://bucket/part-0.parquet")

Fractional GPUs let two small models share one device, but Ray does not partition GPU memory. Both tasks still compete for the same VRAM, so you must make sure they fit. Custom resources such as resources={"fast_disk": 1} let you label special nodes and target them, which suits one-off hardware.

The memory option declares the heap memory a task needs, expressed in bytes. It is used for scheduling and, as the next section shows, for deciding which tasks to kill under pressure. Declaring it for memory-hungry tasks is one of the cheapest reliability improvements available.

Placement groups for gang scheduling

Some jobs need several resource bundles at once or not at all. A distributed training job that needs four GPUs on four different machines is useless with three. A placement group reserves bundles together, and the strategy decides how they spread.

from ray.util.placement_group import placement_group
from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy

pg = placement_group([{"CPU": 1, "GPU": 1}] * 2, strategy="STRICT_SPREAD")
ray.get(pg.ready())     # blocks until both bundles are reserved

ref = trainer_step.options(
    scheduling_strategy=PlacementGroupSchedulingStrategy(placement_group=pg)
).remote()

pg.ready() returns a future that completes when every bundle is reserved. If the cluster cannot satisfy the group it stays pending, and ray status shows the unmet demand, which is also what the autoscaler reads to decide it needs more nodes.

Scheduling strategies and labels

The default strategy is locality and load aware: it tries to run a task near its input data and spreads load sensibly. "SPREAD" forces tasks across nodes, useful when you want to avoid hot spots. NodeAffinitySchedulingStrategy pins work to a specific node, and it is a sharp tool, because a pinned task cannot move when the node dies or fills. The scheduling documentation describes labels as the newer, preferred mechanism for placing work on nodes with particular properties, so reach for a label selector before a hard node affinity.

Fault tolerance in practice

A cluster of many machines will lose some of them, and long jobs must expect that. Ray gives you separate knobs for tasks, actors and objects, and good production code sets them deliberately instead of accepting defaults.

Tasks are retried when the worker dies, controlled by max_retries. By default a task is retried only for system failures such as a crashed worker or lost node, not for exceptions your code raises. To retry on application exceptions as well, set retry_exceptions=True, which only makes sense for work that is safe to run twice. Retried tasks must be idempotent: if a task writes to a database or an output file, a retry may do it twice, so write to a unique path and commit atomically.

@ray.remote(max_retries=3, retry_exceptions=True)
def flaky_fetch(url):
    return fetch(url)       # safe to repeat: read-only

Actors do not restart by default. With max_restarts you let Ray recreate a dead actor, using -1 for unlimited restarts. A restarted actor runs __init__ again, so its in-memory state is gone: anything that must survive belongs in external storage or must be rebuilt on start. max_task_retries controls whether calls that were in flight during a crash are retried. Named actors created with lifetime="detached" outlive the driver that created them and are found again with ray.get_actor.

@ray.remote(max_restarts=-1, max_task_retries=2)
class Scorer:
    def __init__(self):
        self.model = load_model()          # rebuilt on every restart
    def score(self, x):
        return self.model(x)

svc = Scorer.options(name="scorer", lifetime="detached").remote()
same = ray.get_actor("scorer")

Objects that were lost with a node can be recreated through lineage reconstruction: Ray re-runs the task that produced them. It works only when that task is retryable and its inputs can still be obtained, so objects created by ray.put have no lineage and cannot be rebuilt if their owner dies.

The memory monitor and out-of-memory kills

Ray runs a memory monitor on each node. When node memory use crosses a threshold, which defaults to 95 percent, Ray kills a worker to protect the node and the task fails with ray.exceptions.OutOfMemoryError, message beginning "Task was killed due to the node running low on memory". Retriable tasks are preferred victims, so declaring max_retries affects which work dies. The configuration names and defaults are described in the memory management documentation, and you should confirm them for your version before tuning.

The fix is almost never the threshold. Declare memory= on the heavy tasks so the scheduler stops packing too many together, reduce concurrency, process in smaller batches, or move to larger nodes. Treat an OOM kill as information about your peak working set.

Runtime environments and Ray Jobs

Locally everything shares your Python environment. On a cluster, workers on other machines need your code and dependencies, and getting them there is a classic source of "works on my laptop" failures. Ray offers two mechanisms: bake dependencies into the image, or declare a runtime environment that Ray builds on demand.

ray.init(runtime_env={
    "working_dir": ".",
    "pip": ["requests==2.32.3"],
    "env_vars": {"STAGE": "staging"},
})

A runtime environment can be set for the whole job, or per task or actor with .options(runtime_env=...). Ray uploads the working directory, creates the environment on each node, and caches it. That is convenient for development, but installing packages at job start is slow and fragile in production: a package index outage becomes a job outage. For production, bake pinned dependencies into the container image and use the runtime environment only for small differences such as environment variables or your own code.

Submitting jobs

A Ray Job is a script submitted to a running cluster. The cluster runs it as a managed entrypoint with captured logs, a status, and the ability to stop it. Submission goes through the dashboard's HTTP endpoint on port 8265.

ray job submit --address http://HEAD_IP:8265 \
  --working-dir . \
  -- python train.py --epochs 3

ray job status <job_id>
ray job logs <job_id>
ray job stop <job_id>

The same operations are available from Python through JobSubmissionClient, which is what a CI pipeline or scheduler would call. Because the script runs on the cluster, your laptop can disconnect without killing it, the opposite of a driver attached from your laptop. Ray Client, the ray:// connection on port 10001, exists for interactive use, but jobs are the pattern for anything that must keep running.

The version rule

Head and workers must run the same Ray version and the same Python version. A mismatch fails at connect time with a message beginning "Version mismatch" that names both sides. The official position is that Ray does not guarantee compatibility between minor versions in one cluster, so the fix is never to tolerate drift: pin one image tag everywhere, including the client that submits the job, and upgrade by creating a new cluster rather than editing a running one.

Ray Data for batch pipelines and inference

Ray Data is the library for loading, transforming and writing large datasets, and for running batch inference. The problem it solves is that a naive pipeline reads everything, transforms everything, then writes everything, which needs memory for the whole dataset at every stage. Ray Data instead uses streaming execution: data is split into blocks, and blocks flow through the stages while earlier stages are throttled by backpressure when later ones fall behind. Memory stays bounded and GPUs are kept fed while CPU stages work on the next blocks.

Datasets are lazy. Reading and mapping build a plan, and nothing runs until something consumes the dataset by writing it, iterating over it, or calling an action such as count(). That laziness lets Ray fuse compatible stages and decide parallelism at run time.

import ray

ds = ray.data.read_parquet("s3://my-bucket/events/")   # lazy

def clean(batch):
    batch["text"] = [t.strip().lower() for t in batch["text"]]
    return batch

ds = ds.map_batches(clean, batch_size=1024)
ds.write_parquet("s3://my-bucket/events-clean/")       # execution happens here

Batch inference with a callable class

For a model you do not want to reload for every batch, pass a class to map_batches. Ray starts a pool of actors, each constructing the model once in __init__, and streams batches through them. This is the standard pattern for GPU inference, and it is the reason map_batches accepts concurrency and num_gpus.

class Embedder:
    def __init__(self):
        from sentence_transformers import SentenceTransformer
        self.model = SentenceTransformer("all-MiniLM-L6-v2", device="cuda")

    def __call__(self, batch):
        batch["vector"] = list(self.model.encode(batch["text"]))
        return batch

vectors = ds.map_batches(
    Embedder,
    batch_size=256,
    concurrency=4,        # four actors
    num_gpus=1,           # one GPU each
)
vectors.write_parquet("s3://my-bucket/vectors/")

The batch size is a tuning parameter with a real trade-off: bigger batches use the GPU better but need more memory and raise the cost of a retry. A sensible habit is to start with the largest batch that fits comfortably in GPU memory, then measure. Keep CPU-only preprocessing in a separate stage with no GPU request, so that the scheduler can run it on cheaper CPU nodes and the GPU actors are not waiting on string handling.

Choosing a file format and a sink

Parquet is the common default because it is columnar and splits cleanly into blocks. Release 2.58 adds write_delta for writing Delta Lake tables, which makes Ray a reasonable producer for lakehouse tables that Delta Lake and Spark consumers read. As with any new writer, test it against a copy of your table first.

Ray Train and Ray Tune at work

Training that does not fit one GPU, or must finish faster, needs several workers coordinating, and the coordination is what makes it hard: starting processes with the right ranks, wiring up the communication backend, handling a worker that dies, and saving checkpoints that a restart can use. Ray Train takes that job. You write a training function for one worker, and a trainer launches it on N workers with the distributed setup done.

import ray.train
from ray.train import ScalingConfig, RunConfig
from ray.train.torch import TorchTrainer

def train_loop_per_worker(config):
    model = ray.train.torch.prepare_model(build_model())
    loader = ray.train.torch.prepare_data_loader(build_loader())
    for epoch in range(config["epochs"]):
        loss = run_epoch(model, loader)
        ray.train.report({"loss": loss, "epoch": epoch})

trainer = TorchTrainer(
    train_loop_per_worker,
    train_loop_config={"epochs": 3},
    scaling_config=ScalingConfig(num_workers=4, use_gpu=True),
    run_config=RunConfig(storage_path="s3://my-bucket/runs"),
)
result = trainer.fit()

Treat the import paths above as the shape rather than a promise. Ray Train has gone through a major redesign, and older tutorials use APIs that have since been retired, including the ray.air namespace. When a tutorial's imports fail, check the Train documentation for your version before debugging anything else.

Shared storage is not optional

The storage_path is where checkpoints and results go, and it must be reachable from every node and survive the cluster. Nodes are ephemeral, so a checkpoint written to a local disk disappears with the node that wrote it, and a restarted run has nothing to resume from. Use object storage or a shared filesystem. Inside the loop, ray.train.report(metrics, checkpoint=...) records metrics and saves a checkpoint together, and ray.train.get_checkpoint() at the start of the loop returns the latest one when the run is being resumed after a failure.

That combination of checkpoints on shared storage and automatic retries is what makes cheap preemptible instances viable for training: a worker loss costs you the work since the last checkpoint rather than the whole run. Release 2.58 notes added a preemption-related API in Train, so check the release notes if spot capacity is part of your cost plan. Libraries such as DeepSpeed and PyTorch Lightning plug into Train, so you keep the framework you already use.

Tune for search

Ray Tune runs many trials of a trainable with different hyperparameters in parallel and uses a scheduler to stop poor ones early. The Tuner is the entry point, and the older tune.run function is the legacy style you will meet in tutorials.

from ray import tune

def trainable(config):
    for step in range(20):
        score = evaluate(config["lr"], step)
        tune.report({"score": score})

tuner = tune.Tuner(
    trainable,
    param_space={"lr": tune.loguniform(1e-5, 1e-1)},
    tune_config=tune.TuneConfig(num_samples=16, metric="score", mode="max"),
)
results = tuner.fit()
print(results.get_best_result().config)

The reporting call has moved between releases, so confirm the form in the Tune documentation for the version you run. Resource requests are the cost lever: a trial that reserves a GPU for a CPU-bound search wastes the cluster, and a search with 16 samples on four GPUs runs four at a time. Log results to MLflow or Weights & Biases with the integration callbacks so that the experiment history outlives the cluster.

Ray Serve in production

Serving a model is a different problem from training one. Requests arrive unpredictably, latency matters, and capacity must follow load. Ray Serve is Ray's library for that, and it has its own guide in this catalogue, so this section covers only what a Ray user needs to run it well. Its vocabulary is: an application is a deployable graph, a deployment is a scalable class or function, a replica is an actor serving requests, and a proxy accepts HTTP or gRPC traffic on each node and routes it to replicas.

from ray import serve

@serve.deployment(
    num_replicas=2,
    max_ongoing_requests=16,
    ray_actor_options={"num_cpus": 1},
)
class Model:
    def __init__(self):
        self.model = load_model()          # once per replica

    async def __call__(self, request):
        payload = await request.json()
        return {"prediction": self.model(payload["x"])}

app = Model.bind()

During development serve run module:app runs it in the foreground. For production, describe the application in a YAML config and apply it with serve deploy, so that the configuration is a reviewed file in version control and not a Python script that happens to call serve.run.

applications:
  - name: classifier
    route_prefix: /
    import_path: app:app
    deployments:
      - name: Model
        max_ongoing_requests: 16
        autoscaling_config:
          min_replicas: 2
          max_replicas: 10
          target_ongoing_requests: 8
        ray_actor_options:
          num_cpus: 1

serve deploy serve-config.yaml
serve status

Autoscaling and its two layers

Replicas scale on a target for ongoing requests per replica: when average load per replica exceeds target_ongoing_requests, Serve adds replicas up to max_replicas. Older tutorials use target_num_ongoing_requests_per_replica and max_concurrent_queries, which were renamed to target_ongoing_requests and max_ongoing_requests, so if a copied config is rejected, check the names first. Note that Serve scaling is only half of the story. New replicas need resources, and if the cluster has none, they wait for the cluster autoscaler to add nodes, which can take minutes for GPU nodes. Set min_replicas high enough that you do not depend on a cold start for ordinary traffic.

Batching and backpressure

For models that run faster on batches, @serve.batch(max_batch_size=..., batch_wait_timeout_s=...) collects concurrent requests and calls your function once with a list. The wait timeout is the latency you are willing to spend to fill a batch, so tune it against your latency budget. Under overload, max_ongoing_requests and the queue limit protect replicas, and rejected requests return a 503; release 2.58 documents a configurable status code and a Retry-After header for that case, which is worth reading if clients retry aggressively.

For large language models, Serve integrates with vLLM through Serve LLM, and release 2.58 added KV-cache and token-aware request routing there. If your goal is an HTTP API around a plain model, compare it honestly with FastAPI on a few replicas: Serve earns its complexity when you need model composition, autoscaling or fractional GPUs.

Running Ray on Kubernetes with KubeRay

Production Ray usually means Kubernetes, because it already handles scheduling, networking and restarts, and Kubernetes is where most teams run everything else. KubeRay is the operator that teaches Kubernetes about Ray through three custom resources.

RayJob is the right default for batch and training, because the cluster exists only for the job, and its shutdownAfterJobFinishes setting removes the cluster when the job ends. That turns idle GPUs into a non-problem. RayService is the right default for online serving, where a new version is brought up alongside the old one and traffic switches when it is healthy. A shared RayCluster suits interactive exploration, and its cost is that jobs share a failure domain and have no strong isolation.

Images, autoscaling and the head

Build one image that contains your pinned Ray version and dependencies, and use that same image for head and workers, which satisfies the version rule automatically. Pin the tag, never latest, and build it as you would any Docker image, with the lockfile resolved at build time. Install the operator and clusters with Helm charts, and keep manifests in Git so that Argo CD reconciles them; the KubeRay project documents the charts.

For autoscaling, the Ray autoscaler running inside the cluster adds and removes worker pods by watching unmet resource demand, enabled in the cluster spec with enableInTreeAutoscaling, with minimum and maximum replicas per worker group and an idle timeout. It then relies on the Kubernetes node autoscaler or a provisioning tool to supply machines, so your cluster-level capacity must allow it. Use separate worker groups for CPU and GPU work so each scales independently.

Treat the head pod as the most important and least replaceable one. Give it modest, predictable resources, schedule no tasks on it, and use shared storage for everything durable. Without GCS fault tolerance backed by external Redis, losing the head loses the running cluster state. The fault tolerance documentation describes the setup and the KubeRay support for Redis-backed GCS, and it is worth configuring for RayService workloads. Where your employer has data-residency obligations, run the cluster and its storage in a region inside the required jurisdiction, and keep the dataset buckets in the same region to avoid cross-region transfer costs.

Security: Ray trusts everyone who can reach it

This is the part of Ray that surprises people most. The security documentation states that Ray has no authentication by default and allows any client to run arbitrary code. Anyone who can reach the dashboard on port 8265, the Jobs API, Ray Client, or the cluster's internal ports can execute code on your machines, with whatever cloud credentials those machines hold. Search results regularly show Ray dashboards exposed to the internet, and that is how clusters get hijacked.

So security is a property of the network and the platform around Ray, not of Ray. In practice:

1. Never expose the dashboard or Jobs API to the public internet. Keep clusters in private subnets, restrict ingress with security groups or Kubernetes network policies, and put an authenticating reverse proxy or VPN in front of anything humans use.
2. Give nodes the minimum cloud permissions, through workload identity rather than long-lived keys, because any code that runs on the cluster inherits them.
3. Use token authentication, available since release 2.52 through the RAY_AUTH_MODE and RAY_AUTH_TOKEN settings, as an additional layer, and read its documentation for the exact setup. Do not treat it as a replacement for network controls.
4. Keep secrets out of runtime_env in committed code and out of job submission command lines, since job logs and the dashboard can reveal them. Mount Kubernetes Secrets or read from a secret manager such as Vault when the worker starts.
5. Do not deserialise data from untrusted sources. Ray uses cloudpickle, and unpickling untrusted bytes is code execution.
6. Do not put mutually distrusting teams in one cluster. There is no strong tenant isolation: tasks share machines and the object store. Run one cluster per trust boundary, which on KubeRay means separate clusters in separate namespaces with quotas.

Observability and debugging beyond the basics

When a Ray job is slow or stuck, the question is always where the time or the resources went. Ray gives you layers of tools, and a good habit is to go from the cheapest to the most detailed.

The State API answers "what is the cluster doing right now" from the command line. ray list nodes, ray list actors, ray list tasks and ray summary tasks show states such as pending, running and failed. A job whose tasks are pending with nothing running is a scheduling problem: compare the resource request with ray status. A job whose tasks are running but slow is a code or data problem.

Logs live per node, under /tmp/ray/session_latest/logs, and ray logs retrieves them from the CLI. Worker output is also streamed to the driver, with a prefix naming the task or actor. Ray deduplicates repeated log lines by default, which can hide the fact that a message came from many workers; the RAY_DEDUP_LOGS setting controls it, as the logging documentation describes.

Metrics are exported for Prometheus, and Ray can generate Grafana dashboards for the cluster, so the same observability stack you use for services covers Ray. Pair Prometheus and Grafana with alerts on node memory, object store use and pending resource demand, which catch most Ray incidents before users do. If you trace requests across services, check the Ray observability documentation for its OpenTelemetry support in your version.

The timeline from ray.timeline() produces a trace you load into a Chrome-compatible viewer to see which tasks ran where and when. It is the way to find stragglers, idle gaps, and a serial section that should have been parallel.

Reading common failures

Testing and performance

Distributed code is easier to test than it looks, because Ray runs the same code locally. The strategy is to keep logic in plain functions that you test without Ray at all, and keep the thin @ray.remote layer separate. A function that takes data and returns data needs a normal unit test. The remote wrapper needs a few integration tests that confirm resources, retries and serialisation behave.

import pytest
import ray

@pytest.fixture(scope="module")
def ray_start():
    ray.init(num_cpus=2, include_dashboard=False)
    yield
    ray.shutdown()

def test_square_runs_remotely(ray_start):
    @ray.remote
    def square(x):
        return x * x
    assert ray.get([square.remote(i) for i in range(4)]) == [0, 1, 4, 9]

def test_task_survives_one_failure(ray_start):
    @ray.remote
    class Counter:
        def __init__(self): self.n = 0
        def bump(self):
            self.n += 1
            return self.n

    c = Counter.remote()
    @ray.remote(max_retries=2, retry_exceptions=True)
    def flaky(counter):
        if ray.get(counter.bump.remote()) < 2:
            raise RuntimeError("first attempt fails")
        return "ok"

    assert ray.get(flaky.remote(c)) == "ok"

Starting Ray costs seconds, so use a module or session fixture and not one per test. Local mode, where tasks run in the driver process, was a debugging convenience in older versions, but its status has changed over time, so check the documentation of your version before relying on it. A cheaper debugging technique is to call the underlying Python function directly, or to run with a single CPU so that ordering is simple. Running the same suite in CI with a pinned Ray version also catches the upgrade breakage that comes from Ray's frequent releases.

Performance: a short checklist in order of leverage

1. Task granularity. Each task has scheduling overhead. A task that runs for a millisecond spends most of its time on overhead, so batch small units into one task. A task that runs for an hour recovers poorly from failure, so keep tasks to seconds or minutes.
2. Parallelism. Remove sequential ray.get calls and use ray.wait to overlap work.
3. Data movement. Pass large inputs by ObjectRef, keep producers and consumers on the same node when you can, and avoid returning large values to the driver that only another task needs.
4. Resource honesty. Declare real CPU, GPU and memory needs so the scheduler packs nodes properly.
5. Streaming over materialising. Use Ray Data for pipelines, so memory stays bounded.
6. Right-size the cluster. Measure utilisation in Grafana before adding nodes. An idle GPU is a cost, and a saturated one is a signal.

Configuration, upgrades and cost

Production Ray needs the same discipline as any other platform component. Keep cluster specs, Serve configs and job definitions in version control, and apply them through CI or a GitOps tool so that every change is reviewed, in the same way you would manage infrastructure with Terraform and charts with Helm. Pin the Ray version in one place, such as the image tag, and have everything else derive from it.

Upgrades deserve a routine, because Ray ships a minor release every few weeks and the notes regularly include deprecations and removals. Release 2.57 removed the old Ray Data cluster autoscaler, release 2.56 dropped Pydantic v1 support, and 2.58 deprecates several scheduling and Data APIs. Read the notes for each version between yours and your target, run your test suite on the new version, then create a new cluster and move work across. For serving, RayService automates the cutover with no downtime. For batch, a new RayJob simply starts on the new image.

Cost follows from a few decisions. Autoscale workers down aggressively with the idle timeout. Use RayJob so clusters do not outlive jobs. Put preprocessing on CPU nodes and keep GPU nodes for the model stage. Use preemptible instances for work that checkpoints and retries safely, and keep the head on stable capacity. For serving, use fractional GPUs only when models fit together in memory, and make sure min_replicas reflects real traffic and not a guess. Cross-region data transfer is a quiet cost, so keep compute near storage.

Putting it all together

Here is the shape of a small production pipeline that uses most of this level. The task: nightly, embed new documents with a GPU model, write vectors to object storage, and keep an online service that answers queries.

First, the image. One Dockerfile installs a pinned Ray, the model dependencies and your code, and CI builds and tags it with the Git commit. Head and workers use that tag, which makes version drift impossible by construction.

Second, the nightly job. A RayJob runs a script on a cluster with a small CPU worker group and a GPU group that scales from zero. The script reads Parquet from object storage with Ray Data, runs a CPU cleaning stage, then a callable-class embedding stage with concurrency and num_gpus set, and writes Parquet back. Heavy stages declare memory, and the I/O stages are idempotent because they write to a run-specific path and a final commit step publishes it. With shutdownAfterJobFinishes the GPUs disappear when the job ends. The job's metrics go to Prometheus, and an alert fires if it has not finished by morning.

Third, the online service. A RayService runs a Serve application with a deployment for the query model, min_replicas set to cover the normal load, an autoscaling target set from load tests, and batching enabled with a wait timeout inside the latency budget. Its YAML config is in Git and applied by Argo CD, and upgrades go through the RayService zero-downtime path.

Fourth, the guard rails. The clusters live in private subnets in the region your data-residency rules require, the dashboard is reachable only through the company proxy, the workers use workload identity with read access to one bucket and write access to another, and no secrets appear in job definitions. Each team has its own cluster. Tests run the core functions without Ray and a small set of integration tests with a two-CPU local cluster, against the pinned version.

Fifth, the operations loop. When a night run fails, you start with ray summary tasks and the job logs, then ray memory if memory is implicated, and last the timeline if it was slow. Each failure becomes a setting in code, such as a memory declaration, a retry or a smaller batch, and not a note in someone's head.

None of this is exotic. It is the same discipline you apply to any service, applied to a runtime that makes it very easy to write code that works in a notebook and falls over on a cluster.

Where you are now

You can read what a Ray cluster is doing and predict why, rather than hoping. You know that .remote() is asynchronous and that ownership decides which objects survive, that resources are logical declarations you must make honestly, and that objects live in a store that fills, spills and pins memory through references. You can make tasks and actors survive failure with retries, restarts and checkpoints, and you know which of those need idempotent code. You can submit jobs, build environments reproducibly, and keep versions aligned. You can build streaming data and batch inference pipelines, run training and tuning with durable storage, and operate Serve with autoscaling and batching. You can run the whole thing on Kubernetes with KubeRay, and you know that security comes from the network and the platform around Ray because the cluster trusts every client it can reach.

The senior level takes the next step: owning Ray as a platform for several teams. That means GCS fault tolerance and head failure, the limits of the autoscaler, tenant isolation, upgrade programmes, cost governance and the question of when Ray is the wrong tool. Before you go on, run one real workload of your own through the timeline and the State API, and fix the first thing they show you.

Sources

• Ray documentation home
• Installing Ray
• Ray Core walkthrough
• Scheduling in Ray Core
• Ray security
• GCS fault tolerance
• Ray Serve
• Ray observability
• Ray on Kubernetes
• Ray releases on GitHub
• KubeRay on GitHub

Part two of three. A cumulative review of Beginner and Mid-level Ray, organised by scenario rather than by level, in about thirty-five minutes. Interviewers at this level stop asking what a task is and start asking what you would check when something is slow, lost or expensive. Answers below are written the way a strong candidate would say them, followed by what a weak answer misses. Senior reviews all three levels.

Where this picks up

The thirty-second answer, at mid level

Ray is a distributed runtime for Python. You mark functions and classes as remote, Ray schedules them across a cluster using declared resources, passes data by reference through a shared-memory object store, and gives you libraries on top for data processing, training, tuning and serving. It does not make slow code fast by itself. It makes parallel code easy to write, and it leaves resource declarations, fault tolerance, storage and security to you.

If you can say that, then add one sentence of production judgement, you have passed the opening. For instance: "In production I run it on Kubernetes through KubeRay, keep durable state on shared storage, and treat the cluster as fully trusting of anyone who can reach it."

Scenario: a job is slower than expected

Question: Your Ray job takes an hour on a cluster where you expected ten minutes. What do you check first?

A strong answer works in a fixed order and says why. First, is the cluster busy or idle? ray status and the dashboard show resource usage. If most CPUs or GPUs are idle, the problem is scheduling or code structure. If they are saturated, the cluster is simply too small or the work is too heavy.

Second, are tasks pending or running? ray summary tasks and ray list tasks separate the two. Pending tasks with idle nodes usually mean a resource request that does not fit, for example a task asking for more memory or a custom resource no node offers, and ray status shows the unmet demand.

Third, is the work actually parallel? The classic cause is a ray.get inside the submission loop, which waits for each task before submitting the next. A timeline from ray.timeline() shows it immediately as one task running at a time with gaps between.

Fourth, is task size sensible? Tasks that run for milliseconds are dominated by scheduling overhead. Batching many items per task is the usual cure.

Fifth, is data movement the cost? Large arguments captured by closure are re-serialised, and results pulled back to the driver are copied. Passing ObjectRefs and keeping data in the cluster removes that.

A weak answer says "add more nodes". More nodes help only if the cluster is saturated, and in a sequential-in-disguise program they change nothing. The point the interviewer is listening for is that you diagnose before you scale.

Scenario: out-of-memory and a full object store

Question: Workers are dying with OutOfMemoryError and sometimes you see ObjectStoreFullError. Are those the same problem?

They are related but different, and a strong candidate separates them. OutOfMemoryError, with the message that a task was killed because the node was running low on memory, comes from Ray's memory monitor. When node memory use passes a threshold, by default 95 percent, Ray kills a worker to protect the node. The cause is too much heap memory in use across the tasks running at once on that node.

ObjectStoreFullError is about the shared-memory store, which has a fixed size. Ray spills objects to disk to make room, and the error appears when spilling cannot keep up. The cause is too many live objects, usually because references are held longer than needed.

Fixes differ accordingly. For OOM: declare memory= on heavy tasks so the scheduler packs fewer on a node, reduce concurrency, shrink batch sizes, and use larger nodes if the working set is genuinely big. For a full store: hold fewer ObjectRefs, release references promptly, avoid returning large intermediate values, and use Ray Data so blocks stream with backpressure. ray memory shows which references pin which objects.

A weak answer reaches for the memory threshold setting. The threshold is almost never the fix, because a higher threshold just moves the kill closer to the machine's real limit. Another weak answer is "add swap". What the interviewer wants is the habit of declaring real memory needs.

Scenario: tasks pending with nothing running

Question: You submit tasks and the job sits there. ray status shows pending demand. Why?

The scheduler cannot find a node that satisfies a task's declared resources. Possibilities, in the order I would check them: the request is infeasible, meaning larger than any node, such as num_gpus=8 on four-GPU nodes; the cluster is fully claimed by actors that are holding resources while idle; a placement group cannot be satisfied, for example STRICT_SPREAD over more nodes than exist; or the cluster is autoscaling and new nodes are still starting.

For the actor case, the warning says that the request cannot be scheduled because all cluster resources are claimed by actors. The remedy is to reduce what actors reserve, use num_cpus=0 for actors that mostly wait, or add capacity. For the autoscaling case, patience and a look at the autoscaler state in ray status are the answer, along with the knowledge that GPU nodes can take minutes to arrive.

What distinguishes a strong answer is noticing that demand drives the autoscaler: pending resource requests are what trigger new nodes, so a request no node type can ever satisfy never resolves, and the job will wait indefinitely.

Scenario: making a job survive failures

Question: A nightly job runs for six hours on spot instances. How do you make it survive node loss?

Cover the three object kinds. For tasks, set max_retries so a task whose worker died is rerun, make the task idempotent so a rerun is safe, and write outputs to unique paths with a final atomic commit. Use retry_exceptions=True only where application errors are safe to retry. For actors, set max_restarts and make __init__ rebuild whatever state is needed, because a restart discards memory. For objects, understand that lineage reconstruction can recompute a lost task result, but objects made with ray.put cannot be rebuilt if their owner dies.

For training or any long computation, add checkpoints on shared storage. Nodes disappear with their local disks, so a checkpoint on local disk is lost with the node. With a durable storage_path, a restarted run resumes from the latest checkpoint and loses only the work since then. Put the head on stable, non-preemptible capacity, because losing it ends the cluster unless GCS fault tolerance with external Redis is configured.

A weak answer says "Ray handles failures automatically". It handles some, with settings you must choose, and it cannot make a non-idempotent task safe.

Scenario: dependencies and environments

Question: Your job works locally, but on the cluster it fails with an import error. What are the options?

The workers lack the dependency. There are two approaches. A runtime environment declares pip, conda, env_vars or a working_dir, and Ray builds it on the nodes. It is quick for development. The alternative is to bake dependencies into the image, which is what I do in production because installation at job start is slow, depends on a package index being reachable, and makes runs less reproducible. The runtime environment still has a place for small differences, such as an environment variable or your own code directory.

Add the version rule. Head, workers and the submitting client must run the same Ray and Python versions. A mismatch fails at connect time with a version mismatch error that names both sides. The fix is to pin one image tag everywhere, not to try to tolerate drift.

A weak answer says "pip install it on every node by hand", which does not survive autoscaling, since new nodes arrive without your change.

Scenario: choosing Ray over its alternatives

Question: A colleague proposes Ray for a nightly ETL job. Is that right?

Ask what the work is. Ray fits Python workloads that need fine-grained parallelism, stateful workers or GPUs: batch inference, embedding generation, simulation, distributed training, hyperparameter search. For large tabular transformations in SQL-shaped logic, Spark is often a better fit: it has mature SQL, a large connector ecosystem and well-understood operations in most data teams. Ray Data can read and write the same Parquet and Delta data, so the two coexist.

A strong answer weighs the team: if nobody operates Ray, the cost of operating a new platform is part of the bill. It also mentions where Ray is not the best choice, namely when a single large machine suffices, or the problem is a simple scheduled workflow, where an orchestrator is simpler.

Trade-off questions

Runtime environment versus baked image. The runtime environment is fast to iterate and needs no image build, but it installs at job start and depends on external indexes. The baked image is reproducible and fast to start, at the price of a build pipeline. Strong answers choose by stage: runtime environments in development, images in production.

RayJob versus a long-lived RayCluster. RayJob creates a cluster for a job and can delete it afterwards with shutdownAfterJobFinishes, which removes idle cost and gives each job its own failure domain. A shared cluster starts jobs instantly and suits interactive work, but jobs interfere, and everyone shares the same lack of isolation. Choose RayJob for batch and training, a cluster for exploration.

Ray Serve versus a plain web framework. A framework such as FastAPI is simpler to operate for a single model with modest traffic. Ray Serve earns its place with model composition, autoscaling, request batching, fractional GPUs or LLM serving through vLLM. The answer to give is "start simple, move when a requirement demands it".

Many small tasks versus fewer large ones. Small tasks parallelise well but pay scheduling overhead per task. Large tasks have low overhead but recover slowly from failure and can unbalance the load. Aim for tasks of seconds to minutes and batch smaller items.

Fractional GPUs. Sharing a GPU improves utilisation for small models. Ray schedules by the number you declare and does not isolate GPU memory, so two tasks can still exhaust the device. Use it when you know the memory footprint.

One cluster for all teams versus a cluster per team. One cluster is cheaper and easier to run, but Ray provides no strong tenant isolation and any client can run arbitrary code. Separate clusters, using namespaces and quotas on Kubernetes, are the safe default when teams do not fully trust each other.

Security questions

Question: What is Ray's security model?

There is none by default, and a strong candidate says so first. Ray allows any client that can reach it to run arbitrary code. That covers the dashboard, the Jobs API, Ray Client and the internal cluster ports. So the controls live outside Ray: private networks, security groups or network policies, no public dashboard, and an authenticating proxy for human access. Token authentication, introduced in release 2.52, adds a layer and should not be the only one. Workers should use workload identity with least privilege, since code that runs on them inherits their permissions. Secrets should come from a secret manager or Kubernetes Secrets such as Vault rather than being placed in runtime_env in committed code.

A weak answer is "we put the dashboard behind a password". Another weak answer misses the pickle risk: deserialising data from untrusted sources can run code.

Operations questions

Question: How do you upgrade Ray without an outage?

Ray releases every few weeks and does not promise compatibility across minor versions within a cluster. So you upgrade by replacing the cluster. Read the release notes for each version between yours and the target, since they list removals and deprecations, run your tests against the new version, build a new image, and cut over. For serving, RayService on KubeRay supports a zero-downtime upgrade. For batch, new RayJobs start on the new image. In place rolling upgrades of nodes fail the version check.

Question: What do you monitor?

Node memory, object store use and spilling, pending resource demand, replica counts and request latency for Serve, task failure counts, and job completion time against the expected window. Ray exports metrics for Prometheus and generates Grafana dashboards, so the data goes into the same stack used for services, and alerts follow the same on-call process. The State API and ray.timeline() are for investigation, not alerting.

Question: How do you control cost?

Scale workers down with the idle timeout, use RayJob so clusters end with the work, keep CPU stages off GPU nodes, use preemptible capacity for checkpointed work, set min_replicas from real traffic, and keep compute near its data to avoid transfer fees.

Code-reading questions

Question: What is wrong with this code?

results = []
for path in paths:
    results.append(ray.get(process.remote(path)))

The ray.get inside the loop waits for each task before submitting the next, so the work runs sequentially. The fix is to submit all tasks first, then call ray.get once on the list, or use ray.wait to handle results as they finish.

Question: And this?

client = make_db_client()

@ray.remote
def lookup(key):
    return client.get(key)

The function captures a client object that holds a socket or a lock. Ray serialises captured objects into the function, and such objects often fail with a pickling error, for example about a _thread.lock object, or work incorrectly when copied. Create the client inside the task, or inside an actor's __init__ so it is built once per worker and reused.

Question: And this?

@ray.remote(max_retries=5, retry_exceptions=True)
def charge_customer(order_id):
    payments.charge(order_id)

Retrying a function with a side effect can charge twice. The function must be idempotent, for example by sending an idempotency key to the payment provider, or retries must be disabled for it.

Common interview questions

What is the difference between ray.get, ray.wait and ray.put? get blocks until a result is ready and returns it. wait returns lists of ready and not-ready refs with an optional timeout, which lets you process results as they complete. put stores a value in the object store and returns a reference, so you can share a large object without re-serialising it per task.

Why do actor calls not run in parallel? By default an actor is one process that runs method calls serially in order, which makes its state safe without locks. For concurrency use async methods or max_concurrency, or create several actors.

What does the head node do? It runs the GCS, which stores cluster metadata and the actor directory, plus the dashboard and Jobs endpoint. It is a single point of failure unless GCS fault tolerance with external Redis is set up, so keep it stable and lightly loaded.

What is a placement group for? Reserving several resource bundles together under a strategy such as STRICT_SPREAD, so that distributed jobs get all their workers or none.

How does Ray Data avoid running out of memory? Streaming execution and backpressure: blocks pass through the stages incrementally, so only a bounded amount is in flight, instead of the whole dataset being materialised.

How would you serve a model that needs a GPU, with variable traffic? A Serve deployment with ray_actor_options requesting a GPU, an autoscaling_config with a target for ongoing requests per replica, batching if the model benefits, and a minimum replica count that covers normal load, because GPU nodes start slowly. Run it as a RayService for zero-downtime upgrades.

What a strong answer covers that a weak one misses

• Names the mechanism. "Tasks are pending because the demand is infeasible" beats "Ray is slow".
• Diagnoses before scaling. Strong answers ask what the cluster is doing before adding hardware.
• Mentions idempotency whenever retries come up.
• Separates memory from the object store, and OOM kills from store-full errors.
• Starts the security answer with "there is no authentication by default", then the network controls.
• Knows the version rule and replaces clusters to upgrade.
• Has an opinion about when not to use Ray, with a named alternative.

Sixty-second self-test

Answer out loud without looking.

1. Why does ray.get inside a loop make a program sequential?
2. Which two errors come from memory, and what causes each?
3. What does it mean that resources are logical?
4. Name the three failure-handling knobs for tasks, actors and objects.
5. Why must checkpoints go to shared storage?
6. When do you choose RayJob, and when RayService?
7. Why does Ray need the same version on every node?
8. What does Ray's security model assume about network access?
9. Which of your answers would change if the cluster were shared by two teams?
10. When would you tell a colleague not to use Ray?

Part two of three. At this level your Ray jobs run, so the problems change. They are no longer "why won't this start" but "why is this slower than a single machine", "why did the cluster fill up", and "why does this work in a notebook and not in the nightly job". Start with the error table, then the practices in order of leverage, then the practice cards and operational recipes at the end.

Common errors at this level

These are cumulative. Beginner's errors still apply, and the following are the ones that appear once things work.

Practices that pay off most, in order of leverage

1. Diagnose before you scale

The most expensive habit is adding nodes. Before that, ask three questions in order: is the cluster idle or saturated (ray status), are tasks pending or running (ray summary tasks), and is the program actually parallel (ray.timeline()). Most slow Ray jobs are a code structure problem or a resource declaration problem. Adding machines to a sequential program increases your bill and leaves the run time unchanged.

2. Declare resources honestly

Resources are logical, and Ray does not enforce them. A task declared with one CPU can use eight, and the scheduler will still place seven more tasks beside it. Give real values for num_cpus and num_gpus, and add memory= on anything heavy. The memory declaration is one of the cheapest reliability improvements available, because it stops the scheduler packing too many large tasks on one node and it informs which tasks the memory monitor kills first.

3. Pass references, not data

Use ray.put once, and pass the resulting ObjectRef to every task that needs the data. Capturing a large object in a closure serialises it into the function. When tasks chain, pass the output ref of one directly into the next, so the data stays in the object store and never travels through the driver.

4. Make everything that retries idempotent

Retries are the main reason Ray jobs survive node loss, and they are also how side effects happen twice. Write outputs to a path that includes the run and task identity, and publish with an atomic rename or a final commit step. For calls to outside systems, use idempotency keys. Only enable retry_exceptions=True where repeating the work is harmless.

5. Checkpoint to shared storage

Anything long-running needs a durable checkpoint on storage that every node can reach and that outlives the cluster. Local disk dies with the node. With a shared storage_path for Train, or your own periodic writes for custom jobs, a restart costs you minutes and not hours, and that is what makes preemptible capacity worth using.

6. Pin one version and replace clusters to upgrade

Head, workers and clients run the same Ray and Python version, set from one place, normally the image tag. Do not upgrade nodes in place. Read the release notes for each minor version between yours and the target, since Ray ships often and the notes list removals. Release 2.57 removed an old Ray Data autoscaler and 2.56 dropped Pydantic v1, so an upgrade skipped for several versions can be a surprise.

7. Treat the network as the security boundary

Ray has no authentication by default, and any client that reaches the dashboard, the Jobs API or the cluster ports can run arbitrary code. Keep clusters on private networks, restrict ingress, front human access with an authenticating proxy, and use workload identity with narrow permissions. Token authentication, available since release 2.52, is an extra layer and not a substitute.

8. Keep the head light

Start the head with --num-cpus=0 so tasks do not run beside the GCS and the dashboard, give it stable capacity, and keep nothing important on its local disk. Without GCS fault tolerance, losing the head loses cluster state, so make the head boring.

9. Bake dependencies into the image

A runtime_env that installs packages at job start is handy in development and fragile in production, because a package index outage becomes a job outage and runs stop being reproducible. Use it for small differences such as environment variables, and keep real dependencies in the pinned image.

10. Test the logic without Ray

Put the real work in plain functions that you unit test without a cluster, and wrap them thinly with @ray.remote. Add a handful of integration tests with a small local cluster started in a module-scoped fixture for the behaviours that depend on Ray: resources, retries, serialisation.

Habits that prevent the next incident

Some habits do not fit a single error row, but they decide whether a team spends its week on features or on firefighting.

Write down what each job needs. For every recurring job, keep a short note beside the code with its peak memory, GPU count, expected duration and the storage it reads and writes. When a run goes wrong, you compare the run against that note instead of guessing, and when you size a cluster or a RayJob the numbers are already there. Update the note whenever the data volume changes noticeably, because a job that fit comfortably last quarter often does not fit now.

Alert on the cause, not only the symptom. A job that has not finished is a symptom. Pending resource demand that stays above zero for ten minutes, an object store that spills continuously, and a node whose memory sits near the monitor threshold are causes, and they fire earlier. Put the first two on a dashboard that the on-call person actually opens.

Rehearse the failure you fear. Kill a worker node during a staging run and watch whether retries and checkpoints do what you believe they do. Stop the head and see how long recovery takes and what is lost. A failure rehearsed once in staging costs an afternoon, and the same failure met for the first time in production costs a night.

Keep one example that always works. Maintain a tiny job, a few lines long, that submits a handful of tasks and one actor call, and run it after every cluster change or upgrade. When it fails, you know the platform is broken and not your pipeline, and that distinction saves hours.

Review release notes on a schedule. Ray moves quickly, so put a recurring slot in the team calendar to read the latest notes and decide whether to upgrade. The notes for recent releases included security fixes in Ray Data and Serve, which is reason enough to avoid falling many versions behind.

Share the lessons. When a diagnosis took longer than an hour, add a row to your team's own version of the error table above, with the symptom, the cause and the command that found it. The tables in this guide are a starting point, and the ones that help most are the ones that name your own cluster, your own image and your own storage.

Practice cards

Each card is a ten-minute exercise on a real project. Do them in order.

Card 1: Find the serial section. Run a job of yours, capture ray.timeline() to a file, and load it in a trace viewer. Look for stretches where only one task runs. Fix the cause and capture again. What you should end up with is a before and after timeline.

Card 2: Break it with memory. Write a task that allocates a large array and run it with more concurrency than the node's memory can hold, to provoke OutOfMemoryError. Then add memory= and observe the scheduler run fewer at a time. This teaches you what the declaration does without waiting for production to teach it.

Card 3: Fill the store. Create many ray.put objects and keep their references in a list, then run ray memory. Delete the list, collect garbage, and run it again. Note which references were pinning memory.

Card 4: Kill an actor. Create an actor with max_restarts=2 and some state, kill it with ray.kill(actor, no_restart=False), and call it again. Observe that the state is gone and that it came back. Then decide where that state should have lived.

Card 5: Provoke an infeasible request. Submit a task asking for more CPUs than any node has. Read the warning, then read ray status for the unmet demand, and connect the two. Do the same with a STRICT_SPREAD placement group on a one-node cluster.

Card 6: Submit a job properly. Start a local head and run ray job submit --address http://127.0.0.1:8265 --working-dir . -- python script.py. Check status and logs from the CLI, stop a long job, and find the same job in the dashboard.

Card 7: Batch inference with a class. Write a callable class that loads a small model in __init__, run it through map_batches with concurrency=2, and confirm in ray list actors that two actors serve the stream.

Card 8: Deploy and scale a Serve app. Put an application in a YAML file, apply it with serve deploy, load it with a request generator, and watch serve status. Change the autoscaling target in the file and apply again.

Card 9: Version mismatch on purpose. Start a cluster with one Ray version in one virtual environment and connect with a different version from another. Read the error. This is the error you will see in production when an image tag drifts, so recognise it now.

Card 10: Expose check. From another machine on your network, try to open the head's dashboard port. Write down who could run code on your cluster if they could.

Operational recipes

Is the cluster healthy

ray status                     # nodes, resource use, pending demand, autoscaler state
ray list nodes                 # which nodes are alive
ray list actors                # actor states
ray summary tasks              # tasks grouped by function and state

Read ray status for three things: whether all expected nodes are listed, whether usage is near total, and whether there is unmet demand. Unmet demand with idle nodes is a shape problem, while unmet demand with full nodes is a capacity problem.

Why is memory full

ray memory                     # object references, sizes, who holds them
ray logs --help                # list the log retrieval options for your version

Look for large objects with a driver-held reference type. Those are the references you can release. Compare the object store usage with node memory use to decide whether you have an object store problem or a heap problem.

Submit, watch and stop a job

export RAY_ADDRESS=http://HEAD_IP:8265
ray job submit --working-dir . -- python train.py
ray job status <job_id>
ray job logs <job_id>
ray job stop <job_id>

With RAY_ADDRESS set you can drop --address from every command. Keep the job ID in your pipeline's output, so that anyone can look up logs later.

Run a cheap local cluster to reproduce a problem

ray start --head --num-cpus=2 --dashboard-host=127.0.0.1
python repro.py
ray stop

Limiting CPUs reproduces scheduling contention on a laptop. Binding the dashboard to the loopback address keeps it off the network, which is a good default on any shared machine.

A pytest fixture for Ray

import pytest
import ray

@pytest.fixture(scope="session")
def ray_cluster():
    ray.init(num_cpus=2, include_dashboard=False)
    yield
    ray.shutdown()

Start Ray once per session, since startup costs seconds, and put the fixture in conftest.py. Pin the Ray version in your lockfile so a test run in CI exercises the same release that runs in production.

Check a deploy of Serve

serve deploy serve-config.yaml
serve status

Wait for every application and deployment to report healthy before sending traffic. If replicas stay in a starting state, compare the resources in ray_actor_options with what ray status says is free. If a config is rejected, check for the old option names that were replaced by max_ongoing_requests and target_ongoing_requests.

Upgrade checklist

1. Read the release notes for every minor release between your version and the target, looking for removals and deprecations of APIs you use.
2. Update the pinned version in one place and rebuild the image.
3. Run the unit tests and the Ray integration tests against the new image.
4. Start a new cluster, or a new RayJob, on the new image and compare results and timings.
5. For Serve, apply the change through RayService so the cutover has no downtime.
6. Keep the old image available until the new one has run a full cycle.

Pre-flight for a production job

• Resources and memory are declared on every heavy task and actor.
• Tasks that retry are idempotent, and outputs go to unique paths with a final commit.
• Checkpoints and outputs go to shared storage.
• The image is pinned, and the client uses the same version.
• The dashboard and Jobs API are not reachable from outside the private network.
• Workers use workload identity with narrow permissions, and no secrets sit in runtime_env or job commands.
• Metrics reach Prometheus and Grafana, with alerts on memory, pending demand and job duration.
• The cluster is deleted when the job ends, using a RayJob on Kubernetes.

When to look further

If you run Ray Serve for production traffic, read the dedicated Ray Serve guide. If your cluster manifests live in Git and are applied by a GitOps tool, Argo CD and Helm cover that side. For tracking experiments across Tune and Train runs, see MLflow. The senior level turns to running Ray as a shared platform for several teams, where head failure, isolation and cost governance become the main concerns.

This is part three of three. It takes every topic from Beginner and Mid one level further and adds the work you own when Ray is not your tool but your team's platform: the cluster other people submit code to, the thing that pages you at night, and the line item finance asks about.

Until now the questions were about making a distributed program run. From here they are different. What happens when the head node dies? Which component saturates first when forty people share the cluster? What can a stranger do if they reach port 8265? How do you move from one Ray release to the next when the project ships every few weeks? And when is Ray simply the wrong answer?

The version this guide is verified against is Ray 2.58.0 (released 2026-08-23). Ray ships a minor release every two to six weeks, so treat every version-specific statement here as something to recheck against the release notes before you act on it.

Where this picks up

I start with the architecture, because every operational decision after this one is a statement about which component you expect to fail.

Architecture and failure modes

You know the nouns: a head node, worker nodes, a raylet on each, an object store on each, a GCS. A platform owner needs the verbs: what each component is responsible for, and what the cluster does when it disappears.

The GCS (Global Control Service) is the cluster's memory of itself. It knows which nodes exist, which actors are alive and where, which placement groups are reserved and which jobs are running. It lives on the head node, there is one of it, and by default its state is held in memory. That is the first fact a platform owner must absorb: by default the head node is a single point of failure for the control plane.

What breaks when the GCS goes away depends on what you were doing. Tasks already running on workers keep running, because the raylets schedule locally and talk to each other for data. But anything that needs the directory fails: creating a new actor, looking up a named actor, creating a placement group, a new node joining. Without fault tolerance, restarting the head means a new GCS with an empty memory, and the old workers cannot rejoin a cluster that has forgotten them. In practice a head failure without GCS fault tolerance means the cluster is rebuilt and every job resubmitted.

Ownership is the second idea. Every object has an owner: the worker process that created the ObjectRef by calling .remote() or ray.put. The owner tracks where the object lives and how many references exist. This is why the exception names you saw at Mid level matter operationally:

• OwnerDiedError means the process that created the reference has died, so nobody remains who knows where the value is. A long-lived actor that creates references for many callers is a risk: if it restarts, every outstanding reference it created is orphaned.
• ObjectLostError means the value itself is gone, for example because its node died. Ray can often rebuild it by re-running the task that produced it (lineage reconstruction), but only if the task is deterministic enough, retries are allowed, and the inputs still exist.
• ObjectStoreFullError means a node's shared memory is full and spilling to disk could not keep up.

The practical design rule: put long-lived state in actors or external storage, keep references short-lived, and never make a single driver or actor the owner of everything. The driver is also an owner, which is why a job whose driver dies loses its objects even if every worker is healthy.

Failure modes worth memorising

One consequence follows. Because you cannot reason about a failure you cannot see, the platform must make the cluster's state visible before the failure arrives. The observability section covers that. First, the failure you can prevent outright.

Making the head survivable

GCS fault tolerance is the first piece of work a platform owner does, because it converts a head failure from "rebuild everything" into "a short pause".

The mechanism is simple to state. The GCS normally keeps its state in process memory. With fault tolerance, it writes that state to an external Redis. When the head restarts, the new GCS reads the state back and the existing workers reconnect. You enable it by pointing the head at Redis:

RAY_REDIS_ADDRESS=redis.internal:6379 \
ray start --head --redis-password "$REDIS_PASSWORD" --redis-username default

Two behaviours matter. First, while the head is down, the cluster is degraded rather than dead: running tasks continue, but control-plane operations wait. Workers keep trying to reconnect, and the environment variable RAY_gcs_rpc_server_reconnect_timeout_s (default 60 seconds) sets how long they wait before concluding the GCS is gone. If your head takes longer than that to come back, for example because a replacement node must be provisioned, raise it, or the workers will terminate themselves just before the head returns. Second, the Redis is now a dependency you must run. Treat it as a production datastore: persistence on, replicated, monitored, in the same failure domain as the cluster or deliberately outside it.

On Kubernetes the official guidance is that KubeRay supports Redis-backed GCS fault tolerance, and the documentation presents it as supported for Ray Serve. If you run RayService for online inference, enable it.

There is also an alpha embedded option using RocksDB, configured with RAY_gcs_storage=rocksdb and RAY_gcs_storage_path=<dir>, with tuning variables RAY_gcs_rocksdb_io_pool_size (default 4) and RAY_gcs_rocksdb_strand_buckets (default 64). It is single-writer, Linux-only, and the path must survive a head restart. Because it is labelled alpha, I would not make it the foundation of a platform other teams depend on. Know it exists; choose Redis.

Fault tolerance of the control plane does not cover the data plane. For that you rely on task retries, actor restarts, and, for anything that matters, checkpoints in durable storage. Nodes are ephemeral; the platform contract should say so in plain words: local disk is scratch, and anything you need after the job ends goes to S3, GCS, Azure Blob or a shared filesystem.

Scaling: which component breaks first

Ray scales by adding nodes, but the components do not scale together. A platform owner needs to know the order in which things saturate, because that order tells you where to put the monitoring and the limits.

The head node comes first. Every node, every task event, every dashboard query and every job submission passes through it. Symptoms of a saturated head are a sluggish dashboard, slow ray list calls, and delayed scheduling decisions. Two contributors are worth knowing. Task events (the record of every task's lifecycle that powers the State API and dashboard) generate load proportional to task volume. Release 2.58 added an option to move task-event handling off the GCS hot path, behind the flag RAY_enable_task_events_to_dashboard_head. Second, the dashboard and Jobs API run on the head, so a heavy use of either competes with the GCS. Size the head for the largest cluster you intend to run, not the smallest.

The object store comes second. Each node has a shared-memory object store, and large intermediate results live there. When it fills, Ray spills to disk, and when spilling cannot keep up you see ObjectStoreFullError or simply slow jobs. The default size is a fraction of node memory; tune it for data-heavy workloads, and in containers give /dev/shm enough room (Docker's --shm-size) or Ray falls back to /tmp and warns you. Ray Data's streaming execution and backpressure exist precisely to keep intermediate data bounded; a pipeline that materialises everything defeats them.

Memory on workers comes third, and it is the most common cause of incidents. Ray's resource model is logical. num_cpus=1 means the scheduler will place one such task per CPU, not that the task is limited to one CPU. The same holds for memory=: it is a scheduling reservation, not a cap. So a single task that allocates twice what it declared can push the node toward exhaustion. Ray's memory monitor defends against this by killing workers once node memory passes a threshold (RAY_memory_usage_threshold, 0.95 by default), and the killed task surfaces as an OutOfMemoryError. For a platform this has a direct consequence: declare realistic memory= values and enforce them in review, because the scheduler can only pack safely what it has been told.

Scheduling throughput comes last. Ray can schedule a great many small tasks, but tasks that run for a few milliseconds spend more time in scheduling and serialisation than in your code. If a user complains that Ray is slower than a loop, the usual answer is that they should batch work into larger tasks or use Ray Data's map_batches, not that the cluster needs more nodes.

Three autoscalers, one bill

A newcomer assumes Ray has one autoscaler. It has three, at different layers, and they must be configured to agree.

1. The cluster autoscaler adds and removes nodes based on pending resource demand: tasks and actors that cannot be placed. On Kubernetes this is enabled on the RayCluster with enableInTreeAutoscaling; outside Kubernetes it is configured in the cluster launcher YAML. Its main controls are minimum and maximum workers per node type, an upscaling speed, and idleTimeoutSeconds, which decides how long an idle node survives.
2. Ray Serve's autoscaler changes the number of replicas of a deployment based on load, using autoscaling_config with min_replicas, max_replicas and target_ongoing_requests. It asks for more replicas; the cluster autoscaler must then supply the resources.
3. Ray Data's autoscaling adjusts the actor pools that run a pipeline. Release 2.57 removed the old Ray Data v1 cluster autoscaler, so older tutorials describing it are out of date.

The failure pattern is a chain with a delay at each link. A traffic spike makes Serve want more replicas; the replicas are unschedulable; the cluster autoscaler launches a node; the node takes minutes to boot and pull an image; meanwhile requests queue and back up. The remedy is not one setting but a budget: decide the time to add capacity end to end, and set minimums, queue limits and warm pools to cover it.

The related tool is the placement group. Gang scheduling reserves a set of bundles together (strategies PACK, SPREAD, STRICT_PACK, STRICT_SPREAD), and a distributed training job needs exactly that, because eight workers that start one at a time while the other four wait for a GPU is a deadlock in slow motion. Placement groups are also a fairness problem: a large group that cannot be satisfied sits pending while smaller requests get through. Know what is pending with ray status.

Security: the trust model

This is the section I would put first in an onboarding document, because the default behaviour surprises people and the consequence of getting it wrong is total.

Ray has no authentication by default, and the project's security documentation says it allows any client to run arbitrary code. Anyone who can reach the dashboard on port 8265, the Jobs API, the Ray Client port, or the cluster's internal ports can execute code on your cluster with the privileges of the Ray processes. This is not a bug to be patched. It is the design: Ray is a system for running your own code on many machines, so a network client that can submit a job is, by definition, trusted to run code.

The consequence is that the security boundary is the network, not Ray. Your platform must enforce it from outside:

• Run the cluster on a private network, with security groups or Kubernetes network policies that allow only the intended clients to reach the dashboard and Jobs API.
• Never expose port 8265 to the public internet. Scanners look for it.
• If users need browser access to the dashboard, put an authenticating reverse proxy (your identity provider, with TLS) in front of it, and decide who may reach the Jobs endpoint behind it.
• Restrict ports between nodes to what the cluster needs, including the GCS port (6379 by default).

Release 2.52 introduced token authentication, controlled by the environment variables RAY_AUTH_MODE and RAY_AUTH_TOKEN among others. Treat it as defence in depth. It raises the cost of a casual mistake, and it is worth enabling, but a shared token is not an identity system: it does not tell you who did what, and it does not separate one team from another.

Deserialisation is code execution

Ray moves data and functions with cloudpickle, and unpickling untrusted bytes executes code. Two consequences follow. Never load objects, checkpoints or datasets from sources you do not control into a Ray process. And keep current: release 2.58.0 fixed a Ray Data remote code execution issue reachable through read_lance and nested pickle objects, and a Serve problem in which a replica's ASGIService could bypass token authentication. Those are concrete reasons that a stale cluster is a security liability, not just a missing-features problem.

Secrets need their own rule. Environment variables set through runtime_env appear in job submissions and can be visible through the dashboard and logs, so do not put credentials there in committed code. Prefer workload identity (IAM roles for service accounts on AWS, GKE Workload Identity, Azure workload identity) so that pods obtain short-lived cloud credentials with no secret at all. Where a secret is unavoidable, mount a Kubernetes Secret into the pod, or read it from a secret manager such as Vault when the worker starts. The principle matches the one in Docker: a secret that reaches a layer, a log or a job record is permanent.

Release 2.58 also introduced an experimental Ray Sandbox that uses gVisor to isolate task and actor code and can run Docker-built images. It points in a promising direction for running less-trusted code, but it is experimental, so evaluate it rather than depend on it.

Multi-tenancy: one cluster is one trust boundary

The question every platform owner is asked is "can all our teams share one big cluster?" The honest answer from Ray's own design is not safely, if the teams do not trust each other.

Inside a single Ray cluster there is no strong tenant isolation. All users share the object store and run arbitrary code, and a job can inspect or interfere with other workloads on the same nodes. Ray namespaces (ray.init(namespace=...)) are worth understanding but they scope the names of actors, not their access: they prevent two jobs from colliding on an actor name, not one job from attacking another.

This leaves two defensible models:

1. One cluster per team or trust boundary. On Kubernetes this is the natural fit: each team gets its own RayCluster (or RayJob) in its own namespace, with resource quotas, network policies and its own service account. Isolation comes from Kubernetes and the network, which you already know how to operate (see Kubernetes).
2. A shared cluster for trusted users only, where everyone is in the same team or the same trust domain and the goal is simply higher utilisation. This is reasonable for a research group, and wrong for a service shared between a payments team and a data-science contractor.

The economics push in the opposite direction, which is the tension you manage. Many small clusters waste idle capacity; one large cluster raises utilisation but widens the blast radius. The usual resolution is to let the cluster layer be ephemeral and cheap to create: a RayJob creates a cluster for the job and can delete it when the job finishes (shutdownAfterJobFinishes), so a team pays for compute only while it runs, and utilisation is solved by the Kubernetes node pool and its autoscaler beneath the Ray clusters rather than by sharing one Ray cluster.

Running Ray on Kubernetes

Production Ray belongs on Kubernetes for most organisations, through the KubeRay operator. It gives you three custom resources, and choosing the right one is the main design decision.

• RayCluster declares a cluster: a head group, one or more worker groups, and optionally the in-tree autoscaler. You manage its lifecycle yourself.
• RayJob declares a job together with the cluster it needs. KubeRay creates the cluster, submits the job, waits, and with shutdownAfterJobFinishes removes the cluster. This is the right default for batch and training.
• RayService declares a Ray Serve application and manages upgrades with zero downtime, by bringing up a new cluster and switching traffic once it is healthy.

A trimmed RayCluster shows where the platform decisions live:

apiVersion: ray.io/v1
kind: RayCluster
metadata:
  name: team-a
  namespace: team-a
spec:
  rayVersion: "2.58.0"
  enableInTreeAutoscaling: true
  headGroupSpec:
    rayStartParams:
      num-cpus: "0"
    template:
      spec:
        containers:
          - name: ray-head
            image: rayproject/ray:2.58.0
            resources:
              requests: { cpu: "2", memory: 8Gi }
              limits: { memory: 8Gi }
  workerGroupSpecs:
    - groupName: cpu
      replicas: 2
      minReplicas: 2
      maxReplicas: 20
      rayStartParams: {}
      template:
        spec:
          containers:
            - name: ray-worker
              image: rayproject/ray:2.58.0
              resources:
                requests: { cpu: "4", memory: 16Gi }
                limits: { memory: 16Gi }

Read it for the decisions rather than the syntax. The head has num-cpus: "0" so it runs no user tasks. The image is pinned to an exact version, matching rayVersion. Memory requests equal limits so the Kubernetes scheduler and the container runtime agree on what the pod may use. Field names and the Helm charts (kuberay-operator, ray-cluster) change between KubeRay releases, so generate your real manifests from the version of the operator you run rather than copying this one.

Three Kubernetes-specific habits pay off. Separate node pools for CPU and GPU workloads, with taints so that CPU pods do not land on expensive GPU nodes. Pin images by version (and ideally by digest) and keep the Ray version in the image identical across head and workers. Install the operator and the clusters through Helm and a GitOps tool such as Argo CD, so cluster definitions are reviewed, versioned and reproducible, with the infrastructure beneath them managed by Terraform.

In the Gulf and in Egypt, data-residency rules often decide the topology before engineering does. Running KubeRay in a cluster located in the region where the data must stay, with storage and the container registry in the same region, is simpler to defend to an auditor than a global cluster with regional exceptions. Pick the cloud region first, then size the platform.

Observability and incident response

A cluster you cannot see is a cluster you cannot operate. Ray gives you four layers, and a platform should wire all of them.

The dashboard (port 8265) shows nodes, actors, jobs and logs. It is where a user starts, and it is also on the head node, so it is not what you want to depend on when the head is struggling.

The State API and CLI answer questions programmatically: ray list actors, ray list tasks, ray list nodes, ray summary tasks, ray logs, and ray memory for object references. ray status summarises resource usage and autoscaler demand. Use these in runbooks, because they work from a terminal during an incident.

Metrics are exported for Prometheus, and Ray generates Grafana dashboards. Scrape them with Prometheus and display them in Grafana. The signals worth alerting on are not exotic: pending resource demand that does not drain, object store utilisation and spill volume, node count against the expected, memory-monitor kills, Serve queue length and request latency, and the age of the oldest pending placement group.

Logs live on each node, under /tmp/ray/session_latest/logs. On ephemeral nodes that is exactly where they vanish, so ship them to central storage (an ELK stack or your cloud's logging service) before you need them. For deeper tracing, OpenTelemetry is the standard direction, and ray.timeline() produces a Chrome-tracing view of task execution that is excellent for finding scheduling gaps.

Incident playbooks

Write each of these as a runbook entry with the exact commands, the expected output, and who to escalate to. The value of the table above is not in the diagnosis; it is in having decided the diagnosis before the page arrives.

Operating Ray Serve as a service

Ray Serve has its own guide, so this section is only what a platform owner adds: the operational contract for a latency-sensitive service on a system designed for batch.

A Serve application is a graph of deployments; each deployment is a group of replicas, which are Ray actors, behind proxies that accept HTTP and gRPC traffic. Production configuration belongs in a YAML file that you version and deploy with serve deploy, rather than in Python calls, so the configuration can be reviewed:

applications:
  - name: fraud
    route_prefix: /fraud
    import_path: fraud.app:app
    deployments:
      - name: Model
        max_ongoing_requests: 16
        autoscaling_config:
          min_replicas: 2
          max_replicas: 12
          target_ongoing_requests: 8
        ray_actor_options:
          num_cpus: 2

Use the current names: max_ongoing_requests and target_ongoing_requests replaced the older max_concurrent_queries and target_num_ongoing_requests_per_replica, so an old tutorial's configuration may silently mean something different on a new version. Check the Serve configuration reference for the exact set of fields your version accepts before you standardise a template.

The service-level questions are these. Backpressure: a deployment with a bounded queue rejects excess requests rather than collapsing, and the client sees a 503; release 2.58 made the status code and a Retry-After header configurable. Decide the limit from your latency budget, and make clients retry with jitter. Upgrades: RayService performs zero-downtime upgrades by standing up a new cluster and shifting traffic, which costs a temporary doubling of capacity, so budget for it. Cost: replicas that hold GPUs are your largest expense; fractional GPU requests (num_gpus=0.5), dynamic batching with @serve.batch, and model multiplexing with @serve.multiplexed let several small models share hardware. For large language models, Serve's LLM support (which 2.58 extended with KV-cache and token-aware routing) is built around vLLM; evaluate it against running vLLM directly, and decide whether you need Ray around it at all.

Cost

Ray makes it easy to spend money quickly, because a loop can request a thousand CPUs. A platform needs cost controls in three places.

Shut things down. The largest waste is idle capacity: a cluster left running over a weekend, a GPU head, a worker group whose idleTimeoutSeconds is days. Prefer ephemeral clusters through RayJob, set sensible idle timeouts, and make a cluster's owner and expiry visible with labels so a scheduled sweep can find strays.

Use cheaper capacity safely. Spot or preemptible workers cost much less and Ray is built to tolerate them: tasks retry, actors restart, objects reconstruct. But tolerance is only as good as your checkpointing. For training, checkpoint to durable storage on a regular cadence, because a preempted worker without a checkpoint restarts from zero. Release 2.58 added a preemption API to Ray Train to help with this. Keep the head and any single-instance state on regular capacity.

Match the work to the hardware. Run CPU-bound preprocessing on CPU nodes and keep GPUs for the model. In Ray Data, separating a CPU read-and-decode stage from a GPU inference stage lets each scale on its own resource, which usually beats one big GPU node doing both. Right-size the GPU: a small model on a whole GPU is waste that fractional resources can recover.

Then make the cost visible. Attribute spend per team through namespace and node labels and your cloud's cost reports, so that the person who requested the thousand CPUs sees the bill. A platform that hides cost gets used as if cost did not exist.

Upgrades and migrations

Ray's release cadence is its defining operational property. At roughly one minor version every two to six weeks, a platform that upgrades "when convenient" is always many versions behind, and each skipped release adds deprecations you must absorb at once.

The official position is firm: Ray does not guarantee compatibility across minor versions between head and workers, and every node and client must run the same Ray and Python versions. This rules out the rolling in-place upgrade that works for most services. The pattern that works is replace, don't mutate: build a new cluster on the new version, move workloads to it, and retire the old one. RayService automates this for Serve. For batch work, RayJob makes it natural, because each job already creates its own cluster, so an upgrade is just changing the image tag in the job template.

A sustainable cadence looks like this. Subscribe to the release notes. Every month or two, read the notes for each minor version between yours and the target. Upgrade a staging cluster, run a representative workload suite, and then promote. Keep a pinned compatibility matrix of Ray version, Python version, CUDA and framework versions, because the Python and GPU stack move with Ray.

Recent releases show the kind of breaking changes to scan for:

• 2.56 dropped Pydantic v1 support, removed tfx-bsl support from read_tfrecords, disabled DataSourceV2 by default, and deprecated low-level scheduling APIs and HTTPOptions.location.
• 2.57 removed the Ray Data v1 cluster autoscaler and deprecated ConcurrencyCapBackpressurePolicy, DataIterator.to_torch and pandas UDF batches.
• 2.58 deprecated ray_remote_args_fn and Dataset.zip, and carried the security fixes described earlier.

The longer-running migrations are the ones users trip over in old tutorials. The Ray AIR namespace was retired, so ray.air imports in old code do not work. Ray Train has moved to a newer flow, so legacy Trainer.fit patterns differ from current examples, and you should confirm the exact import paths against the Train documentation for your version. Ray Client (the ray:// address) is in maintenance mode, and Ray Jobs is the preferred way to submit work. A useful platform habit is to publish a blessed template per workload type (training, batch inference, Serve), keep those templates current, and tell users to start from them rather than from a search result.

Integrating with the rest of the stack

A platform is judged by how it fits its neighbours. The integrations that matter most for a Ray platform:

• Experiment tracking. Tune and Train have callbacks for MLflow and Weights and Biases. Standardise one tracking server and a naming convention, so that results from different teams are comparable and a model's lineage points back to a run.
• Data. Ray Data reads Parquet, and release 2.58 added write_delta for Delta Lake. If your data platform is built on Spark, Ray is typically the engine for the model-shaped part (GPU batch inference, training data loading) while Spark remains the engine for large relational transformations. Ray also offers a way to start a Ray cluster on a Spark cluster through ray.util.spark; verify its status in the documentation before you plan around it.
• Training frameworks. Ray Train wraps PyTorch, PyTorch Lightning, Hugging Face, XGBoost and others; for large-model work, DeepSpeed integrates as a training backend.
• Orchestration. A scheduler such as Airflow or Kubeflow decides when work runs and Ray decides how it runs in parallel. Have the scheduler submit a RayJob or a Jobs API call and read its status; do not hide long Ray drivers inside scheduler workers.
• Containers. Build your Ray images with the practices from the Docker guide: pinned bases, no secrets in layers, non-root users, and a software bill of materials. Remember --shm-size for local container runs.

The common thread is a contract at each boundary: where data is read, where checkpoints and artifacts are written, which tracking server is used, how jobs are submitted. A platform with those contracts documented and templated can be adopted by a new team in days.

Governance

Governance on a Ray platform is less about forms than about a short list of decisions made once and enforced mechanically.

Who may submit work, and to which cluster. Because Ray itself will not authenticate users, access is enforced by the network and the proxy in front of it. Keep the mapping from identity to cluster in your identity provider, not in tribal knowledge.

What a job may request. Cap the resources per team with Kubernetes quotas. Require memory= declarations. Review any actor that declares zero CPUs, because the scheduler cannot account for work it was told is free.

What goes into an image. One approved base image per Python and CUDA combination, rebuilt on a schedule, scanned with a tool such as Trivy, and referenced by version. Users add dependencies through a reviewed layer or a runtime_env limited to pure-Python packages; heavy native installs at job start make jobs slow and non-reproducible.

What gets recorded. Every job should carry an owner, a cost centre and a tracking run, as labels or job metadata. When something fails or overspends, "who owns this?" must take seconds to answer.

How exceptions work. A team will need a larger cluster, a newer Ray, or a looser network rule. Have a lightweight, written path to ask, with a time limit on the exception, so exceptions expire instead of accumulating.

Where Ray stops

Part of owning a platform is saying no, and the honest answer to several requests is that Ray is the wrong tool.

Ray is not a security boundary. If the requirement is hostile multi-tenancy, with tenants who must not see each other's data or code, do not build it from one Ray cluster. Use separate clusters at minimum, or a different system.

Ray is not a data warehouse or a relational engine. For large SQL-shaped transformations, joins and aggregations over tables, Spark or a warehouse is better engineered and better understood. Ray Data is strong at model-adjacent, per-record or per-batch work, not at replacing your lakehouse.

Ray is not a workflow scheduler. It does not give you cron, dependencies between jobs, or backfills. Pair it with Airflow, Dagster or a Kubeflow pipeline.

Ray is not necessary for a single machine or a small job. If a process pool or a managed training job meets the need, the operational surface of a cluster, a GCS, an autoscaler and a security perimeter is cost without benefit. Ray earns its place when the work genuinely needs many machines, heterogeneous resources, or a unified runtime for data, training and serving.

Ray is not the only way to serve models. For a single model with modest traffic, a plain containerised service on Kubernetes is simpler to run and to reason about. Other serving systems cover specific needs: KServe for a Kubernetes-native model-serving standard, BentoML for packaging, Triton for GPU-optimised multi-framework inference. Choose Serve when you want its composition of many models and Python-native business logic on the same runtime as your training and data.

Managed options exist. Anyscale, the managed platform from Ray's creators, and the managed ML services of the major clouds trade control for operations. Compare the full cost, including your engineers' time to run the platform, rather than only the instance prices.

The review checklist

Use this for a design document or a pull request that changes a Ray deployment.

The complete picture

Put it together as a single story. A team submits a RayJob from a reviewed template. KubeRay creates a cluster in the team's namespace, subject to its quota, on a pinned image built from your approved base. The head runs no user tasks and, for long-lived clusters, stores its state in Redis. Workers scale with the in-tree autoscaler within the limits you set, and spot capacity is used for fault-tolerant work with checkpoints written to object storage. The job reads data from the lakehouse, logs to your tracking server, and writes results to durable storage. The dashboard is not reachable from outside; operators reach it through an authenticating proxy. Metrics flow to Prometheus and Grafana, logs to central storage, and one alert watches pending demand. When the job finishes, the cluster is deleted. When the next Ray release arrives, you change an image tag in a template, test it in staging, and promote.

None of that is exotic. Each element answers one of the failure modes in the opening section, which is the real test of a platform: you can point at each piece and say which failure it exists to handle.

Where the series leaves you

You began this series by calling .remote() on a function. You now own the thing underneath it: a runtime whose control plane needs protecting, whose resource model is a promise rather than a limit, whose security depends on the network around it, and whose release cadence demands a habit of replacing rather than patching.

The skills that matter most from here are judgement skills. Know which failure you are designing against. Know which of your teams trust one another. Know when the answer to "can Ray do this?" is "yes, but you should not". Keep reading the release notes, keep running the drills, and keep your platform documentation as current as your cluster. For the neighbouring pieces, continue with Ray Serve, Kubernetes, and Spark.

Sources

• Ray documentation home
• Installing Ray
• Ray Core walkthrough
• Ray Core scheduling
• Ray security
• GCS fault tolerance
• Ray Serve
• KubeRay on Kubernetes
• Ray observability
• Ray releases on GitHub
• KubeRay repository

Part three of three. A senior Ray interview is rarely about syntax. It tests whether you can design a platform on Ray, reason about its failures under pressure, and argue a trade-off to people who do not care about actors. This file is organised as system-design prompts, incident questions, and leadership conversations, with the shape of a strong answer for each. About forty minutes. Version context: Ray 2.58.

The thirty-second answer

Ray is a distributed runtime for Python. Its core gives you tasks, actors and a shared object store; libraries for data, training, tuning and serving are built on top. As a platform owner I treat a Ray cluster as one trust boundary with no built-in authentication, a head node whose control plane needs fault tolerance, logical resources that are promises rather than limits, and a release cadence that means I replace clusters rather than patch them.

If you can say that and then name the four consequences (network security, GCS fault tolerance, declared memory, replace-don't-mutate upgrades), you have shown the interviewer you have operated it, not just run it.

What interviewers listen for

Strong senior answers have four habits in common, and weak ones lack them.

1. They name the failure before the feature. "I would enable external Redis for the GCS because losing the head otherwise loses the cluster" beats "Ray has a fault tolerance option".
2. They state the trust boundary. Any answer involving several teams that does not mention that Ray has no authentication by default and no strong tenant isolation is incomplete.
3. They quantify. Time to add a GPU node, idle cost per day, the reconnect timeout against head recovery time.
4. They say when not to use Ray. Naming the cases where a simpler tool wins is a sign of experience, not of doubt.

System-design prompts

Design a shared ML platform on Ray for five teams

The prompt. "We have five teams, two of which handle regulated data. Design a Ray platform."

A strong answer opens with the trust model: Ray gives no tenant isolation inside a cluster, so the cluster is the isolation unit, and the regulated teams get separate clusters, namespaces, network policies and service accounts. Then the shape:

• Kubernetes with KubeRay. RayJob for batch and training so each run gets a clean, ephemeral cluster that is deleted on completion; RayService for online inference; a long-lived RayCluster only for trusted interactive work.
• Utilisation comes from the Kubernetes node pool and its autoscaler beneath the Ray clusters, not from sharing one Ray cluster.
• Separate CPU and GPU node pools with taints.
• Pinned images from an approved, scanned base, one Ray and Python version across head, workers and clients.
• Durable storage for checkpoints and outputs, reached through workload identity rather than keys.
• Quotas per namespace; labels for owner and cost centre; cost reports per team.
• Region: for employers in the Gulf or Egypt with residency requirements, the clusters, storage and registry live in the required region.

What a weak answer misses: it proposes one big shared cluster for utilisation and never mentions isolation, or it mentions Kubernetes without explaining which custom resource fits which workload.

Follow-up to expect. "Why not one cluster with namespaces?" Ray namespaces scope actor names, not access. Any job can run arbitrary code and share the object store, so a namespace is not a security control.

Design low-latency inference on Ray Serve with spiky traffic

The prompt. "A fraud model must answer in under 100 ms at the 99th percentile, traffic spikes tenfold at certain hours, and GPUs are expensive."

A strong answer separates three layers and the delay between them. Serve's autoscaling_config (min_replicas, max_replicas, target_ongoing_requests) decides replica count; the cluster autoscaler must supply nodes; a GPU node can take many minutes to appear. So: measure the end-to-end time to add capacity, set min_replicas to cover the known baseline and the predictable spikes (scheduled scaling or pre-warming if the spike is predictable), and use max_ongoing_requests and queue limits so overload becomes a fast 503 with a retry hint rather than a latency collapse. Reduce cost with fractional GPUs (num_gpus=0.5), @serve.batch for batching, and multiplexing for many small models. Run it as a RayService so upgrades are zero-downtime, with Redis-backed GCS fault tolerance, because the head's loss must not take the service down.

What a weak answer misses: relying on autoscaling alone for latency, or ignoring the node-boot delay.

Follow-up. "When would you not use Serve?" For one small model with steady traffic, a plain containerised service on Kubernetes is easier to operate. See Ray Serve and KServe for the alternatives.

Design distributed training that survives spot preemption

The prompt. "We train on spot GPUs to save money. Preemptions keep wasting days of work."

A strong answer starts from the fact that nodes are ephemeral and Ray tolerates their loss only to the extent that state is checkpointed. Report checkpoints through Ray Train to durable object storage at a cadence chosen from the cost of lost work against the cost of writing; configure the run to resume from the latest checkpoint; use a placement group so that the workers are gang-scheduled and the job does not hold half its GPUs while waiting for the rest; keep the head on on-demand capacity. Mention that Ray 2.58 added a preemption API to Train, and that you would read its documentation before depending on it. Measure the result: fraction of paid GPU-hours that produced kept progress.

What a weak answer misses: checkpoints on local disk, or no mention of gang scheduling.

Design a batch inference pipeline over 200 million records

A strong answer uses Ray Data with a streaming read, a CPU stage for decoding and preprocessing, and a GPU stage with a callable class in map_batches so the model loads once per actor, with concurrency and num_gpus set per stage so each scales independently. Backpressure keeps intermediate data bounded, so the object store does not fill. Write with write_parquet, or write_delta if the lakehouse is Delta (new in 2.58). Idempotence matters: if a task is retried, the output must not duplicate. Mention that the large relational parts of the job (joins, aggregations) may belong in Spark upstream, with Ray handling the model-shaped stage.

Design the upgrade process for a platform with 30 clusters

A strong answer begins with the constraint: Ray does not guarantee compatibility across minor versions between head and workers, so in-place rolling upgrades are unsafe. The process is replace, don't mutate. Batch clusters upgrade by changing the image tag in the RayJob template. Serve upgrades through RayService. Long-lived clusters are rebuilt blue/green. A staging cluster runs a representative workload suite against each new minor release; release notes are read for every version between current and target; a compatibility matrix tracks Ray, Python, CUDA and framework versions. An SLA bounds how far behind any cluster may be, justified partly by security: release 2.58 fixed a Ray Data remote code execution issue and a Serve token-authentication bypass.

Incident-response questions

For each, the interviewer wants the first command, the hypothesis tree, and the durable fix.

"Jobs are stuck pending and nobody can run anything."

First command: ray status. It shows resources in use and pending demand. The tree:

• Demand larger than any node (for example a request for more GPUs on one node than exist): the request is infeasible; fix the declaration.
• Actors holding all CPUs: long-lived actors have claimed the cluster; find them with ray list actors.
• Valid demand, no new nodes: the cluster autoscaler is not launching. Check its limits, the cloud quota, and whether the node type matches the demand.
• A large placement group pending: it is waiting for a gang that cannot form; smaller requests may be starving or being served first.

Durable fix: quotas, declared resources checked in review, and an alert on demand that has not drained.

"Workers keep dying with out-of-memory errors."

The memory monitor kills workers once node memory passes its threshold (RAY_memory_usage_threshold, 0.95 by default), and the task sees an OutOfMemoryError. The hypotheses: tasks use more than they declared; too many tasks per node because memory= was not set; a driver or actor holds large references; the object store has taken memory the workers need. Commands: ray summary tasks, ray memory, the per-node logs. The weak answer is "use bigger nodes". The strong answer finds who is over-using, fixes the declaration or the concurrency, and only then considers sizing. Durable fix: require memory= declarations and track memory-monitor kills as a metric.

"The head node died."

The answer depends on whether GCS fault tolerance was enabled. If yes: the replacement head reads state from Redis and workers reconnect within RAY_gcs_rpc_server_reconnect_timeout_s (60 seconds by default); the question is whether head recovery fits inside that window, and whether Redis itself is healthy. If no: the cluster is effectively lost, so capture logs, rebuild, resubmit, and then do the work you should have done before. A strong candidate also asks what the head was doing: if user tasks were scheduled on it, the cause is likely a memory failure on a node that should have been idle, and the fix includes setting the head's CPUs to zero.

"Someone found our Ray dashboard on the public internet."

Treat it as a security incident, not a configuration ticket. Ray has no authentication by default, so an exposed dashboard or Jobs API means anyone could have run arbitrary code with the cluster's privileges, including reading anything its identity can reach. Immediate steps: remove the exposure, preserve logs, rotate every credential the cluster could read (cloud roles, secrets in the environment, tokens), review job history and cloud audit logs for unfamiliar jobs, and rebuild the cluster rather than trusting it. Then the prevention: private networking, authenticating proxy, network policies, token authentication as an additional layer, and a scan that alerts on exposure. Do not claim token authentication would have been sufficient on its own; it is defence in depth.

"The new Ray version broke half our pipelines."

Ask which version jump and read the release notes for each minor in between. Typical culprits include removed or deprecated APIs (in recent releases, Ray Data items such as Dataset.zip and ray_remote_args_fn, DataIterator.to_torch), Pydantic v1 no longer supported since 2.56, and changed defaults. Roll back by redeploying the previous image on a new cluster, because mixing versions in one cluster is unsupported. Durable fix: a staging suite, a deprecation-warning scan in CI, and smaller, more frequent upgrades.

"Serve is returning 503 under load."

A deployment that has reached its queue limit rejects requests. Check whether replicas are scaling, whether they are stuck starting because their ray_actor_options cannot be scheduled, and whether the cluster has room. Remember the chain: Serve autoscaler, then cluster autoscaler, then node boot. The fix may be a higher minimum, a larger queue within the latency budget, or a faster-booting node image, and clients should retry with jitter.

"A job produces duplicate rows after a node failure."

The task was retried and was not idempotent. Ray retries tasks on system failure by default. The fix is to make the output write idempotent (a deterministic output path, an upsert, or a commit step), or to set max_retries=0 for tasks where duplicates are worse than failure. This question checks whether you understand retries as a semantic choice, not a free reliability feature.

Trade-off questions

Ray or Spark?

They overlap less than their marketing suggests. Spark is a mature engine for large relational and ETL workloads, with a SQL optimiser and a long history in data platforms. Ray is a general Python runtime with a fine-grained task and actor model, strong for model-shaped work: GPU batch inference, distributed training, hyperparameter search, serving, reinforcement learning. A strong answer uses both: Spark for the large table transformations, Ray for the stage that needs GPUs or Python-native parallelism, joined at the storage layer.

Ray Serve or a plain service on Kubernetes?

For one model with predictable traffic, a containerised service with Kubernetes autoscaling has fewer moving parts: no GCS, no second autoscaler. Serve earns its place when you compose several models and business logic in one application, need dynamic batching, fractional GPUs and multiplexing, or want serving to share a runtime with your training and data code. The cost is operating a Ray cluster to serve traffic.

One big cluster or many small ones?

One big cluster has better utilisation and a wider blast radius, and no tenant isolation. Many ephemeral clusters cost startup time and need a good template system, but give isolation, clean environments, and cost attribution. The usual senior answer is many clusters on a shared Kubernetes node pool, with RayJob as the default.

Self-managed KubeRay or a managed platform?

Be neutral and concrete. Self-managed buys control, region choice and lower unit cost, and spends engineering time on upgrades, security and on-call. A managed platform (Anyscale, or a cloud's managed ML service) buys operations and spends control and money. The deciding factors are team size, how much the platform is a differentiator, data-residency constraints, and how many engineers you can dedicate. Compare total cost, not instance prices.

Redis fault tolerance or accept head loss?

For long-lived clusters and Serve, enable Redis-backed GCS fault tolerance: the Redis dependency is cheaper than the outage. For short ephemeral RayJob clusters where a failure simply restarts the job, the added dependency may not be worth it. The embedded RocksDB option is alpha, so it is not a basis for a platform.

Arguing a trade-off to leadership

The senior skill the title implies is translating. Here are four conversations and the framing that works.

"Why can't all teams share one cluster? It is cheaper." Say: "It is cheaper per idle hour and more expensive per incident. A shared cluster has no isolation between teams, so one team's mistake, or one compromised job, reaches everyone's data. We get the utilisation benefit one layer down, in the Kubernetes node pool, while each team keeps its own boundary." Offer the exception: a single trusted research group can share.

"Why do we need to upgrade so often?" Say: "The project releases every few weeks, does not guarantee compatibility between minor versions, and recently fixed a remote code execution issue and an authentication bypass. Each skipped release makes the next upgrade bigger and riskier. A small, scheduled, tested upgrade each cycle is the cheapest path." Propose the SLA and the staging suite.

"Why is the GPU bill so high?" Show the breakdown: idle time, over-sized GPUs for small models, always-on heads. Then the levers with an estimated saving each: ephemeral clusters, spot with checkpointing, fractional GPUs, batching. Attach a per-team report so spending has an owner.

"Can we just use the managed service?" Do not argue from ideology. Lay out the cost of running the platform (people, on-call, upgrades, security) against the managed premium, the data-residency constraint, and the exit cost. Recommend, with the conditions that would change your mind.

Questions to ask the interviewer

Strong candidates ask. Useful ones: How many clusters do you run, and who owns each? How do teams get isolation? What is your Ray version policy? Has the dashboard ever been reachable from outside? What was your last Ray-related incident, and what changed afterwards?

Rapid-fire

• Why can't I upgrade workers first? Head, workers and clients must match in Ray and Python versions.
• What is the GCS, in one sentence? The cluster's control-plane memory: nodes, actors, placement groups, jobs.
• Does num_cpus=1 limit CPU use? No. It is a scheduling reservation, not an enforcement.
• Does a namespace isolate tenants? No. It scopes actor names.
• What does ray status tell you first? Resource use and pending demand.
• What does the memory monitor do? Kills workers once node memory passes the threshold (0.95 default).
• What did 2.52 add? Token authentication.
• What replaced the AIR namespace? The libraries now live under their own packages, so ray.air imports from old code fail.
• Where should checkpoints live? In durable object storage, never on a node's local disk.
• Which Kubernetes resource for a training run? RayJob.

Sixty-second self-test

Answer aloud, then check against this file.

1. What happens to a running cluster when the head dies, with and without GCS fault tolerance?
2. Why is the dashboard port a security problem, and what are three controls?
3. Name the three autoscalers and the order in which a traffic spike travels through them.
4. Why must a retried task be idempotent?
5. When is one cluster per team better than a shared one, and why?
6. What is the upgrade method when versions cannot be mixed?
7. Name two cases where you would recommend something other than Ray.
8. What is the first command in a "jobs pending" incident?

If you stumble on more than two, reread the matching section of the detailed guide, then continue with Ray Serve and Kubernetes.

Part three of three. These are the traps that appear only when Ray is a shared platform running at scale, the habits that keep it healthy, and the checklists to use when reviewing other people's designs. Each trap has the symptom, the cause and the fix. Version context: Ray 2.58.

Production traps that only appear at scale

The head node is doing user work

Symptom. The dashboard slows, ray list times out, and then the whole cluster goes quiet. Cause. The head hosts the GCS, the dashboard and the Jobs API. By default it also accepts tasks, so a memory-hungry task scheduled there can starve the control plane or get the node killed. Fix. Give the head zero schedulable CPUs (num-cpus: "0" in the head's start parameters on KubeRay, or num_cpus=0 in the launcher configuration), size it for the largest cluster you expect, and enable external Redis for GCS fault tolerance where the head's loss would hurt.

Local disk is treated as durable

Symptom. A job that "worked yesterday" has no results today, or a preempted training run restarts from epoch zero. Cause. Workers are ephemeral. Logs, checkpoints and outputs written to a node's disk disappear with the node. Fix. Make the contract explicit: results and checkpoints go to object storage or a shared filesystem. In Ray Train, set a storage path and report checkpoints. Ship logs off the nodes before an incident, not after.

Resources are believed to be limits

Symptom. Random worker deaths with OutOfMemoryError, and a node running far hotter than the scheduler thought possible. Cause. num_cpus and memory are logical reservations used for placement. Nothing stops a task using more. The memory monitor kills workers when node memory passes its threshold (RAY_memory_usage_threshold, 0.95 by default), after the damage is done. Fix. Measure real use, declare realistic memory= values, and reject pull requests that leave them out. Set Kubernetes memory limits equal to requests so the container runtime and the scheduler tell the same story.

Retries that duplicate side effects

Symptom. Duplicate rows, double notifications or two charges after a node loss. Cause. Tasks are retried on system failure by default, and lineage reconstruction can re-run a task to rebuild a lost object. A task with side effects runs twice. Fix. Make side effects idempotent (deterministic output paths, upserts, a final commit step) or set max_retries=0 where a failure is better than a duplicate. Review every task that writes anywhere.

One actor owns everything

Symptom. OwnerDiedError across many unrelated tasks after a single actor restarts. Cause. Objects are owned by the process that created their references. A coordinator actor that creates references on behalf of everyone is a shared fate. Fix. Keep ownership distributed: let tasks produce their own references, keep references short-lived, and persist anything important to storage rather than relying on an actor to remember it.

A giant object captured in a closure

Symptom. Slow job starts, a warning that a remote function is very large, and driver memory growing with the number of tasks. Cause. A large array or model captured in a remote function's closure is serialised with the function, repeatedly. Fix. Put the object in the store once with ray.put, pass the reference as an argument, and let Ray share it by reference.

Millisecond tasks

Symptom. Ray is slower than the plain loop it replaced, and the head's CPU is high. Cause. Scheduling and serialisation cost per task dominates when tasks are tiny. Fix. Batch the work: larger tasks, or Ray Data's map_batches. Adding nodes does not help.

Task-event load on the control plane

Symptom. At very high task volumes, the dashboard and State API lag and the head's load climbs. Cause. Task lifecycle events flow through the control plane. Release 2.58 added an option to move that handling off the GCS hot path, controlled by RAY_enable_task_events_to_dashboard_head. Fix. Check the release notes and the observability documentation, test the option in staging with a realistic volume, and watch head CPU and memory before and after. Treat it as a tuning step to evaluate, not a default to apply blindly.

The reconnect window is shorter than head recovery

Symptom. After a head failure with Redis fault tolerance on, workers terminate themselves just before the new head is ready. Cause. Workers wait RAY_gcs_rpc_server_reconnect_timeout_s (60 seconds by default) for the GCS, and a replacement node that takes longer to start loses the race. Fix. Measure your head's real recovery time, including node provisioning and image pull, and set the timeout above it. Keep a warm replacement path where you can.

GPU scale-up latency in the request path

Symptom. Latency spikes during every traffic surge, with queues building while replicas are "starting". Cause. Serve wants more replicas, but the cluster needs a new GPU node that takes minutes to boot. Fix. Know the end-to-end time to add capacity, hold enough minimum replicas to cover it, bound queues so overload becomes a fast, retryable error, and consider pre-warming ahead of predictable spikes.

Mixed versions

Symptom. A Version mismatch error on connect, or subtle failures from a client library newer than the cluster. Cause. Head, workers and clients must run the same Ray and Python versions, and a rolling upgrade, a stale virtual environment, or a different base image breaks that. Fix. Pin the image tag, build clients from the same image, and upgrade by replacing clusters rather than patching them.

Security traps

The dashboard is reachable

Symptom. None, until someone else uses it. Cause. Ray has no authentication by default and allows clients to run arbitrary code. A load balancer, a port-forward left running or a permissive security group is enough to expose it. Fix. Private networking and network policies, an authenticating reverse proxy for people who need the dashboard, and a recurring external check that port 8265 is closed. Token authentication (from 2.52, via RAY_AUTH_MODE and RAY_AUTH_TOKEN) is an extra layer, not a replacement.

Credentials in runtime_env

Symptom. A key shows up in a job log, a screenshot of the dashboard, or a repository. Cause. Environment variables set in a runtime_env travel with the job and can be visible in the places people look when debugging. Fix. Use workload identity so pods get short-lived cloud credentials, mount Kubernetes Secrets, or read from a secret manager such as Vault at worker start. Search repositories for credentials in job definitions on a schedule.

Unpickling untrusted data

Symptom. None, which is the danger. Cause. Ray's serialisation can execute code when it deserialises, and release 2.58.0 fixed a Ray Data remote code execution issue reached through read_lance and nested pickle objects. Fix. Never load datasets, checkpoints or objects from sources outside your trust boundary, and stay current. Add the Ray version to your vulnerability tracking as you would any dependency.

One cluster for teams that do not trust each other

Symptom. A "shared cluster" works well until the first incident across teams. Cause. Ray has no strong tenant isolation in a cluster, and namespaces scope actor names, not access. Fix. One cluster per trust boundary, with Kubernetes namespaces, quotas and network policies providing the isolation. See the Kubernetes guide.

Platform-ownership habits

These are ordered by leverage: the first returns the most per hour spent.

1. Publish blessed templates. A reviewed RayJob template, a RayService template and a training template, each with pinned images, memory declarations, storage paths and labels. Most incidents are users reinventing something badly.
2. Make clusters ephemeral by default. A cluster per job, deleted when finished, removes whole categories of drift, leak and cost. Reserve long-lived clusters for a reason you can write down.
3. Pin and replace. One version across head, workers and clients; image tags fixed; upgrades done by building a new cluster. Keep a compatibility matrix for Ray, Python, CUDA and frameworks.
4. Upgrade on a schedule. Read the release notes for every minor you skip. Bound how far behind any cluster may be, partly for security.
5. Decide the trust model in writing. Which teams share a cluster, and why. Put it in the platform's documentation so it survives the person who decided it.
6. Write the runbooks before the incidents. Pending jobs, out-of-memory, object store full, head down, version mismatch, 503 from Serve. Each with the first command and the escalation.
7. Drill. Delete a head, kill a spot worker, fill an object store, on a test cluster, following only the runbook. Fix the runbook where it fails.
8. Attribute cost. Labels for owner and cost centre on every cluster, reports per team, and a sweep that deletes unlabelled clusters after a grace period.
9. Measure time to add capacity. For CPU and GPU, from demand to running task. Publish it, because users and Serve configurations depend on it.
10. Keep the contract small. Where data comes from, where checkpoints go, which tracking server, how jobs are submitted. The fewer decisions a team has to make, the fewer it makes badly.

Metrics worth putting on the platform's own dashboard

Review checklist for a pull request

Use this when a team changes a Ray job, a Serve application or a cluster definition.

• Does every task and actor declare num_cpus, num_gpus and memory realistically?
• Is every retried task safe to run twice? Are max_retries and actor restart settings chosen on purpose?
• Does any remote function capture a large object instead of receiving a reference?
• Are results, checkpoints and logs written somewhere durable?
• Are there any credentials in runtime_env, in job arguments, or in the image?
• Is the Ray version pinned and identical across head, workers and the submitting client?
• Does the head run zero user tasks?
• For Serve: are max_ongoing_requests, target_ongoing_requests, the replica bounds and the queue behaviour set from a latency budget? Are the field names current (not the older max_concurrent_queries)?
• For training: is the job gang-scheduled with a placement group where it needs several workers together, and does it resume from a checkpoint?
• Does the job use any API listed as deprecated in recent releases, such as Dataset.zip, ray_remote_args_fn or DataIterator.to_torch?
• Does the cluster have an owner, a cost label and an expiry?
• Is Pydantic v2 in the environment? Ray dropped v1 support in 2.56.

Review checklist for a design document

Ask these of any proposal that puts Ray at the centre of a system.

1. Why Ray? What requirement needs many machines, mixed resources or a unified runtime? If a process pool, a managed training job or a plain service would do, say so.
2. What is the trust boundary? Who can reach the cluster, and what can they do once they can? Where is the authentication, given that Ray has none by default?
3. What is the failure story? Head loss, worker loss, object loss, preemption. Which are tolerated, which need a restart, and what is the recovery time?
4. Where does state live? Anything that must survive a node must be named, with its storage, owner and retention.
5. How does it scale, and what breaks first? Head load, object store, worker memory or scheduling overhead. Which limit will you hit, and what will you watch?
6. What does it cost, and who sees it? Idle policy, spot use, GPU sizing, and the team the spend is attributed to.
7. How is it upgraded? The replacement process, the test suite, the maximum version lag.
8. How is it observed? Metrics, logs, alerts and the runbook for each alert.
9. What is the exit? If Ray stops fitting, which parts are Ray-specific code and which are portable?
10. Where does the data live? Region and residency, for regulated workloads in the Gulf or Egypt, covering the cluster, the storage and the registry.

Practice cards

Operational recipes

These are the commands to have ready. They are the ones the detailed guide uses, so they work against the version in the catalogue.

Inspect the state of a cluster during an incident:

ray status
ray list nodes
ray list actors
ray summary tasks
ray memory

ray status shows resource usage and pending demand; ray list queries the State API; ray summary tasks groups tasks by state; ray memory shows which references are holding objects.

Submit a job to a running cluster with its working directory:

ray job submit --address http://HEAD:8265 --working-dir . -- python script.py
ray job status <job-id>
ray job logs <job-id>
ray job stop <job-id>

Inspect and change a Serve application from configuration:

serve deploy serve-config.yaml
serve status

Pin the version everywhere you install, so that clients match the cluster:

pip install "ray[default]==2.58.0"

Confirm the install and see what the cluster believes it has:

ray --version
python -c "import ray; ray.init(); print(ray.cluster_resources())"

For a node that has fallen out of the cluster, the first stop is its log directory, /tmp/ray/session_latest/logs, and the first question is whether the logs were shipped elsewhere before the node vanished.

A short checklist to keep

• The head runs no user tasks and has fault tolerance where it matters.
• The network, not Ray, is the security boundary; the dashboard is private.
• One cluster per trust boundary.
• Declared resources are realistic; retried tasks are idempotent.
• State lives in durable storage, never on a node.
• Versions are pinned and identical; upgrades replace clusters.
• Time to add capacity is measured, and queues are bounded.
• Cost has an owner and an expiry.
• Runbooks exist and have been rehearsed.

If you want to go further, read Ray Serve for the serving side, Kubernetes for the platform beneath, and Prometheus for the metrics that make the checklist measurable.