Ray

Ray is an open-source unified compute framework that makes it easy to scale AI and Python workloads across distributed clusters with Ray Train, Ray Data, and Ray Serve.

When to Use

• Distributed Python Applications: Scaling compute-heavy Python tasks seamlessly from a single laptop to large cloud clusters.
• Distributed ML Training with Ray Train: Orchestrating multi-node PyTorch, XGBoost, and LightGBM training jobs.
• Production Model Serving with Ray Serve: Serving LLMs and multi-model microservice pipelines with dynamic autoscaling.
• Hyperparameter Optimization with Ray Tune: Running Bayesian, Optuna, and population-based training sweeps across hundreds of workers.

Quick Start

import ray

ray.init()

# Remote distributed task
@ray.remote
def square(x):
    return x * x

# Launch tasks in parallel across cluster
futures = [square.remote(i) for i in range(10)]
results = ray.get(futures)
print("Computed squares:", results)

Core Concepts

Distributed Tasks & Stateful Actors

Scaling functions and classes across cluster workers:

import ray
import time

# Initialize local or connect to remote Ray cluster
ray.init(ignore_reinit_error=True)

# 1. Stateless Remote Function (Distributed Task)
@ray.remote
def process_data_chunk(chunk_id: int) -> dict:
    time.sleep(0.5) # Simulating CPU/IO bound computation
    return {"chunk_id": chunk_id, "processed_items": 1000 * chunk_id}

# Launch tasks concurrently across cluster
futures = [process_data_chunk.remote(i) for i in range(8)]
results = ray.get(futures) # Gather results
print("Processed Chunks:", results)

# 2. Stateful Remote Class (Distributed Actor)
@ray.remote
class GlobalCounter:
    def __init__(self):
        self.count = 0

    def increment(self) -> int:
        self.count += 1
        return self.count

counter_actor = GlobalCounter.remote()
inc_futures = [counter_actor.increment.remote() for _ in range(5)]
print("Final actor count:", ray.get(inc_futures[-1]))

Production Model Serving with Ray Serve

Deploying scalable REST endpoints with dynamic request batching:

from ray import serve
import torch
from transformers import pipeline

@serve.deployment(num_replicas=2, ray_actor_options={"num_cpus": 1, "num_gpus": 0})
class SentimentClassifier:
    def __init__(self):
        self.classifier = pipeline("sentiment-analysis")

    async def __call__(self, request):
        json_data = await request.json()
        text = json_data.get("text", "")
        return self.classifier(text)

app = SentimentClassifier.bind()
# serve.run(app)

Distributed Hyperparameter Tuning with Ray Tune

Running parallel hyperparameter trials:

from ray import tune

def training_function(config):
    # Simulated model training
    for step in range(10):
        intermediate_score = config["alpha"] * step + config["beta"]
        tune.report(loss=1.0 / (intermediate_score + 0.1))

search_space = {
    "alpha": tune.uniform(0.1, 1.0),
    "beta": tune.choice([1, 2, 5])
}

tuner = tune.Tuner(
    training_function,
    param_space=search_space,
    tune_config=tune.TuneConfig(num_samples=10, metric="loss", mode="min")
)
results = tuner.fit()
print("Best hyperparameters:", results.get_best_result().config)

Common Patterns

Stateful Distributed Actors for Model Serving

Problem: Re-loading large model weights into memory on every function invocation.

Solution: Use Ray Actors to hold in-memory state across requests:

@ray.remote(num_gpus=1)
class ModelServer:
    def __init__(self):
        # Load weights once into GPU VRAM
        self.model = load_large_model()

    def predict(self, sample):
        return self.model(sample)

server_actor = ModelServer.remote()
prediction = ray.get(server_actor.predict.remote(sample_data))

Best Practices

Do:

• Check the Ray Dashboard (port 8265) to monitor node CPU/GPU utilization, actor placement, and object store memory.
• Pass large read-only datasets via the Ray plasma object store (ray.put(data)) to prevent repetitive serializations.
• Specify resource requirements explicitly in decorators (@ray.remote(num_cpus=2, num_gpus=1)).
• Use Ray Serve for production multi-model microservices with built-in auto-batching.

Don't:

• Call ray.get() inside remote tasks; pass ObjectRefs directly to other remote tasks to preserve pipelining.
• Create millions of micro-tasks; batch fine-grained tasks together to minimize scheduler overhead.
• Pass large stateful objects (like database connections or open files) inside remote arguments.

Troubleshooting

| Error | Cause | Solution | | :-------------------------------------------------------- | :------------------------------------------------------------------- | :----------------------------------------------------------------------------- | | RaySystemError: Object store full | Tasks producing more data than allocated shared memory plasma store. | Use ray.get() to consume results, or store large arrays in disk storage. | | Actor died unexpectedly | Worker process crashed due to unhandled exception or OOM. | Check logs in Ray dashboard (http://localhost:8265) and increase worker RAM. | | Task failed to schedule: Unsatisfiable resource request | Requested more CPUs/GPUs (num_gpus=2) than available on cluster. | Verify available resources via ray.cluster_resources(). |

References

• Ray Documentation