The Ray Field Guide: From @ray.remote to a Production GPU Cluster

By Gopi Krishna Tummala

Companion to Module 2: Dataloaders, Pipeline Orchestration and the vLLM Trilogy. Those posts kept running into the same question: who launches the processes, places them on GPUs, moves the data between them, and restarts them when a node dies? In a large share of modern ML stacks, the answer is Ray. This post is the full map.

TL;DR

• Ray is a distributed runtime built on three primitives: tasks (stateless remote functions), actors (stateful remote processes) and objects (immutable values in a shared-memory object store, addressed by ObjectRef).
• Ownership is the key architectural idea: the worker that creates a ref owns its metadata and lineage, which keeps the control plane (GCS) off the hot path. The price is fate sharing: if the owner dies, its objects are lost.
• Placement groups give you gang scheduling: reserve 8 GPUs atomically, optionally on one node for NVLink.
• The libraries are thin layers on the primitives: Ray Data (streaming CPU→GPU pipelines with backpressure), Ray Train (supervises DDP/FSDP worker groups, checkpoints, restarts), Ray Tune (hyperparameter search), Ray Serve (online inference with composition, batching, autoscaling, vLLM-backed LLM serving), RLlib.
• For GPU-to-GPU hot paths, compiled graphs and Ray Direct Transport bypass the CPU object store and use NCCL.
• On Kubernetes, KubeRay gives you RayCluster, RayJob and RayService.
• Interview one-liner: “Ray is a Python-native distributed runtime: tasks, actors and an object store with ownership-based metadata. I use placement groups for gang-scheduled training, Ray Data to stream preprocessing on CPU nodes into GPU inference or training, Ray Train to supervise DDP and recover from checkpoints, and Serve for composable, batched, autoscaled inference, all deployed with KubeRay.”

APIs below reflect Ray 2.x. Newer pieces (Train v2, compiled graphs, RDT, label selectors, Serve LLM) evolve quickly, so confirm signatures against the docs for the version you deploy.

Part I: Ray Core

1. The Mental Model: Three Primitives

Calls are asynchronous. Invoking .remote() schedules work and hands back an ObjectRef immediately. Passing refs into other calls builds a dynamic dataflow graph; Ray resolves each dependency before running the step that needs it. You only block when you call ray.get.

import ray
ray.init()  # local cluster, or connect: ray.init(address="auto")

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

refs = [square.remote(i) for i in range(8)]   # 8 tasks scheduled in parallel
print(ray.get(refs))                           # block once, at the end

2. Architecture: Head Node, GCS, Raylets

flowchart LR
  subgraph Head["Head node"]
    GCS["GCS<br/>cluster metadata"]
    D["Driver<br/>(your script)"]
    R0["Raylet<br/>scheduler + object mgr"]
    S0[("Object store<br/>shared memory")]
    DB["Dashboard / Jobs API"]
  end
  subgraph W1["Worker node"]
    R1["Raylet"]
    S1[("Object store")]
    P1["Worker procs<br/>tasks / actors"]
  end
  D --> R0
  R0 <--> GCS
  R1 <--> GCS
  R0 <-->|"object transfer"| R1
  P1 --- S1
  R1 --- P1

• GCS (Global Control Service) lives on the head node. It stores cluster membership, the actor directory, placement groups, job info and an internal KV store. It is the control plane, not the data plane. For high availability on Kubernetes, back it with external Redis so a head restart doesn’t lose the cluster.
• Raylet runs one per node. It does local scheduling, manages the worker process pool, and runs the object manager that moves objects between nodes.
• Object store (Plasma) is per-node shared memory, sized by default at about 30% of RAM. Workers on the same node read numpy arrays zero-copy.
• Driver is the process that ran ray.init(). A job is a driver plus everything it creates.
• Workers are Python processes the raylet starts. A task borrows a pooled worker; an actor gets a dedicated worker for its whole life.

Key idea: decentralized ownership. The worker that creates an ObjectRef (by calling .remote() or ray.put) is its owner. The owner tracks reference counts, where the value lives, and the lineage needed to rebuild it. This keeps the GCS off the hot path, which is how Ray reaches very high task throughput. The cost: if the owner dies, the objects it owns are lost (fate sharing).

