Ray User Guide

Ray is a unified framework for scaling AI and Python applications from a laptop to a cluster. It pairs a general-purpose distributed runtime (Ray Core) with a set of AI libraries — Data, Train, Tune, Serve, and RLlib — that build on top of it. If your workload is written in Python, you can scale it with Ray without standing up any other infrastructure.

This guide shows you how to install Ray, write your first distributed program, use each of the libraries, drive the ray command-line tool, submit jobs to a cluster, and troubleshoot the problems you are most likely to hit. Deep architecture, the full configuration reference, deployment, and security hardening live in the companion Admin Guide.

Note

Ray's three core abstractions are Tasks (stateless remote functions), Actors (stateful remote worker processes), and Objects (immutable values stored in a distributed object store). Almost everything else in Ray — including all five libraries — is built from these three primitives.

1. Quick Start

Install Ray and run a small program that distributes work across all the cores on your machine.

1.1 Install

pip install -U "ray[default]"

Successfully installed ray-2.x.x ...

Tip

ray[default] includes the dashboard and cluster launcher, which you almost always want during development. The bare pip install ray omits them. See Installation for the full list of install options.

1.2 Your first distributed program

Create quickstart.py:

import ray

# Start a local Ray instance (or connect to an existing one).
ray.init()

# A "task": a stateless function that runs somewhere in the cluster.
@ray.remote
def square(x):
    return x * x

# Launch four tasks in parallel. Each call returns immediately with an
# ObjectRef (a future), not the result.
futures = [square.remote(i) for i in range(4)]

# Block until all results are ready, then fetch them.
print(ray.get(futures))

Run it:

python quickstart.py

[0, 1, 4, 9]

The four square.remote(i) calls are scheduled across your CPU cores and run concurrently. ray.get() blocks until the results are available and returns them in input order.

Important

@ray.remote does not run the function. Calling square.remote(i) schedules it and instantly returns an ObjectRef. The work happens asynchronously; you retrieve the value with ray.get().

1.3 Add a stateful actor

Actors are classes whose instances live as long-running processes in the cluster and keep state between calls:

import ray
ray.init()

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

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

# Create one actor process in the cluster.
counter = Counter.remote()

# Method calls are remote and return ObjectRefs.
results = ray.get([counter.increment.remote() for _ in range(5)])
print(results)

[1, 2, 3, 4, 5]

Because the calls all target the same actor, state (self.n) accumulates.

2. Installation

Ray ships as a single PyPI package with optional extras that pull in the dependencies for each library. Install only what you need.

Core + dashboard

pip install -U "ray[default]"

A specific library

pip install -U "ray[data]"      # Ray Data
pip install -U "ray[train]"     # Ray Train
pip install -U "ray[tune]"      # Ray Tune
pip install -U "ray[serve]"     # Ray Serve
pip install -U "ray[rllib]"     # RLlib
pip install -U "ray[llm]"       # Ray LLM (serving + batch inference)

Everything

pip install -U "ray[all]"

2.1 Install options (extras)

Every extra name below is defined in Ray's setup.py and can be combined, e.g. pip install "ray[data,serve]".

Note

The base dependencies of every install include click, msgpack, protobuf, pyyaml, requests, filelock, jsonschema, watchfiles, and grpcio. Library extras add the heavier dependencies (pyarrow, numpy, pandas, fastapi, gymnasium, and so on).

2.2 Requirements

• Python: a currently supported CPython release (3.9+).
• Operating system: Linux and macOS are fully supported. Windows is supported with some limitations (for example, memray-based profiling is unavailable on Windows).
• GPUs are optional. Ray detects GPUs automatically; use them by requesting num_gpus on tasks and actors (see Section 5.4).

2.3 Verify the installation

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

{'CPU': 8.0, 'node:127.0.0.1': 1.0, 'object_store_memory': 2147483648.0, 'memory': 4294967296.0}

Tip

There is also a hidden self-check, ray sanity-check, that confirms the installation can start and stop a local instance.

3. Core Concepts

These three abstractions are the foundation of everything in Ray.

3.1 Tasks — stateless remote functions

Decorate a function with @ray.remote and call it with .remote():

@ray.remote
def add(a, b):
    return a + b

ref = add.remote(2, 3)   # returns an ObjectRef immediately
print(ray.get(ref))      # 5

Tasks are stateless: each invocation is independent and may run on any worker in the cluster.

3.2 Actors — stateful remote classes

