{"id":"ray-train","name":"ray-train","summary":"クラスター間の分散トレーニングオーケストレーション。PyTorch/TensorFlow/HuggingFaceをノートパソコンから数千のノードにスケールさせることができます。","body":"# Ray Train - Distributed Training Orchestration\n\n## Quick start\n\nRay Train scales machine learning training from single GPU to multi-node clusters with minimal code changes.\n\n**Installation**:\n```bash\npip install -U \"ray[train]\"\n```\n\n**Basic PyTorch training** (single node):\n\n```python\nimport ray\nfrom ray import train\nfrom ray.train import ScalingConfig\nfrom ray.train.torch import TorchTrainer\nimport torch\nimport torch.nn as nn\n\n# Define training function\ndef train_func(config):\n    # Your normal PyTorch code\n    model = nn.Linear(10, 1)\n    optimizer = torch.optim.SGD(model.parameters(), lr=0.01)\n\n    # Prepare for distributed (Ray handles device placement)\n    model = train.torch.prepare_model(model)\n\n    for epoch in range(10):\n        # Your training loop\n        output = model(torch.randn(32, 10))\n        loss = output.sum()\n        loss.backward()\n        optimizer.step()\n        optimizer.zero_grad()\n\n        # Report metrics (logged automatically)\n        train.report({\"loss\": loss.item(), \"epoch\": epoch})\n\n# Run distributed training\ntrainer = TorchTrainer(\n    train_func,\n    scaling_config=ScalingConfig(\n        num_workers=4,  # 4 GPUs/workers\n        use_gpu=True\n    )\n)\n\nresult = trainer.fit()\nprint(f\"Final loss: {result.metrics['loss']}\")\n```\n\n**That's it!** Ray handles:\n- Distributed coordination\n- GPU allocation\n- Fault tolerance\n- Checkpointing\n- Metric aggregation\n\n## Common workflows\n\n### Workflow 1: Scale existing PyTorch code\n\n**Original single-GPU code**:\n```python\nmodel = MyModel().cuda()\noptimizer = torch.optim.Adam(model.parameters())\n\nfor epoch in range(epochs):\n    for batch in dataloader:\n        loss = model(batch)\n        loss.backward()\n        optimizer.step()\n```\n\n**Ray Train version** (scales to multi-GPU/multi-node):\n```python\nfrom ray.train.torch import TorchTrainer\nfrom ray import train\n\ndef train_func(config):\n    model = MyModel()\n    optimizer = torch.optim.Adam(model.parameters())\n\n    # Prepare for distributed (automatic device placement)\n    model = train.torch.prepare_model(model)\n    dataloader = train.torch.prepare_data_loader(dataloader)\n\n    for epoch in range(epochs):\n        for batch in dataloader:\n            loss = model(batch)\n            loss.backward()\n            optimizer.step()\n\n            # Report metrics\n            train.report({\"loss\": loss.item()})\n\n# Scale to 8 GPUs\ntrainer = TorchTrainer(\n    train_func,\n    scaling_config=ScalingConfig(num_workers=8, use_gpu=True)\n)\ntrainer.fit()\n```\n\n**Benefits**: Same code runs on 1 GPU or 1000 GPUs\n\n### Workflow 2: HuggingFace Transformers integration\n\n```python\nfrom ray.train.huggingface import TransformersTrainer\nfrom transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments\n\ndef train_func(config):\n    # Load model and tokenizer\n    model = AutoModelForCausalLM.from_pretrained(\"gpt2\")\n    tokenizer = AutoTokenizer.from_pretrained(\"gpt2\")\n\n    # Training arguments (HuggingFace API)\n    training_args = TrainingArguments(\n        output_dir=\"./output\",\n        num_train_epochs=3,\n        per_device_train_batch_size=8,\n        learning_rate=2e-5,\n    )\n\n    # Ray automatically handles distributed training\n    from transformers import Trainer\n    trainer = Trainer(\n        model=model,\n        args=training_args,\n        train_dataset=train_dataset,\n    )\n\n    trainer.train()\n\n# Scale to multi-node (2 nodes × 8 GPUs = 16 workers)\ntrainer = TransformersTrainer(\n    train_func,\n    scaling_config=ScalingConfig(\n        num_workers=16,\n        use_gpu=True,\n        resources_per_worker={\"GPU\": 1}\n    )\n)\n\nresult = trainer.fit()\n```\n\n### Workflow 3: Hyperparameter tuning with Ray Tune\n\n```python\nfrom ray import tune\nfrom ray.train.torch import TorchTrainer\nfrom ray.tune.schedulers import ASHAScheduler\n\ndef train_func(config):\n    # Use hyperparameters from config\n    lr = config[\"lr\"]\n    batch_size = config[\"batch_size\"]\n\n    model = MyModel()\n    optimizer = torch.optim.Adam(model.parameters(), lr=lr)\n\n    model = train.torch.prepare_model(model)\n\n    for epoch in range(10):\n        # Training loop\n        loss = train_epoch(model, optimizer, batch_size)\n        train.report({\"loss\": loss, \"epoch\": epoch})\n\n# Define search space\nparam_space = {\n    \"lr\": tune.loguniform(1e-5, 1e-2),\n    \"batch_size\": tune.choice([16, 32, 64, 128])\n}\n\n# Run 20 trials with early stopping\ntuner = tune.Tuner(\n    TorchTrainer(\n        train_func,\n        scaling_config=ScalingConfig(num_workers=4, use_gpu=True)\n    ),\n    param_space=param_space,\n    tune_config=tune.TuneConfig(\n        num_samples=20,\n        scheduler=ASHAScheduler(metric=\"loss\", mode=\"min\")\n    )\n)\n\nresults = tuner.fit()\nbest = results.get_best_result(metric=\"loss\", mode=\"min\")\nprint(f\"Best hyperparameters: {best.config}\")\n```\n\n**Result**: Distributed hyperparameter search across cluster\n\n### Workflow 4: Checkpointing and fault tolerance\n\n```python\nfrom ray import train\nfrom ray.train import Checkpoint\n\ndef train_func(config):\n    model = MyModel()\n    optimizer = torch.optim.Adam(model.parameters())\n\n    # Try to resume from checkpoint\n    checkpoint = train.get_checkpoint()\n    if checkpoint:\n        with checkpoint.as_directory() as checkpoint_dir:\n            state = torch.load(f\"{checkpoint_dir}/model.pt\")\n            model.load_state_dict(state[\"model\"])\n            optimizer.load_state_dict(state[\"optimizer\"])\n            start_epoch = state[\"epoch\"]\n    else:\n        start_epoch = 0\n\n    model = train.torch.prepare_model(model)\n\n    for epoch in range(start_epoch, 100):\n        loss = train_epoch(model, optimizer)\n\n        # Save checkpoint every 10 epochs\n        if epoch % 10 == 0:\n            checkpoint = Checkpoint.from_directory(\n                train.get_context().get_trial_dir()\n            )\n            torch.save({\n                \"model\": model.state_dict(),\n                \"optimizer\": optimizer.state_dict(),\n                \"epoch\": epoch\n            }, checkpoint.path / \"model.pt\")\n\n            train.report({\"loss\": loss}, checkpoint=checkpoint)\n\ntrainer = TorchTrainer(\n    train_func,\n    scaling_config=ScalingConfig(num_workers=8, use_gpu=True)\n)\n\n# Automatically resumes from checkpoint if training fails\nresult = trainer.fit()\n```\n\n### Workflow 5: Multi-node training\n\n```python\nfrom ray.train import ScalingConfig\n\n# Connect to Ray cluster\nray.init(address=\"auto\")  # Or ray.init(\"ray://head-node:10001\")\n\n# Train across 4 nodes × 8 GPUs = 32 workers\ntrainer = TorchTrainer(\n    train_func,\n    scaling_config=ScalingConfig(\n        num_workers=32,\n        use_gpu=True,\n        resources_per_worker={\"GPU\": 1, \"CPU\": 4},\n        placement_strategy=\"SPREAD\"  # Spread across nodes\n    )\n)\n\nresult = trainer.fit()\n```\n\n**Launch Ray cluster**:\n```bash\n# On head node\nray start --head --port=6379\n\n# On worker nodes\nray start --address=<head-node-ip>:6379\n```\n\n## When to use vs alternatives\n\n**Use Ray Train when**:\n- Training across multiple machines (multi-node)\n- Need hyperparameter tuning at scale\n- Want fault tolerance (auto-restart failed workers)\n- Elastic scaling (add/remove nodes during training)\n- Unified framework (same code for PyTorch/TF/HF)\n\n**Key advantages**:\n- **Multi-node orchestration**: Easiest multi-node setup\n- **Ray Tune integration**: Best-in-class hyperparameter tuning\n- **Fault tolerance**: Automatic recovery from failures\n- **Elastic**: Add/remove nodes without restarting\n- **Framework agnostic**: PyTorch, TensorFlow, HuggingFace, XGBoost\n\n**Use alternatives instead**:\n- **Accelerate**: Single-node multi-GPU, simpler\n- **PyTorch Lightning**: High-level abstractions, callbacks\n- **DeepSpeed**: Maximum performance, complex setup\n- **Raw DDP**: Maximum control, minimal overhead\n\n## Common issues\n\n**Issue: Ray cluster not connecting**\n\nCheck ray status:\n```bash\nray status\n\n# Should show:\n# - Nodes: 4\n# - GPUs: 32\n# - Workers: Ready\n```\n\nIf not connected:\n```bash\n# Restart head node\nray stop\nray start --head --port=6379 --dashboard-host=0.0.0.0\n\n# Restart worker nodes\nray stop\nray start --address=<head-ip>:6379\n```\n\n**Issue: Out of memory**\n\nReduce workers or use gradient accumulation:\n```python\nscaling_config=ScalingConfig(\n    num_workers=4,  # Reduce from 8\n    use_gpu=True\n)\n\n# In train_func, accumulate gradients\nfor i, batch in enumerate(dataloader):\n    loss = model(batch) / accumulation_steps\n    loss.backward()\n\n    if (i + 1) % accumulation_steps == 0:\n        optimizer.step()\n        optimizer.zero_grad()\n```\n\n**Issue: Slow training**\n\nCheck if data loading is bottleneck:\n```python\nimport time\n\ndef train_func(config):\n    for epoch in range(epochs):\n        start = time.time()\n        for batch in dataloader:\n            data_time = time.time() - start\n            # Train...\n            start = time.time()\n            print(f\"Data loading: {data_time:.3f}s\")\n```\n\nIf data loading is slow, increase workers:\n```python\ndataloader = DataLoader(dataset, num_workers=8)\n```\n\n## Advanced topics\n\n**Multi-node setup**: See [references/multi-node.md](references/multi-node.md) for Ray cluster deployment on AWS, GCP, Kubernetes, and SLURM.\n\n**Hyperparameter tuning**: See [references/hyperparameter-tuning.md](references/hyperparameter-tuning.md) for Ray Tune integration, search algorithms (Optuna, HyperOpt), and population-based training.\n\n**Custom training loops**: See [references/custom-loops.md](references/custom-loops.md) for advanced Ray Train usage, custom backends, and integration with other frameworks.\n\n## Hardware requirements\n\n- **Single node**: 1+ GPUs (or CPUs)\n- **Multi-node**: 2+ machines with network connectivity\n- **Cloud**: AWS, GCP, Azure (Ray autoscaling)\n- **On-prem**: Kubernetes, SLURM clusters\n\n**Supported accelerators**:\n- NVIDIA GPUs (CUDA)\n- AMD GPUs (ROCm)\n- TPUs (Google Cloud)\n- CPUs\n\n## Resources\n\n- Docs: https://docs.ray.io/en/latest/train/train.html\n- GitHub: https://github.com/ray-project/ray ⭐ 36,000+\n- Version: 2.40.0+\n- Examples: https://docs.ray.io/en/latest/train/examples.html\n- Slack: https://forms.gle/9TSdDYUgxYs8SA9e8\n- Used by: OpenAI, Uber, Spotify, Shopify, Instacart","author":"@Orchestra-Research","ownerProfile":null,"authorContacts":null,"sourceUrl":"https://github.com/Orchestra-Research/AI-Research-SKILLs/tree/main/08-distributed-training/ray-train","license":"MIT","category":"document","lang":"en","tokens":2402,"stars":0,"calls30d":1,"claimed":false,"visibility":"public","origin":"crawler","version":"0.1.0","createdAt":"2026-08-22","updatedAt":"2026-08-22","files":[{"path":"references/multi-node.md","size":13526,"sha256":"79b7f3cb8264ca2d8897b8e6e2e7bdb4d9207e2a2ea6e4b4d18d8fe073afd401"}],"requires":{"mcp":[],"tools":[]},"safety":{"flags":[],"scannedAt":"2026-08-22","hasScripts":false,"networkEndpoints":["docs.ray.io","forms.gle","ray-project.github.io"]}}