Small return values (under about 100 KB) are inlined directly into the owner’s in-process memory store. Large values go to the shared-memory store on the node that produced them and are pulled to other nodes on demand.

3. Tasks

@ray.remote(num_cpus=2, num_gpus=0.5, max_retries=3, retry_exceptions=[IOError])
def preprocess(shard_path: str) -> dict:
    ...

# Override options per call without redefining the function
ref = preprocess.options(num_gpus=1, name="big-shard").remote("s3://bucket/shard-07")

# Dataflow: pass refs straight into other tasks. Ray waits for the dependency
# and moves the data; the driver never touches it.
@ray.remote
def train_step(batch): ...
out = train_step.remote(ref)

What to remember:

• Resource requests are logical. num_gpus=1 reserves a GPU slot and sets CUDA_VISIBLE_DEVICES. It does not stop your code from touching other GPUs or using more memory. Tasks default to num_cpus=1.
• Fractional GPUs (num_gpus=0.25) pack several small models onto one card. You manage GPU memory yourself.
• Nested tasks are fine: tasks can launch tasks. Ray releases the parent’s CPU while it blocks in ray.get, which prevents deadlock.
• Multiple returns: @ray.remote(num_returns=2) gives two refs, so consumers can depend on just one output.
• Granularity: each task costs on the order of a millisecond of overhead. Batch tiny work so each task runs for at least tens of milliseconds.
• Cancellation: ray.cancel(ref) interrupts a pending or running task; force=True kills the worker.

4. Objects, the Object Store and Ownership

import numpy as np
weights = np.zeros((1024, 1024, 256), dtype=np.float32)   # ~1 GB

w_ref = ray.put(weights)      # store once
refs = [score.remote(w_ref, i) for i in range(100)]  # 100 tasks share one copy

# Anti-pattern: score.remote(weights, i) would serialize 1 GB a hundred times.

• Immutable. Once an object is in the store, it never changes. Updating means producing a new object.
• Zero-copy reads for numpy arrays and Arrow data on the same node: the deserialized array is a read-only view of shared memory. Torch tensors are serialized; convert to numpy when you want zero-copy CPU sharing.
• Top-level refs are resolved, nested refs are not. f.remote(ref) gives f the value. f.remote([ref]) gives f a list containing the ref, which is useful when you want to pass data along without fetching it.
• Reference counting. An object is freed when no ref to it is alive anywhere. Holding refs in a long-lived list on the driver is the classic object-store leak.
• Spilling. When the store fills, objects spill to local disk (or S3 if configured) and are restored on access. You see this as slowdowns, not crashes.
• Lineage reconstruction. If a node holding an object dies, the owner re-executes the task that produced it (if the task is retriable). Objects from ray.put have no lineage and cannot be rebuilt.

In production: use ray memory or the dashboard’s memory view to see which call sites own which objects and why each is still pinned (local ref, captured in a pending task, borrowed by another worker).

5. Actors

@ray.remote(num_gpus=1, max_restarts=3, max_task_retries=2)
class Embedder:
    def __init__(self, model_name):
        import torch
        self.model = load(model_name).cuda().eval()   # load once, keep warm

    def embed(self, batch):
        with torch.inference_mode():
            return self.model(batch).cpu().numpy()

emb = Embedder.remote("clip-vit-l")
refs = [emb.embed.remote(b) for b in batches]   # executed one at a time, in submission order

Concurrency inside an actor

• Default: one method at a time, in order. This is what makes actor state safe without locks.
• Threaded actors: .options(max_concurrency=8) runs methods on a thread pool. You own thread safety.
• Async actors: define methods with async def and Ray runs them on an event loop, with up to 1000 concurrent calls by default. Best for I/O-bound work such as calling a model server.
• Concurrency groups let you give, for example, health checks their own thread pool so they aren’t stuck behind inference calls.

Naming and lifetime

counter = Counter.options(name="global_counter", namespace="metrics",
                          lifetime="detached", get_if_exists=True).remote()

# Any other job in the same namespace can now find it:
c = ray.get_actor("global_counter", namespace="metrics")
ray.kill(c)   # detached actors live until you kill them or the cluster dies

