ray-distributed-computing

name	ray-distributed-computing
description	Expert guidance for building distributed Python applications with Ray. Use when implementing parallel processing, distributed actors, remote tasks, GPU workloads, or scaling Python code across clusters. Covers Ray Core primitives, design patterns, anti-patterns, and performance optimization.

Ray Distributed Computing Skill

Overview

Ray is a unified framework for scaling AI and Python applications. It provides a core distributed runtime with three key primitives: Tasks (stateless functions), Actors (stateful workers), and Objects (distributed data).

When to Use This Skill

• Parallelizing Python functions across multiple cores or machines
• Building stateful distributed workers (actors)
• Processing large datasets in parallel
• Running GPU-accelerated workloads
• Implementing machine learning pipelines with distributed compute
• Scaling existing Python code without major rewrites

Core Concepts

1. Ray Initialization

import ray

# Initialize Ray (auto-detects resources)
ray.init()

# Or with explicit configuration
ray.init(
    num_cpus=16,              # Number of CPUs to use
    num_gpus=1,               # Number of GPUs to use
    object_store_memory=8 * 1024**3,  # Object store size in bytes
)

2. Remote Tasks (@ray.remote functions)

Stateless functions that run in parallel across the cluster.

@ray.remote
def process_document(doc_path: str) -> dict:
    """Process a single document - runs in parallel."""
    # Your processing logic here
    return {"path": doc_path, "result": "processed"}

# Launch parallel tasks
futures = [process_document.remote(path) for path in document_paths]

# Retrieve results
results = ray.get(futures)

Resource specification:

@ray.remote(num_cpus=2, num_gpus=0.5, memory=1024*1024*1024)
def gpu_intensive_task(data):
    # Uses 2 CPUs and half a GPU
    pass

3. Actors (@ray.remote classes)

Stateful workers that maintain state between method calls.

@ray.remote
class EmbeddingService:
    def __init__(self, model_name: str):
        from sentence_transformers import SentenceTransformer
        self.model = SentenceTransformer(model_name)
    
    def embed(self, texts: list[str]) -> list[list[float]]:
        return self.model.encode(texts).tolist()

# Create actor instance
embedder = EmbeddingService.remote("all-MiniLM-L6-v2")

# Call methods (returns futures)
embeddings = ray.get(embedder.embed.remote(["Hello world"]))

GPU Actor:

@ray.remote(num_gpus=1)
class GPUModel:
    def __init__(self):
        import torch
        self.device = torch.device("cuda")
        self.model = self.load_model()

4. Objects (ray.put / ray.get)

Efficiently share data across tasks and actors.

import numpy as np

# Put large data in object store (returns ObjectRef)
large_array = np.ones((10000, 10000))
data_ref = ray.put(large_array)

# Pass reference to multiple tasks (zero-copy)
futures = [process.remote(data_ref) for _ in range(10)]

Design Patterns

Pattern 1: Nested Tasks for Parallelism

@ray.remote
def process_batch(items: list) -> list:
    # Spawn sub-tasks for each item
    futures = [process_item.remote(item) for item in items]
    return ray.get(futures)

Pattern 2: Actor Pool for Stateful Processing

from ray.util import ActorPool

# Create pool of actors
actors = [MyActor.remote() for _ in range(4)]
pool = ActorPool(actors)

# Process items using the pool
results = list(pool.map(lambda a, item: a.process.remote(item), items))

Pattern 3: Pipeline Processing

@ray.remote
def stage1(data):
    return preprocess(data)

@ray.remote  
def stage2(data_ref):
    data = ray.get(data_ref)
    return transform(data)

# Chain stages (stage2 starts when stage1 finishes)
ref1 = stage1.remote(raw_data)
ref2 = stage2.remote(ref1)  # Pass ObjectRef directly
result = ray.get(ref2)

Pattern 4: Limit Concurrent Tasks with ray.wait

# Process many items with limited concurrency
MAX_PENDING = 10
pending = []

for item in items:
    if len(pending) >= MAX_PENDING:
        # Wait for at least one to complete
        ready, pending = ray.wait(pending, num_returns=1)
        results.extend(ray.get(ready))
    pending.append(process.remote(item))

# Get remaining results
results.extend(ray.get(pending))

Anti-Patterns to Avoid

❌ Anti-Pattern 1: ray.get in a Loop

# BAD - serializes execution
for item in items:
    result = ray.get(process.remote(item))  # Blocks each iteration!

# GOOD - parallelize then collect
futures = [process.remote(item) for item in items]
results = ray.get(futures)

❌ Anti-Pattern 2: Passing Large Objects by Value

# BAD - copies data for each task
large_data = load_huge_dataset()
futures = [process.remote(large_data) for _ in range(10)]  # Copies 10x!

# GOOD - use ray.put for zero-copy
data_ref = ray.put(large_data)
futures = [process.remote(data_ref) for _ in range(10)]

❌ Anti-Pattern 3: Too Fine-Grained Tasks

# BAD - overhead exceeds computation
futures = [add_one.remote(x) for x in range(1000000)]  # Tiny tasks!

# GOOD - batch work into larger chunks
def process_batch(batch):
    return [x + 1 for x in batch]

batches = [items[i:i+1000] for i in range(0, len(items), 1000)]
futures = [process_batch.remote(batch) for batch in batches]

❌ Anti-Pattern 4: Global Variables for State

# BAD - global state doesn't sync across workers
global_counter = 0

@ray.remote
def increment():
    global global_counter
    global_counter += 1  # Each worker has its own copy!

# GOOD - use an Actor for shared state
@ray.remote
class Counter:
    def __init__(self):
        self.count = 0
    def increment(self):
        self.count += 1
        return self.count

Performance Tips

1. Batch small operations - Group tiny tasks into larger batches
2. Use ray.put for large data - Avoid copying data repeatedly
3. Specify resources accurately - Help Ray schedule efficiently
4. Use async actors for I/O-bound work - Concurrent method execution
5. Monitor with Ray Dashboard - http://localhost:8265

Ray Dashboard

Access at http://localhost:8265 to monitor:

• Cluster resource utilization
• Task and actor execution timeline
• Memory usage and object store
• Logs and errors

Installation

pip install "ray[default]>=2.46.0"

References

• Ray Documentation
• Ray Core Walkthrough
• Design Patterns
• GitHub Repository