Decorate a class with @ray.remote. Each .remote() constructor call creates a dedicated worker process that holds the instance's state:

@ray.remote
class Accumulator:
    def __init__(self):
        self.total = 0
    def add(self, x):
        self.total += x
        return self.total

acc = Accumulator.remote()
ray.get(acc.add.remote(10))   # 10
ray.get(acc.add.remote(5))    # 15

Tip

Use a task when work is independent and stateless (embarrassingly parallel jobs, map operations). Use an actor when you need to keep state, hold a resource (a model in GPU memory, a database connection), or serialize access to something.

3.3 Objects — the distributed object store

Every .remote() call returns an ObjectRef, a reference to a value that lives in Ray's distributed, shared-memory object store. You can also place a value into the store directly with ray.put():

import numpy as np

big_array = np.zeros(10_000_000)
ref = ray.put(big_array)        # store once

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

# Pass the ref to many tasks; the array is shared, not re-serialized per call.
print(ray.get([total.remote(ref) for _ in range(8)]))

[0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]

Important

Objects in the store are immutable. Sharing a single ObjectRef across many tasks avoids repeatedly copying large arguments — this is the single most effective Ray performance pattern for large data.

3.4 Waiting for results: ray.get and ray.wait

ray.get(refs) blocks until all refs are ready. When you want to process results as they finish, use ray.wait():

refs = [slow_task.remote(i) for i in range(100)]

while refs:
    # Return as soon as one task is done.
    ready, refs = ray.wait(refs, num_returns=1)
    result = ray.get(ready[0])
    handle(result)

Warning

Avoid calling ray.get() inside a loop that also submits work — it serializes your program by blocking on each result. Submit all the tasks first, then ray.get() the list (or drain with ray.wait()).

4. Everyday Usage

4.1 Starting and connecting

ray.init() is the entry point. With no arguments it auto-detects a running cluster (via the RAY_ADDRESS environment variable or a local instance) and otherwise starts a new local instance.

ray.init()                                  # local instance or auto-detect
ray.init(address="auto")                    # attach to local cluster, else error
ray.init(address="ray://1.2.3.4:10001")     # connect to a remote cluster (needs ray[client])
ray.init(num_cpus=4, num_gpus=1)            # cap resources for a local instance
ray.init(ignore_reinit_error=True)          # don't error if already initialized

Common ray.init() arguments:

Always release resources when you are done (especially in notebooks/tests):

ray.shutdown()

4.2 Inspecting the cluster

ray.is_initialized()      # True/False
ray.cluster_resources()   # total CPUs/GPUs/memory across all nodes
ray.available_resources() # currently free resources
ray.nodes()               # list of node records (alive/dead, resources, address)
ray.get_gpu_ids()         # GPU IDs assigned to the current worker

4.3 Chaining tasks

ObjectRefs can be passed directly as arguments to other remote calls. Ray resolves the dependency automatically and only runs the dependent task once its inputs are ready:

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

@ray.remote
def process(data):
    return sum(data)

# No ray.get() needed in between — Ray wires up the dependency.
result_ref = process.remote(load.remote())
print(ray.get(result_ref))   # 4950

4.4 Nested tasks and the driver

Tasks and actors can themselves submit more tasks. A common pattern is a "driver → workers" tree where one task fans out work and aggregates results.

Caution

Do not call ray.get() on a result you are also waiting on from within the same task in a way that blocks a worker indefinitely — this can deadlock when the cluster runs out of workers. Prefer returning ObjectRefs and letting the caller resolve them.

5. Configuring Tasks and Actors

Pass resource and behavior options either to the decorator or, per call, with .options().

5.1 Decorator vs. .options()

# Set defaults on the decorator:
@ray.remote(num_cpus=2, num_gpus=1, max_retries=3)
def train_shard(shard):
    ...

# Override per invocation:
train_shard.options(num_gpus=0, name="cpu_fallback").remote(shard)

5.2 Options reference

All options below are accepted by both @ray.remote(...) and .options(...).

Common (tasks and actors):

Tasks only:

Actors only:

5.3 Named and detached actors

Register an actor by name and retrieve it from anywhere in the cluster:

Counter.options(name="global_counter", lifetime="detached").remote()

# Elsewhere — even in another driver/job:
counter = ray.get_actor("global_counter")
ray.get(counter.increment.remote())

Note

A "detached" actor survives after the script that created it exits. You are responsible for cleaning it up with ray.kill(handle) when it is no longer needed.