• A normal actor dies when all handles to it go out of scope, or when its owner dies.
• ray.util.ActorPool load-balances work over a fixed set of actors. In practice, Ray Data and Serve do this for you.
• An actor reserves its resources for its whole life. Ten idle GPU actors hold ten GPUs.

6. Waiting, Streaming and Async

Process results as they finish with ray.wait:

pending = [work.remote(x) for x in inputs]
while pending:
    done, pending = ray.wait(pending, num_returns=1, timeout=None)
    handle(ray.get(done[0]))          # no head-of-line blocking on the slowest task

Use the same loop to cap in-flight work (backpressure): only submit a new task when len(pending) drops below a limit. This keeps the object store from filling with results you haven’t consumed.

Generators: stream outputs from one task:

@ray.remote
def read_chunks(path):
    for chunk in open_large_file(path):
        yield chunk                   # each yield becomes its own object

gen = read_chunks.remote("s3://bucket/huge.bin")   # ObjectRefGenerator
for ref in gen:
    consume(ray.get(ref))             # starts before the task finishes

Actor methods can also be generators, which is how token streaming works in Serve.

asyncio interop:

async def main():
    result = await square.remote(4)           # ObjectRefs are awaitable
    results = await asyncio.gather(*[square.remote(i) for i in range(10)])

Never call a blocking ray.get inside an async actor method; await the ref instead.

7. Resources, Scheduling and Placement Groups

Resources

• Built-in: CPU, GPU, memory (logical, for scheduling only), and object store memory.
• Custom resources tag nodes: ray start --resources='{"nvme": 1}', then request with resources={"nvme": 1}.
• Accelerator types: @ray.remote(num_gpus=1, accelerator_type="H100") constrains placement to a GPU model.
• Label selectors (newer releases) match on node labels such as zone or market type, much like k8s node selectors.

Scheduling strategies

Placement groups: gang scheduling. A placement group reserves a list of resource bundles atomically. Either all bundles fit or none are reserved, which prevents the half-scheduled training job that holds 6 of 8 GPUs forever.

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

pg = placement_group([{"GPU": 1, "CPU": 8}] * 8, strategy="STRICT_PACK")  # all 8 on one node (NVLink)
ray.get(pg.ready())

workers = [
    Worker.options(scheduling_strategy=PlacementGroupSchedulingStrategy(
        placement_group=pg, placement_group_bundle_index=i)).remote(rank=i)
    for i in range(8)
]

Ray Train and Ray Serve (for multi-GPU LLM replicas) create placement groups for you. You write them directly when building custom distributed systems.

8. Runtime Environments

A runtime environment describes the dependencies a job, task or actor needs. Ray installs and caches it on each node at run time, so you don’t rebuild the cluster image for every change.

ray.init(runtime_env={
    "working_dir": ".",                         # uploads your local code
    "pip": ["transformers==4.44.2", "peft"],    # or "uv": [...] in recent versions
    "env_vars": {"HF_HOME": "/mnt/cache", "NCCL_DEBUG": "WARN"},
    "excludes": ["*.ckpt", "data/"],
})

# Per actor or task, overriding the job-level env:
Legacy.options(runtime_env={"pip": ["torch==2.1.2"]}).remote()

In production: bake heavy dependencies (CUDA, torch, vLLM) into the container image and keep runtime envs for light, fast-changing pieces. Installing torch at actor start time adds minutes to every cold start and every restart.

9. Fault Tolerance and Memory

• Actor state is your job. Checkpoint to durable storage and reload in __init__. Ray Train does this for training.
• The memory monitor kills workers when a node passes about 95% memory use, preferring retriable tasks and the most recently started work. An OutOfMemoryError in logs usually means you need fewer concurrent tasks (request more memory or num_cpus per task) rather than more retries.
• Exceptions keep their original traceback inside RayTaskError; you can still catch the original type with except ValueError as usual.

Part II: The AI Libraries

10. Ray Data: Streaming Batch Processing

Ray Data is the library for loading, transforming and running inference over large datasets. Its defining feature is streaming execution: blocks flow through a pipeline of operators with bounded memory, so CPU decoding and GPU inference run concurrently and the dataset never has to fit in memory.

