Ray - Distributed Computing Framework

# Full ML stack with dashboard and cluster launcher
pip install -U "ray[data,train,tune,serve]"

# For reinforcement learning specifically (includes RLlib)
pip install -U "ray[rllib]"

# Verify installation
ray --version

# Ray with dashboard and cluster launcher
pip install -U "ray[default]"

# Minimal Ray (no dashboard, no cluster launcher)
pip install -U "ray"

# Verify installation
ray --version

import ray

# Initialize Ray (optional - happens automatically on first use)
ray.init()

# Simple distributed task
@ray.remote
def add(x, y):
    return x + y

# Submit tasks - returns ObjectRef futures
futures = [add.remote(i, i + 1) for i in range(10)]

# Get results back
results = ray.get(futures)
print(results)
# Output: [1, 3, 5, 7, 9, 11, 13, 15, 17, 19]

# Shutdown Ray
ray.shutdown()

import ray

ray.init()

# Define an actor class
@ray.remote
class Counter:
    def __init__(self):
        self.value = 0
    
    def increment(self):
        self.value += 1
        return self.value
    
    def get_value(self):
        return self.value

# Create actor instance
counter = Counter.remote()

# Call actor methods
result = counter.get_value.remote()
print(ray.get(result))  # Output: 0

result = counter.increment.remote()
print(ray.get(result))  # Output: 1

result = counter.increment.remote()
print(ray.get(result))  # Output: 2

ray.shutdown()

import ray
from ray import data

ray.init()

# Create a dataset from various sources
# From a list
ds = data.from_items([{"text": "hello"}, {"text": "world"}])

# From JSON files
ds = data.read_json("/path/to/json/*.json")

# From CSV files
ds = data.read_csv("/path/to/csv/*.csv")

# Parquet files
ds = data.read_parquet("/path/to/parquet/")

# Transformations
ds = ds.map(lambda x: {"text_upper": x["text"].upper()})

# Batch transformations (more efficient)
def add_prefix(batch):
    batch["prefixed"] = ["item_" + t for t in batch["text"]]
    return batch

ds = ds.map_batches(add_prefix)

# Filter
ds = ds.filter(lambda x: len(x["text"]) > 3)

# Sort
ds = ds.sort("text")

# Aggregate summary statistics
ds.summary()

# Show sample
ds.show(5)

# Materialize as list
results = ds.to_pandas()

ray.shutdown()

import ray
from ray import train
from ray.train import ScalingConfig
import torch
from torchvision.models import resnet18

ray.init()

# Define training function
def train_func(config):
    model = resnet18(num_classes=10)
    
    # Get training configuration from Ray context
    ctx = train.get_context()
    distributed_config = ctx.distributed_config
    
    # Training loop
    for epoch in range(10):
        for batch in dataloader:
            # Training logic here
            pass
    
    # Save checkpoint and report metrics
    train.report(
        {"loss": 0.5, "accuracy": 0.9},
        checkpoint=Checkpoint.from_dict({"model": model.state_dict()})
    )

# Run distributed training
results = train.fit(
    train_loop_per_worker=train_func,
    scaling_config=ScalingConfig(
        num_workers=4,      # Number of training workers
        use_gpu=True        # Use GPU if available
    )
)

print(f"Best checkpoint: {results.checkpoints}")

ray.shutdown()

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

ray.init()

# Define training function with config
def train_model(config):
    for i in range(10):
        # Simulated training
        current_metric = config["lr"] * (i + 1)
        tune.report(mean_loss=current_metric)

# Define search space
search_space = {
    "lr": tune.uniform(0.001, 0.01),
    "batch_size": tune.choice([32, 64, 128]),
    "num_layers": tune.randint(2, 10)
}

# Run tuning experiment with new API
tuner = tune.Tuner(
    train_model,
    param_space=search_space,
    tune_config=tune.TuneConfig(
        num_samples=5,
        metric="mean_loss",
        mode="min",
        scheduler=ASHAScheduler(
            metric="mean_loss",
            mode="min",
            max_t=10,
            grace_period=2,
            reduction_factor=3
        )
    ),
    run_config=ray.train.RunConfig(
        name="hyperparameter_tuning",
        storage_path="/tmp/ray_results"
    )
)

results = tuner.fit()

# Get best result
best_result = results.get_best_result(
    metric="mean_loss",
    mode="min"
)