5.4 Request CPUs, GPUs, and custom resources

@ray.remote(num_gpus=1)
def infer(batch):
    import torch
    assert torch.cuda.is_available()
    ...

# Custom resources let you pin work to specific hardware:
@ray.remote(resources={"tpu": 1})
def tpu_job():
    ...

Custom resources are advertised when you start a node (ray start --resources) or call ray.init(resources=...).

5.5 Cancelling and killing

ref = long_task.remote()
ray.cancel(ref)              # cancel a task

actor = MyActor.remote()
ray.kill(actor)              # terminate an actor immediately

6. Runtime Environments

A runtime environment specifies the dependencies and working directory a job, task, or actor needs — installed on the fly, without baking a custom container. Pass it to ray.init(), the decorator, or .options().

ray.init(runtime_env={
    "working_dir": "./my_project",          # upload this directory to workers
    "pip": ["requests==2.31.0", "pandas"],  # install these packages
    "env_vars": {"MY_FLAG": "1"},
})

@ray.remote(runtime_env={"pip": ["scipy"]})
def needs_scipy():
    import scipy
    return scipy.__version__

Common runtime_env keys:

Tip

Runtime environments are the recommended way to ship your code and dependencies to a remote cluster during development. For production, bake heavy dependencies into the container image and use runtime_env only for your application code and light packages.

7. The Ray Command-Line Tool

Installing Ray puts the ray command on your PATH. Run ray --help or ray <command> --help for full details. The commands below are grouped by what you use them for.

7.1 Running Ray on the local machine

Start a local multi-node cluster by hand:

# On the head machine:
ray start --head --port=6379

# On each worker machine:
ray start --address='<head-ip>:6379'

Local node IP: <head-ip>
--------------------
Ray runtime started.
--------------------
To connect to this Ray cluster: import ray; ray.init()

Useful ray start flags (run ray start --help for the full ~45):

Warning

--dashboard-host defaults to localhost. Setting it to 0.0.0.0 exposes the dashboard on all network interfaces. Only do this behind a firewall or VPN — the dashboard and Ray ports give code-execution access to the cluster. See the Admin Guide's security section.

7.2 Stopping Ray

ray stop                 # graceful stop
ray stop --force         # SIGKILL everything immediately
ray stop --grace-period 30   # wait up to 30s before force-killing

7.3 Launching cloud/VM clusters

The cluster launcher provisions and manages clusters from a YAML config (for AWS, GCP, Azure, vSphere, or on-prem). All of these take the cluster config file as their first argument.

ray up cluster.yaml --yes
ray submit cluster.yaml train.py --start
ray down cluster.yaml --yes

Tip

ray submit ... --start brings the cluster up first if it isn't already running. Add --tmux or --screen to keep the job running after you disconnect.

7.4 Observability commands (state CLI)

These query the live state of the cluster. They require ray[default].

ray list actors --filter "state=ALIVE" --limit 20
ray list tasks --detail
ray summary tasks
ray logs actor --id <ACTOR_ID>
ray logs job --id <SUBMISSION_ID>

ray list supports --format, --filter, --limit, and --detail.

7.5 Diagnostics and dumps

8. Submitting Jobs

The Ray Jobs API is the recommended way to run a workload on a cluster without holding an interactive connection. A job packages your code and dependencies, runs an entrypoint command on the cluster, and lets you query its status and logs by ID.

8.1 From the command line

# Submit (uploads ./ as the working dir and runs the command on the cluster):
ray job submit --working-dir . -- python my_script.py

Job submission server address: http://127.0.0.1:8265
-------------------------------------------------------
Job 'raysubmit_abc123' submitted successfully
-------------------------------------------------------
Query the logs of the job:   ray job logs raysubmit_abc123
Query the status of the job: ray job status raysubmit_abc123

ray job subcommands:

Key ray job submit options:

Important

The job server is the Ray dashboard address (default port 8265), not the GCS port 6379. Set RAY_ADDRESS=http://<head>:8265 to avoid passing --address every time.

8.2 From Python

from ray.job_submission import JobSubmissionClient

client = JobSubmissionClient("http://127.0.0.1:8265")
job_id = client.submit_job(
    entrypoint="python my_script.py",
    runtime_env={"working_dir": "./", "pip": ["requests"]},
)
print(client.get_job_status(job_id))
print(client.get_job_logs(job_id))