flowchart LR
  A["read_parquet<br/>(CPU tasks)"] --> B["map_batches: decode<br/>(CPU tasks)"]
  B --> C["map_batches: model<br/>(GPU actor pool)"]
  C --> D["write_parquet<br/>(CPU tasks)"]

import ray

ds = ray.data.read_parquet("s3://bucket/images/")       # lazy; nothing runs yet

def decode(batch: dict) -> dict:                          # stateless: runs as tasks
    batch["pixels"] = [load_and_resize(b) for b in batch["bytes"]]
    return batch

class Classifier:                                         # stateful: runs as an actor pool
    def __init__(self):
        self.model = load_model().cuda().eval()           # loaded once per actor
    def __call__(self, batch: dict) -> dict:
        batch["label"] = self.model.predict(batch["pixels"])
        return batch

(ds.map_batches(decode, batch_size=256)
   .map_batches(Classifier, concurrency=8, num_gpus=1, batch_size=128)
   .write_parquet("s3://bucket/predictions/"))           # execution starts here

Concepts:

• Blocks: a dataset is a list of Arrow or pandas blocks stored in the object store. Operators process blocks in parallel.
• Lazy plans: transforms build a logical plan. It executes on consumption (write_*, iter_batches, take, materialize).
• Functions vs classes: a function runs as tasks; a callable class runs in an actor pool (the concurrency argument sets its size or an autoscaling (min, max) range). Use classes for anything with expensive setup, such as model weights.
• Backpressure: each operator gets a share of object store memory, so a fast reader can’t flood a slow GPU stage.
• Row ops vs batch ops: map, filter, flat_map work per row; prefer map_batches for vectorized speed.
• All-to-all ops (random_shuffle, sort, groupby, repartition) are expensive barriers. For training, local_shuffle_buffer_size in iter_torch_batches is usually enough.

Feeding training:

for batch in ds.iter_torch_batches(batch_size=512, device="cuda",
                                   local_shuffle_buffer_size=10_000):
    ...

Inside Ray Train, each worker gets its own shard with ray.train.get_dataset_shard("train"). Preprocessing can then run on CPU-only nodes while GPU nodes only train. That is the main reason to reach for Ray Data over a PyTorch DataLoader at scale, and it is the disaggregated answer to the GPU-starvation problem from Module 2.

11. Ray Train: Distributed Training Orchestration

Ray Train does not replace DDP, FSDP or DeepSpeed. It launches and supervises the worker group, sets up the process group (MASTER_ADDR, ranks, NCCL), wires in data shards, collects metrics, manages checkpoints and restarts the group after failures. Your training loop stays standard PyTorch.

import tempfile, torch
import ray.train, ray.train.torch
from ray.train import ScalingConfig, RunConfig, FailureConfig, Checkpoint
from ray.train.torch import TorchTrainer

def train_loop_per_worker(config):
    model = build_model()
    model = ray.train.torch.prepare_model(model)        # moves to device, wraps in DDP
    # FSDP: wrap it yourself; ray.train.torch.get_device() gives the local device
    opt = torch.optim.AdamW(model.parameters(), lr=config["lr"])

    shard = ray.train.get_dataset_shard("train")
    for epoch in range(config["epochs"]):
        for batch in shard.iter_torch_batches(batch_size=config["bs"]):
            loss = step(model, opt, batch)

        with tempfile.TemporaryDirectory() as d:
            if ray.train.get_context().get_world_rank() == 0:
                torch.save(model.module.state_dict(), f"{d}/model.pt")
            ray.train.report({"loss": loss.item(), "epoch": epoch},
                             checkpoint=Checkpoint.from_directory(d))

trainer = TorchTrainer(
    train_loop_per_worker,
    train_loop_config={"lr": 3e-4, "epochs": 10, "bs": 64},
    scaling_config=ScalingConfig(num_workers=16, use_gpu=True,
                                 resources_per_worker={"CPU": 8, "GPU": 1}),
    run_config=RunConfig(name="vit-pretrain", storage_path="s3://bucket/ray-results",
                         failure_config=FailureConfig(max_failures=3)),
    datasets={"train": train_ds},
)
result = trainer.fit()
print(result.metrics, result.checkpoint)