print(f"Best config: {best_result.config}")
print(f"Best loss: {best_result.metrics['mean_loss']}")

ray.shutdown()

from ray import serve
from transformers import pipeline

# Define the service
@serve.deployment
class TextGenerator:
    def __init__(self):
        self.pipeline = pipeline("text-generation", model="gpt2")
    
    async def __call__(self, text):
        result = self.pipeline(text, max_length=50)
        return {"generated": result[0]["generated_text"]}

# Deploy with configuration
text_gen = TextGenerator.options(
    num_replicas=2,        # Number of replicas
    max_concurrent_queries=100
)

# Deploy the service
depl = text_gen.deploy()

# Run inference
response = depl.remote("The weather today")
print(ray.get(response))

# Alternative: serve.run() for simple deployment
import ray
ray.init()
serve.run(
    text_gen,
    host="0.0.0.0",
    port=8000
)

from ray import tune
from ray.rllib.algorithms import ppo

ray.init()

# Build algorithm config
alg_config = ppo.PpoConfig()
alg_config = alg_config.environment(
    env_id="CartPole-v1"  # Gym environment ID
)
alg_config = alg_config.algorithms(
    gamma=0.99,
    lr=0.001,
    vf_loss_coeff=1,
    entropy_coeff=0.01
)
alg_config = alg_config.resources(
    num_workers=4,
    num_envs_per_worker=4,
    log_level="WARNING"
)

# Build algorithm with config
algo = alg_config.build()

# Train for 10 iterations
for _ in range(10):
    result = algo.train()
    print(f"Reward: {result['episode_reward_mean']}")

# Save the model
checkpoint = algo.save("/tmp/my-checkpoint")
print(f"Saved checkpoint at: {checkpoint}")

ray.shutdown()

# Start Ray with dashboard
ray start --head --port=6379 --dashboard-host=0.0.0.0

# Start worker node
ray start --address=<master-node-address>:6379

# Check cluster status
ray status

# Stop the cluster
ray stop

# View logs
ray logs

# cluster.yaml
cluster_name: my-ray-cluster

# The minimum and max number of workers to scale between.
min_workers: 2
max_workers: 10

# Ray head node settings
head_node:
    MachineType: g4dn.xlarge
    ImageId: ami-0abcdef123456
    DiskSize: 100

# Ray worker node settings (worker types)
worker_node_types:
  - MachineType: g4dn.xlarge
    ImageId: ami-0abcdef123456
    DiskSize: 100
    Resources: {}
    Max_workers: 8

# Setup commands for all nodes
setup_commands:
  - pip install ray[tune,serve,data] --upgrade

# Resources
available_resources:
  CPU: 4
  GPU: 1

# Head node setup commands
head_setup_commands:
  - pip install torch torchvision

# Worker setup commands
worker_setup_commands:
  - pip install torch torchvision
  - echo "Worker node setup complete"

# Submit cluster on AWS
ray up cluster.yaml --no-reuse

# Submit on Google Cloud
ray up cluster.yaml --provider gcp --no-reuse

# Submit on Azure
ray up cluster.yaml --provider azure --no-reuse

# Tear cluster down
ray down cluster.yaml --yes

# SSH into the cluster
ray attach cluster.yaml

# Check cluster status
ray status

# Sync files to cluster
ray sync cluster.yaml /local/path /remote/path

# Start Ray with dashboard
ray start --head --dashboard-host=0.0.0.0 --dashboard-port=8265

# Open browser to http://localhost:8265

# Check cluster info
ray status

# Attach to cluster for debugging
ray attach cluster.yaml

# Sync files and run commands
ray sync cluster.yaml /local/path
ray exec cluster.yaml -- "pip install my-package"

import ray

# Initialize with custom options
ray.init(
    # Cluster settings
    address="auto",  # or "localhost:6379"
    
    # Resource limits
    num_cpus=4,
    num_gpus=2,
    
    # Memory
    memory=8e9,  # 8GB
    
    # Object store
    object_store_memory=2e9,  # 2GB
    
    # Dashboard
    include_dashboard=True,
    dashboard_host="0.0.0.0",
    dashboard_port=8265,
    
    # Logging
    log_to_driver=True,
    ignore_reinit_error=True,
    
    # Namespace
    namespace="default"
)

# Access cluster resources
resources = ray.available_resources()
print(resources)  # {'CPU': 4, 'GPU': 2, 'memory': 8000000000}