9. The AI Libraries

Each library is a Pythonic API that builds on Ray Core. Install the matching extra (see Section 2.1).

9.1 Ray Data — scalable datasets

Distributed, streaming data loading and transformation for ML.

import ray

ds = ray.data.read_parquet("s3://bucket/data/")
ds = ds.map_batches(lambda batch: preprocess(batch), batch_size=1024)
ds.write_parquet("s3://bucket/out/")

# Stream batches into training:
for batch in ds.iter_batches(batch_size=256):
    train_step(batch)

Use it for: ETL, batch inference, and feeding distributed training. It reads from Parquet, CSV, JSON, images, Lance, Hudi, and many other sources, and executes lazily as a streaming pipeline.

9.2 Ray Train — distributed training

Scale PyTorch, Lightning, Hugging Face Transformers, XGBoost, and more across a cluster.

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

def train_loop_per_worker(config):
    # Standard PyTorch loop; Ray Train handles distribution.
    ...

trainer = TorchTrainer(
    train_loop_per_worker,
    scaling_config=ScalingConfig(num_workers=4, use_gpu=True),
)
result = trainer.fit()
print(result.metrics)

Ray Train manages worker groups, fault tolerance and checkpointing, and integrates with Ray Data for sharded input.

9.3 Ray Tune — hyperparameter tuning

Run and orchestrate large hyperparameter searches.

from ray import tune

def objective(config):
    score = train_model(lr=config["lr"])
    tune.report({"score": score})

tuner = tune.Tuner(
    objective,
    param_space={"lr": tune.grid_search([0.01, 0.001, 0.0001])},
)
results = tuner.fit()
print(results.get_best_result(metric="score", mode="max").config)

Tune provides search algorithms (Optuna, HyperOpt, BayesOpt, Ax, and more) and schedulers (ASHA, PBT) that stop or mutate trials during the search.

The standalone tune CLI is also installed:

9.4 Ray Serve — online serving

Build and scale online inference services and multi-model pipelines from Python.

from ray import serve
from starlette.requests import Request

@serve.deployment(num_replicas=2, ray_actor_options={"num_gpus": 1})
class Model:
    def __init__(self):
        self.model = load_model()
    async def __call__(self, request: Request):
        data = await request.json()
        return self.model.predict(data["input"])

serve.run(Model.bind())

Drive Serve with the serve CLI (also installed as ray serve):

9.5 RLlib — reinforcement learning

A scalable RL library with implementations of PPO, DQN, SAC, IMPALA, APPO, DreamerV3, and more.

from ray.rllib.algorithms.ppo import PPOConfig

config = (
    PPOConfig()
    .environment("CartPole-v1")
    .env_runners(num_env_runners=4)
)
algo = config.build()
for _ in range(5):
    print(algo.train()["env_runners"]["episode_return_mean"])

9.6 Ray LLM — LLM serving and batch inference

ray[llm] adds first-class LLM support: deploy LLMs with Ray Serve (backed by vLLM or SGLang) via ray.serve.llm, and run large-scale batch inference with ray.data.llm.

from ray import serve
from ray.serve.llm import LLMConfig, build_openai_app

llm_config = LLMConfig(
    model_loading_config={"model_id": "meta-llama/Llama-3.1-8B-Instruct"},
)
app = build_openai_app({"llm_configs": [llm_config]})
serve.run(app)   # serves an OpenAI-compatible endpoint

10. The Ray Dashboard