What happens under the hood:

1. A placement group reserves 16 bundles of {CPU: 8, GPU: 1}.
2. 16 actors start; Ray sets up torch.distributed with NCCL between them.
3. Each actor runs train_loop_per_worker.
4. ray.train.report is a sync point: metrics go to the driver and the checkpoint is uploaded to storage_path.
5. If a worker or node dies, the whole group is torn down and restarted from the latest checkpoint, up to max_failures times. Load it with ray.train.get_checkpoint() at the top of your loop.

Know the version: recent Ray releases ship a reworked Train (v2) with cleaner failure handling and elastic scaling. The user-facing loop above is the same; some Tune integration and callback APIs changed. Check which one your cluster runs.

Other trainers follow the same shape: Lightning and Hugging Face Transformers/Accelerate run inside the TorchTrainer loop, plus XGBoostTrainer and LightGBMTrainer for tree models.

12. Ray Tune: Hyperparameter Search

from ray import tune
from ray.tune.schedulers import ASHAScheduler

def objective(config):
    model = build(config)
    for epoch in range(50):
        acc = train_one_epoch(model, lr=config["lr"])
        tune.report({"acc": acc})           # lets the scheduler stop bad trials early

tuner = tune.Tuner(
    tune.with_resources(objective, {"cpu": 4, "gpu": 1}),
    param_space={
        "lr": tune.loguniform(1e-5, 1e-2),
        "wd": tune.choice([0.0, 0.01, 0.1]),
        "layers": tune.randint(2, 8),
    },
    tune_config=tune.TuneConfig(metric="acc", mode="max", num_samples=64,
                                scheduler=ASHAScheduler(max_t=50, grace_period=5),
                                max_concurrent_trials=8),
)
best = tuner.fit().get_best_result()
print(best.config, best.metrics["acc"])

• Trial: one configuration, run as an actor. Trainable: the function you tune.
• Search algorithms pick configs: random and grid by default, or Optuna, HyperOpt, Bayesian optimization.
• Schedulers stop or modify running trials: ASHA (aggressive early stopping, the safe default), Population Based Training (mutates hyperparameters mid-run by copying checkpoints from good trials), HyperBand.
• Experiments are resumable with Tuner.restore(path, trainable).

13. Ray Serve: Online Inference

Serve runs deployments: groups of replica actors behind an HTTP proxy and a router. You write Python classes; Serve handles replica placement, routing, autoscaling, rolling updates and request batching.

from ray import serve
from starlette.requests import Request

@serve.deployment(ray_actor_options={"num_cpus": 1})
class Preprocessor:
    def __call__(self, text: str) -> list[int]:
        return tokenize(text)

@serve.deployment(
    ray_actor_options={"num_gpus": 1},
    autoscaling_config={"min_replicas": 1, "max_replicas": 8,
                        "target_ongoing_requests": 4},
)
class Classifier:
    def __init__(self, pre):
        self.pre = pre                                  # DeploymentHandle
        self.model = load().cuda().eval()

    @serve.batch(max_batch_size=32, batch_wait_timeout_s=0.01)
    async def predict(self, tokens: list[list[int]]) -> list[str]:
        return self.model(tokens)                       # one GPU call for up to 32 requests

    async def __call__(self, req: Request) -> str:
        text = (await req.json())["text"]
        tokens = await self.pre.remote(text)            # composition across deployments
        return await self.predict(tokens)

app = Classifier.bind(Preprocessor.bind())              # build the application graph
serve.run(app, route_prefix="/classify")

Concepts:

• Deployment = class + config. Replica = one actor of that class. Application = a graph of bound deployments with one ingress.
• DeploymentHandle: call other deployments with handle.method.remote(), which returns an awaitable response. This is how you build pipelines and ensembles without HTTP hops.
• Autoscaling is driven by queued plus running requests per replica (target_ongoing_requests), not GPU utilization. With the cluster autoscaler, new replicas can bring up new nodes.
• FastAPI: decorate a deployment with @serve.ingress(fastapi_app) for routing and validation.
• Streaming: return a StreamingResponse from a generator for token streaming.
• Production config: serve build emits a YAML config; deploy with serve deploy or a KubeRay RayService, which supports zero-downtime upgrades.
• Model multiplexing: @serve.multiplexed loads many LoRA adapters or small models across replicas and routes each request to a replica that already has its model.

LLM serving:

from ray.serve.llm import LLMConfig, build_openai_app

llm = LLMConfig(
    model_loading_config={"model_id": "llama-3.1-8b",
                          "model_source": "meta-llama/Llama-3.1-8B-Instruct"},
    deployment_config={"autoscaling_config": {"min_replicas": 1, "max_replicas": 4}},
    engine_kwargs={"tensor_parallel_size": 2, "max_model_len": 8192},
)
serve.run(build_openai_app({"llm_configs": [llm]}))   # OpenAI-compatible /v1/chat/completions

Ray Serve LLM wraps vLLM (see the vLLM Trilogy): each replica is a vLLM engine whose tensor-parallel GPUs come from a placement group. Ray Data has a matching batch path (ray.data.llm) for offline inference.

14. RLlib, Briefly

RLlib is Ray’s reinforcement learning library. Its newer API stack is built from EnvRunners (actors that step environments and collect samples), Learners (GPU actors that compute updates, possibly with DDP) and RLModules (your PyTorch networks). Algorithms include PPO, SAC, DQN, IMPALA and APPO, plus multi-agent and offline RL.

from ray.rllib.algorithms.ppo import PPOConfig
algo = (PPOConfig().environment("CartPole-v1")
        .env_runners(num_env_runners=8)
        .learners(num_learners=2, num_gpus_per_learner=1)).build()
for _ in range(20):
    print(algo.train()["env_runners"]["episode_return_mean"])

Even if you never use RLlib, its layout (CPU samplers feeding GPU learners, weights broadcast back) is the template for RLHF systems such as OpenRLHF and veRL, which are built directly on Ray actors and placement groups.

Part III: Production

15. GPU-Native Paths: Compiled Graphs and Direct Transport

Normal Ray calls move data through the CPU object store and pay RPC and scheduling overhead each call. That is fine for millisecond-scale tasks but too slow for pipeline-parallel inference or RL weight sync, where you want GPU-to-GPU transfers over NCCL and microsecond overhead. Ray has three tools for this. All are newer and evolving, so check the docs for your version.

Compiled graphs (formerly accelerated DAGs)

from ray.dag import InputNode

stage1, stage2 = Stage.remote(0), Stage.remote(1)        # actors on different GPUs
with InputNode() as inp:
    h = stage1.forward.bind(inp)
    h = h.with_tensor_transport("nccl")                  # pass tensors GPU→GPU
    dag = stage2.forward.bind(h)

cdag = dag.experimental_compile()     # set up channels and NCCL communicators once
for x in inputs:
    out = ray.get(cdag.execute(x))    # each run reuses the static plan

You define a static graph of actor method calls once. Ray pre-allocates shared-memory or NCCL channels and runs it repeatedly with overhead in the tens of microseconds. vLLM uses this for pipeline parallelism across nodes.

Ray Direct Transport (RDT). An actor method can be annotated so that the torch.Tensors it returns stay on the GPU, and Ray moves them directly to the consuming actor over NCCL, Gloo or RDMA when needed. The ObjectRef programming model stays the same; only the data path changes. This targets RL weight sync and disaggregated prefill/decode.

@ray.remote(num_gpus=1)
class Learner:
    @ray.method(tensor_transport="nccl")
    def weights(self):
        return self.model.state_dict()     # tensors stay on this GPU until consumed

Collective communication. ray.util.collective provides allreduce, broadcast and send/recv among actors with NCCL or Gloo, for when you build your own distributed algorithm instead of using torch.distributed directly.

16. Clusters, Jobs and KubeRay

Starting a cluster by hand:

ray start --head --port=6379 --dashboard-host=0.0.0.0     # head node
ray start --address=HEAD_IP:6379 --num-gpus=8              # each worker node
ray status                                                 # resources and demand

Jobs: how to run work on a cluster:

ray job submit --address http://HEAD_IP:8265 \
  --runtime-env-json '{"working_dir": "./", "pip": ["wandb"]}' \
  -- python train.py --epochs 10