When you start Ray with ray[default], a web dashboard runs on the head node (default http://127.0.0.1:8265).

It shows:

• Cluster view — nodes, CPU/GPU/memory utilization, autoscaler activity.
• Jobs view — submitted jobs, their status, and logs.
• Actors / Tasks — live and historical task and actor state.
• Logs — searchable logs across all nodes.
• Metrics — time-series charts (with Prometheus/Grafana integration).
• Serve / Data / Train — per-library dashboards.

Open it locally:

ray start --head            # then browse to http://127.0.0.1:8265

For a remote cluster, forward the port over SSH (the cluster launcher's ray dashboard <config.yaml> does this for you).

Caution

The dashboard provides job submission and log access. Never expose port 8265 to the public internet. Keep it bound to localhost and reach it through SSH port-forwarding or a private network.

11. Troubleshooting

Format: the symptom, the cause, the fix.

Error: ConnectionError: Could not find any running Ray instance

Cause: You called ray.init(address="auto") (or set RAY_ADDRESS=auto) but no cluster is running. Solution: Start one with ray start --head, or call plain ray.init() to start a local instance. Prevention: Use ray.init() (no address) for local development; reserve address="auto" for code meant to run on an existing cluster.

Error: RuntimeError: Maybe you called ray.init twice by accident?

Cause: ray.init() was called when Ray was already initialized (common in notebooks and pytest). Solution: Pass ray.init(ignore_reinit_error=True), or call ray.shutdown() before re-initializing.

Error: ObjectStoreFullError: Failed to put object ... object store is full

Cause: The object store (shared memory) filled with in-use objects. Solution: Hold fewer ObjectRefs at once, process results with ray.wait() and let refs go out of scope, increase --object-store-memory, or configure object spilling to disk (see Admin Guide). Prevention: Don't accumulate large results in a list; stream them.

Error: GetTimeoutError

Cause: ray.get(refs, timeout=...) elapsed before the objects were ready. Solution: Increase the timeout, or check the dashboard/ray status to see whether tasks are pending for lack of resources.

Symptom: tasks never run / stay PENDING_NODE_ASSIGNMENT

Cause: No node satisfies the requested resources (e.g. you asked for num_gpus=1 on a CPU-only cluster, or a custom resource that isn't advertised). Solution: Run ray status to see available resources; lower the request or start a node with the needed resources (ray start --num-gpus=..., --resources=...).

Symptom: ray job submit cannot connect

Cause: Wrong address — the jobs API uses the dashboard port (8265), not the GCS port (6379). Solution: Use --address http://<head>:8265 or set RAY_ADDRESS=http://<head>:8265.

Symptom: serialization error when calling .remote()

Cause: An argument or return value cannot be pickled (e.g. an open file handle, a thread lock, or a non-serializable client). Solution: Construct non-serializable resources inside the task/actor, not in the driver; pass plain data across the boundary.

Symptom: Ray Serve controller memory grows over time

Cause: A known controller RSS memory growth after repeated autoscaling cycles (community issue [#58815]). Solution: Track the upstream issue; as a mitigation, restart the Serve controller / reduce autoscaling churn until a fixed release is available.

Tip

The dashboard (http://localhost:8265) and the state CLI (ray list tasks --filter "state=FAILED", ray logs ...) are the fastest way to find why a task failed — they surface the worker traceback directly.

12. Frequently Asked Questions

Q: Do I need a cluster to use Ray? No. ray.init() starts a local instance that parallelizes across the cores of a single machine. The same code scales to a multi-node cluster unchanged.

Q: What is the difference between a task and an actor? A task is a stateless function invocation that can run anywhere. An actor is a stateful, long-lived process. Use actors when you must keep state or hold a resource (a loaded model, a connection).

Q: Why does square.remote(2) return an ObjectRef instead of 4? .remote() is asynchronous — it schedules the work and returns a future. Call ray.get(ref) to block for the actual value.

Q: How do I run on more than one machine? Start a head node (ray start --head), join workers (ray start --address=<head:port>), then ray.init(address="auto") from the head — or use the cluster launcher (ray up) on a cloud provider.

Q: How do I ship my code and dependencies to a remote cluster? Use a runtime_env with working_dir and pip/conda/uv. For production, bake heavy dependencies into the container image.

Q: How do I limit how much CPU/GPU a task uses? Set num_cpus/num_gpus (and memory/resources) on @ray.remote or .options(). These are scheduling reservations, not hard OS limits.

Q: Which library do I use for what? Data → loading/transforming/batch inference. Train → distributed model training. Tune → hyperparameter search. Serve → online inference/APIs. RLlib → reinforcement learning. LLM → serving and batch inference of large language models.

Q: How do I connect to a remote cluster from my laptop? ray.init(address="ray://<head>:10001") with ray[client] installed, or submit work as a job with ray job submit. Jobs are preferred for anything long-running.

Q: Is Ray secure to expose on the internet? No. Ray clusters assume a trusted network. Anyone who can reach the Ray ports or dashboard can run code on the cluster. Keep them on a private network. See the Admin Guide.

Q: How do I turn off the anonymous usage stats? Run ray disable-usage-stats, pass ray start --disable-usage-stats, or set the environment variable before starting Ray.

Q: How do I see why my job failed? Open the dashboard's Jobs view, or use ray job logs <id>, ray list tasks --filter "state=FAILED" --detail, and ray logs ....