ray job logs <job_id> --follow

Job submission runs your driver on the head node, so a laptop disconnecting doesn’t kill the job. Ray Client (ray://) exists for interactive use but is not recommended for long-running production work.

Autoscaler

• Scales on pending resource demand (tasks, actors and placement groups that can’t be scheduled), not on CPU utilization.
• Removes nodes after an idle timeout. Nodes holding objects that are still referenced are not idle.
• Configured with worker groups (node types), each with min and max replicas.

KubeRay

apiVersion: ray.io/v1
kind: RayJob
metadata: { name: vit-pretrain }
spec:
  entrypoint: python train.py
  shutdownAfterJobFinishes: true
  rayClusterSpec:
    enableInTreeAutoscaling: true
    headGroupSpec:
      rayStartParams: { num-cpus: "0" }      # keep work off the head node
      template: { spec: { containers: [{ name: ray-head, image: myrepo/ray-ml:2.x-cu12 }] } }
    workerGroupSpecs:
      - groupName: h100
        replicas: 2
        minReplicas: 0
        maxReplicas: 8
        template:
          spec:
            containers:
              - name: ray-worker
                image: myrepo/ray-ml:2.x-cu12
                resources: { limits: { nvidia.com/gpu: 8, cpu: "96", memory: 1Ti } }

In production: set num-cpus: "0" on the head so no tasks or actors land there, give the head enough memory for the GCS and dashboard, mount /dev/shm large enough for the object store, and enable GCS fault tolerance with Redis for RayService.

17. Observability and Debugging

Logs live under /tmp/ray/session_latest/logs on each node: raylet.out, gcs_server.out, and per-worker worker-*.out/err. Worker stdout is also streamed to the driver by default.

Debugging “stuck” jobs: nine times out of ten it’s one of these: requested resources that no node type provides (ray status shows infeasible demand), a placement group that can’t fit, the object store full and spilling, or a deadlock from ray.get inside an actor calling itself.

18. Anti-Patterns That Cost Real Money

19. Cheat Sheet

# --- Core ---
ray.init(address="auto")
f.remote(x)                       # -> ObjectRef
f.options(num_gpus=1).remote(x)
ray.get(refs, timeout=60)
ray.wait(refs, num_returns=k)
ray.put(big)
ray.cancel(ref)
A.remote(); a.method.remote()
ray.get_actor(name, namespace=...)
ray.kill(actor)
ray.cluster_resources(); ray.available_resources()

# --- Options ---
# num_cpus / num_gpus / memory / resources={"custom": 1} / accelerator_type="H100"
# max_retries / retry_exceptions / max_restarts / max_task_retries / max_concurrency
# scheduling_strategy="SPREAD" / name / namespace / lifetime="detached"
# get_if_exists=True / runtime_env={...} / num_returns=2

# --- Data / Train / Tune ---
ray.data.read_parquet(path)
ds.map_batches(Cls, concurrency=4, num_gpus=1, batch_size=64)
ds.iter_torch_batches(batch_size=...)
TorchTrainer(loop, scaling_config=..., run_config=..., datasets=...)
ray.train.report(metrics, checkpoint=...); ray.train.get_checkpoint()
tune.Tuner(f, param_space=..., tune_config=...).fit()

# --- Serve ---
@serve.deployment(...)
@serve.batch(max_batch_size=32)
serve.run(App.bind(Dep.bind()))
await handle.method.remote(x)

ray start --head
ray status
ray list actors
ray memory
ray timeline
ray job submit -- python train.py
serve build app:app -o config.yaml
serve deploy config.yaml

20. Check Yourself

Questions that come up in ML infrastructure interviews and design reviews. Answer out loud, then expand.

Where This Fits

• Data layer: Parquet & Arrow and Lance are what Ray Data reads; Ray Data is the engine that streams them to GPUs.
• Input pipelines: Module 2: Dataloaders covers the single-node version of the problem Ray Data distributes.
• Orchestration: Pipeline Orchestration sits above Ray: Airflow, Kubeflow or Flyte decide when to run; Ray decides where and how the compute runs.
• Serving: the vLLM Trilogy and Life of a Tensor are what runs inside a Ray Serve LLM replica.