企业官网建设流程全解析

• 人工智能
• 分布式训练
• 强化学习
• 任务调度
• 模型推理服务

Ray Train 是 Ray 生态中面向大规模分布式训练与微调的可扩展机器学习库，它以极简的 API 将模型训练代码从单机无缝扩展到云上多机集群，并抽象掉分布式计算的全部复杂性。阅读本文后，你将掌握 Ray Train 的四大核心概念（训练函数、Worker、ScalingConfig、Trainer）、一条命令完成安装的方法，以及 PyTorch、PyTorch Lightning、Hugging Face Transformers、XGBoost、JAX 五大主流框架从单机脚本改造为分布式训练任务的完整实战路径。

Ray Train 是什么：面向大模型与大数据的分布式训练方案

无论你面对的是大模型还是大数据集，Ray Train 都致力于成为最简单的分布式训练解决方案。它将训练代码从单机扩展到云端集群中的多台机器，并自动处理分布式计算带来的复杂性，包括进程组初始化、设备放置、数据分片、指标上报与检查点持久化等。

Ray Train 对主流机器学习框架提供了开箱即用的支持，覆盖 PyTorch 生态与更多框架两大阵营：

其中每个框架都有对应的独立入门指南，本文后续章节将逐一展开。

安装 Ray Train

安装 Ray Train 只需一条命令：

$ pip install -U "ray[train]"

这条命令通过 Ray 的 Python 包安装机制完成。从仓库中的 python/setup.py 可以看出，train这一 extras 依赖组被定义为 Tune extras 的超集（setup_spec.extras["train"] = list(setup_spec.extras["tune"])），也就是说ray[train]会一并安装 Ray Tune（超参数调优）所需依赖——这正是 Ray Train 能与 Ray Tune 无缝集成做超参搜索的基础。

需要说明的是，ray[train]安装的是 Ray 核心训练框架本身，各框架的第三方库需要按需额外安装。例如在 Hugging Face Transformers 入门指南中，完整的安装命令是：

pip install "ray[train]" torch "transformers[torch]" datasets evaluate numpy scikit-learn

更多关于 Ray 及各库的安装细节，可参见仓库中的 安装文档（本文按 doc/source/train/train.md 的指引指向官方安装章节）。

核心概念：训练函数、Worker、ScalingConfig 与 Trainer

要高效使用 Ray Train，首先需要理解 overview.rst 中定义的四个核心概念：

1. 训练函数（Training Function）：一个包含完整模型训练循环逻辑的 Python 函数；
2. Worker：运行训练函数的进程；
3. 伸缩配置（ScalingConfig）：对 Worker 数量与计算资源（如 CPU 或 GPU）的配置；
4. Trainer：将训练函数、Worker 与伸缩配置三者绑定、用于执行分布式训练任务的 Python 类。

训练函数（train_func）

训练函数是用户自定义的 Python 函数，包含端到端的模型训练循环逻辑。当启动分布式训练任务时，每个 Worker 都会执行这个函数。Ray Train 文档的通用约定是：train_func为用户定义的训练函数，它被传入 Trainer 的train_loop_per_worker参数：

def train_func(): """在每个分布式 Worker 进程上运行的用户自定义训练函数。 该函数通常包含：加载模型、加载数据集、训练模型、 保存检查点以及记录指标等逻辑。 """ ...

Worker

Ray Train 将模型训练计算分发到集群中独立的 Worker 进程上执行，每个 Worker 就是一个运行train_func的进程。Worker 的数量决定了训练任务的并行度，并通过ray.train.ScalingConfig配置。

ScalingConfig：定义训练规模

ScalingConfig是定义训练任务规模的机制，最基础的两个参数用于控制 Worker 并行度与计算资源：

• num_workers：分布式训练任务启动的 Worker 数量；
• use_gpu：每个 Worker 是否使用 GPU。

from ray.train import ScalingConfig # 单 Worker + CPU scaling_config = ScalingConfig(num_workers=1, use_gpu=False) # 单 Worker + GPU scaling_config = ScalingConfig(num_workers=1, use_gpu=True) # 多 Worker，每个 Worker 使用一块 GPU scaling_config = ScalingConfig(num_workers=4, use_gpu=True)

从仓库源码 python/ray/train/v2/api/config.py 可以看到，ScalingConfig的字段远比表面丰富，除num_workers、use_gpu外还包括：

• resources_per_worker：每个 Worker 所需的资源字典（例如{"CPU": 4}或{"TPU": 4}）；
• accelerator_type（实验性）：指定加速器类型（如 TPU 代次），确保任务调度到目标资源上；
• use_tpu（实验性）：Ray 2.49.0 新增字段，置 True 时训练在 TPU 上执行；
• topology（实验性）：定义 TPU 芯片的物理排布（如"4x4"），多主机训练时必填；
• 弹性训练支持：num_workers可传(min, max)二元组实现弹性伸缩，elastic_resize_monitor_interval_s控制扩缩容监控间隔（默认 60 秒）；
• 本地模式：当num_workers=0时，训练函数将在同一进程内直接运行（见 config.py 中的本地模式日志逻辑），方便调试。

Trainer：绑定一切并启动训练

Trainer 将上述三个概念绑定在一起，用于启动分布式训练任务。Ray Train 为不同框架提供了对应的 Trainer 类（详见 API 参考）。调用fit()方法执行训练任务，其内部流程为：

1. 按scaling_config的定义启动 Worker；
2. 在所有 Worker 上搭建框架对应的分布式环境（如 PyTorch 的进程组）；
3. 在所有 Worker 上运行train_func。

from ray.train.torch import TorchTrainer trainer = TorchTrainer(train_func, scaling_config=scaling_config) trainer.fit()

在源码层面，所有框架 Trainer 的共同基类是 python/ray/train/base_trainer.py 中的BaseTrainer，其构造参数正是train_loop_per_worker、scaling_config、run_config与datasets四件套；fit()方法（base_trainer.py）在内部将 Trainer 转换为 Tune 可执行的 Trainable，并通过 Ray Tune 的 Tuner 完成实际运行调度，返回一个Result对象。

快速上手：五大框架入门指南

主文档 train.md 提供了 Overview、PyTorch、PyTorch Lightning、Hugging Face Transformers、JAX 五张入门卡片，并附 XGBoost 指南。下面逐一深入。

PyTorch：改造现有训练脚本

入门指南 getting-started-pytorch.rst 演示了如何将现有 PyTorch 脚本改造为 Ray Train 分布式训练，最终代码结构如下：

from ray.train.torch import TorchTrainer from ray.train import ScalingConfig def train_func(): # 你的 PyTorch 训练代码 ... scaling_config = ScalingConfig(num_workers=2, use_gpu=True) trainer = TorchTrainer(train_func, scaling_config=scaling_config) result = trainer.fit()

关键改动集中在训练函数内部，核心是三个工具函数：

1. 准备模型：ray.train.torch.prepare_model(model)

该函数完成两件事：把模型移动到正确的设备（CPU/GPU）上，并包上DistributedDataParallel。它替代了手写的model.to(device)与DistributedDataParallel(model, device_ids=[device_id])：

-from torch.nn.parallel import DistributedDataParallel +import ray.train.torch def train_func(): ... # 创建模型。 model = ... # 设置分布式训练与设备放置。 - device_id = ... # 自行获取正确设备的逻辑。 - model = model.to(device_id or "cpu") - model = DistributedDataParallel(model, device_ids=[device_id]) + model = ray.train.torch.prepare_model(model) ...

2. 准备数据加载器：ray.train.torch.prepare_data_loader(data_loader)

该函数为 DataLoader 添加DistributedSampler并将 batch 移动到正确设备。注意DataLoader的batch_size是每个 Worker 的批大小，全局批大小按global_batch_size = worker_batch_size * ray.train.get_context().get_world_size()换算。若你已手动配置了DistributedSampler，该函数不会重复添加；但DistributedSampler不适用于包装IterableDataset的 DataLoader，此类场景建议改用 Ray Data 做高性能流式数据摄入。

3. 上报指标与检查点：ray.train.report()

metrics = {"loss": loss.item()} # 训练/验证指标 # 从目录构建 Ray Train 检查点 checkpoint = ray.train.Checkpoint.from_directory(temp_checkpoint_dir) # Ray Train 会自动把检查点保存到持久化存储， # 因此本地临时目录 temp_checkpoint_dir 之后可以安全清理。 ray.train.report(metrics=metrics, checkpoint=checkpoint)

下面是以 FashionMNIST + ResNet18 为例的完整可运行版本（含训练完成后加载模型）：

import os import tempfile import torch from torch.nn import CrossEntropyLoss from torch.optim import Adam from torch.utils.data import DataLoader from torchvision.models import resnet18 from torchvision.datasets import FashionMNIST from torchvision.transforms import ToTensor, Normalize, Compose import ray.train.torch def train_func(): # 模型、损失、优化器 model = resnet18(num_classes=10) model.conv1 = torch.nn.Conv2d( 1, 64, kernel_size=(7, 7), stride=(2, 2), padding=(3, 3), bias=False ) model = ray.train.torch.prepare_model(model) # 移动设备 + DDP 包装 criterion = CrossEntropyLoss() optimizer = Adam(model.parameters(), lr=0.001) # 数据 transform = Compose([ToTensor(), Normalize((0.28604,), (0.32025,))]) data_dir = os.path.join(tempfile.gettempdir(), "data") train_data = FashionMNIST(root=data_dir, train=True, download=True, transform=transform) train_loader = DataLoader(train_data, batch_size=128, shuffle=True) train_loader = ray.train.torch.prepare_data_loader(train_loader) # 训练 for epoch in range(10): if ray.train.get_context().get_world_size() > 1: train_loader.sampler.set_epoch(epoch) for images, labels in train_loader: # 设备放置由 prepare_data_loader 完成 outputs = model(images) loss = criterion(outputs, labels) optimizer.zero_grad() loss.backward() optimizer.step() # 上报指标与检查点 metrics = {"loss": loss.item(), "epoch": epoch} with tempfile.TemporaryDirectory() as temp_checkpoint_dir: torch.save( model.module.state_dict(), os.path.join(temp_checkpoint_dir, "model.pt") ) ray.train.report( metrics, checkpoint=ray.train.Checkpoint.from_directory(temp_checkpoint_dir), ) if ray.train.get_context().get_world_rank() == 0: print(metrics) scaling_config = ray.train.ScalingConfig(num_workers=2, use_gpu=True) trainer = ray.train.torch.TorchTrainer( train_func, scaling_config=scaling_config, # 多节点集群中，应在此配置所有 Worker 节点都能访问的持久化存储， # 例如 run_config=ray.train.RunConfig(storage_path="s3://...") ) result = trainer.fit() # 加载训练好的模型 with result.checkpoint.as_directory() as checkpoint_dir: model_state_dict = torch.load(os.path.join(checkpoint_dir, "model.pt")) model = resnet18(num_classes=10) model.conv1 = torch.nn.Conv2d( 1, 64, kernel_size=(7, 7), stride=(2, 2), padding=(3, 3), bias=False ) model.load_state_dict(model_state_dict)

PyTorch Lightning：利用 Ray 专属策略与回调

入门指南 getting-started-pytorch-lightning.rst 指出：Ray Train 已在每个 Worker 上搭好分布式进程组，你只需对 Lightning Trainer 做少量改动。

核心改动一：使用 Ray 提供的分布式策略。Ray Train 为 Lightning 提供三个保留基类同名参数的子类策略，内部会配置根设备与分布式采样器参数：

• ray.train.lightning.RayDDPStrategy
• ray.train.lightning.RayFSDPStrategy
• ray.train.lightning.RayDeepSpeedStrategy

核心改动二：配置 Ray 集群环境插件。ray.train.lightning.RayLightningEnvironment负责配置 Worker 的 local/global/node rank 与 world size：

trainer = pl.Trainer( - devices=[0, 1, 2, 3], - strategy=DDPStrategy(), - plugins=[LightningEnvironment()], + devices="auto", + accelerator="auto", + strategy=ray.train.lightning.RayDDPStrategy(), + plugins=[ray.train.lightning.RayLightningEnvironment()] ) + trainer = ray.train.lightning.prepare_trainer(trainer)

注意：devices与accelerator应始终使用"auto"，因为 Ray TorchTrainer 已为你正确配置了CUDA_VISIBLE_DEVICES。

核心改动三：添加RayTrainReportCallback回调，用于持久化检查点与上报指标：

trainer = pl.Trainer( max_epochs=10, devices="auto", accelerator="auto", strategy=ray.train.lightning.RayDDPStrategy(), plugins=[ray.train.lightning.RayLightningEnvironment()], callbacks=[ray.train.lightning.RayTrainReportCallback()], enable_checkpointing=False, # 可选：关闭默认检查点行为 ) trainer = ray.train.lightning.prepare_trainer(trainer) trainer.fit(model, train_dataloaders=train_dataloader)

RayTrainReportCallback还支持两个进阶选项：checkpoint_upload_mode=CheckpointUploadMode.ASYNC将检查点上传卸载到 Ray Train 管理的后台线程，避免阻塞 Lightning 训练循环；validation参数则启动异步 Ray 任务校验检查点，替代在训练 Worker 中同步执行validation_step（注意此模式与 Lightning 的AsyncCheckpointIO插件不兼容，因为 Ray Train 需要控制上传线程以等待其完成后提交检查点）。上报指标与检查点是支持容错训练与超参数优化的前提。

版本与迁移说明：Ray Train 官方测试覆盖pytorch_lightning的1.6.5与2.1.2版本，建议使用pytorch_lightning>=1.6.5；使用 Lightning 2.x 时请改用lightning.pytorch.xxx导入路径。API 层面，Ray 2.4 曾引入LightningTrainer+LightningConfigBuilder的黑盒式 API，Ray 2.7 起统一为透明的TorchTrainer方案，让你直接控制原生 Lightning 代码（两种写法的完整对照见指南中的迁移章节）。

Hugging Face Transformers：文本模型分布式训练

入门指南 getting-started-transformers.rst 展示了如何将现有 Transformers 训练脚本迁移到 Ray Train。核心思路同样是：把所有逻辑（数据集构造与预处理、模型初始化、Transformers Trainer 定义）放入训练函数，然后做两处关键修改：

1. 添加上报回调：

import transformers from ray.train.huggingface.transformers import RayTrainReportCallback def train_func(): ... trainer = transformers.Trainer(...) trainer.add_callback(RayTrainReportCallback()) ...

2. 准备 Trainer：将 Transformers Trainer 传入ray.train.huggingface.transformers.prepare_trainer(trainer)以校验配置并启用 Ray Data 集成，随后即可启动训练。

一个重要的序列化注意事项：使用 Hugging Face Datasets 或 Evaluate 时，务必在训练函数内部调用datasets.load_dataset与evaluate.load，切勿把已加载的数据集和指标从函数外部传入——否则在把对象传输给 Worker 时可能引发序列化错误。训练完成后，可通过result.checkpoint.as_directory()找到RayTrainReportCallback.CHECKPOINT_NAME指定的检查点文件，用AutoModelForSequenceClassification.from_pretrained直接加载模型。

同样地，Transformers 集成也存在 API 迁移：Ray 2.1 引入的TransformersTrainer（trainer_init_per_worker接口）正在被 Ray 2.7 统一的TorchTrainer取代，新 API 与标准 Transformers 脚本风格更贴近，且支持通过torch_batches迭代 Ray Data 分片（ray.train.get_dataset_shard）作为训练数据。

XGBoost：梯度提升树的分布式训练

入门指南 getting-started-xgboost.rst 讲解了 XGBoost 脚本的改造路径。Ray Train 会自动完成分布式 XGBoost 训练所需的 Worker 通信设置，你只需：

1. 包装训练函数并配置参数。通过train_loop_config字典向train_func传参：

def train_func(config): label_column = config["label_column"] num_boost_round = config["num_boost_round"] ... config = {"label_column": "y", "num_boost_round": 10} trainer = ray.train.xgboost.XGBoostTrainer(train_func, train_loop_config=config, ...)

注意：避免通过train_loop_config传递大数据对象（如数据集、模型），以减少序列化/反序列化开销；应在train_func内部直接初始化大对象。

2. 添加上报回调：

import xgboost from ray.train.xgboost import RayTrainReportCallback def train_func(): ... bst = xgboost.train( ..., callbacks=[ RayTrainReportCallback(metrics=["eval-logloss"], frequency=1) ], ) ...

3. 数据分片。分布式训练时每个 Worker 应使用数据集的不同分片。最简单的方式是预先切分数据集并按ray.train.get_world_rank()分配文件；更灵活的方式是用 Ray Data 在运行时自动分片：先整体加载数据集，在训练函数内通过ray.train.get_dataset_shard(dataset_name)取本 Worker 的分片并转为xgboost.DMatrix，最后把数据集以字典形式传给 Trainer（key 必须与get_dataset_shard的调用一致）：

trainer = XGBoostTrainer(..., datasets={"train": train_dataset, "eval": eval_dataset}) trainer.fit()

4. 配置规模与 GPU。ScalingConfig可配置num_workers、use_gpu、resources_per_worker：

# 4 个节点，每节点 8 CPU scaling_config = ScalingConfig(num_workers=4, resources_per_worker={"CPU": 8}) # 4 个 Worker，每个使用 1 块 GPU scaling_config = ScalingConfig(num_workers=4, use_gpu=True)

使用 GPU 时还需在 XGBoost 参数中设置"device": "cuda"。另有一个实战要点：同时使用 Ray Data 时，不要用resources_per_worker占用集群全部 CPU——Ray Data 需要 CPU 并行执行预处理，否则会成为性能瓶颈。例如每节点 8 CPU 时，可为训练 Worker 分配 6 CPU、留 2 CPU 给 Ray Data。

5. 配置持久化存储与读取结果。通过RunConfig(storage_path=..., name=...)指定结果（含检查点与工件）的保存路径，支持本地路径、S3 云存储与 NFS。对单节点集群，共享存储位置是可选的；但对多节点集群，共享存储（云存储或 NFS）是必需的，使用本地路径会在检查点阶段报错。训练完成后返回的Result对象（源码定义见 python/ray/train/v2/api/result.py）包含：

result.metrics # 训练期间上报的指标 result.checkpoint # 训练期间上报的最新检查点 result.path # 日志存放路径 result.error # 训练失败时抛出的异常

JAX：GPU 与 TPU 上的 SPMD 训练

入门指南 getting-started-jax.rst 介绍 Ray Train 的ray.train.v2.jax.JaxTrainer。JAX 是面向加速器的数组计算与程序变换库，利用 XLA 编译器为 GPU/TPU 生成高度优化的代码；其核心威力在于jax.grad、jax.jit、jax.vmap等变换的可组合性。JaxTrainer 遵循单程序多数据（SPMD）范式，让训练代码在多个 Worker 上同时执行：

• TPU 场景：每个 Worker 运行在 TPU 切片内的一台独立 TPU 虚拟机上，Ray Train 自动完成 TPU 切片的原子性预留；
• GPU 场景：Ray 自动在 CUDA 设备上搭建 JAX 分布式系统。

TPU 伸缩配置需要用到 Ray 2.49.0 在 V2ScalingConfig中新增的两个字段：

from ray.train import ScalingConfig tpu_scaling_config = ScalingConfig( num_workers=4, # TPU 切片中 TPU VM 总数 use_tpu=True, # 初始化 JAX TPU 后端 topology="4x4", # TPU 芯片物理排布，多主机训练必填 accelerator_type="TPU-V6E", # TPU 代次，确保调度到目标切片 placement_strategy="SPREAD", # 将 Worker 分散放置 )

各字段含义：use_tpu布尔开关；topology描述 TPU 芯片排布（如"4x4"），多主机训练必须设置；num_workers应等于所有切片 TPU VM 总数（例如一个 v4-32 切片含 2x2x4 拓扑共 4 台 VM，则设为 4，两个 v4-32 切片则设为 8）；resources_per_worker通常请求每台 VM 的芯片数（如{"TPU": 4}）；accelerator_type指定 TPU 代次。GPU 场景则保持经典写法：ScalingConfig(num_workers=4, use_gpu=True)，默认每 Worker 一块 GPU。

改造 JAX 脚本只需把训练逻辑包进一个函数并传递给JaxTrainer，同时把print替换为ray.train.report({"loss": ..., "epoch": ...})上报指标。训练完成后同样通过result.metrics读取最终指标（如result.metrics['loss']）。

更多框架支持

如果你的框架不在上述五大指南中，可参考 more-frameworks.rst 索引下的集成指南：

• Hugging Face Accelerate 指南
• DeepSpeed 指南
• TensorFlow 与 Keras 指南
• LightGBM 指南
• Horovod 指南

仓库中对应的可直接运行示例代码位于 doc/source/train/doc_code/ 目录，包括tf_starter.py（TensorFlow 起步）、lightgbm_quickstart.py（LightGBM 快速入门）、xgboost_quickstart.py（XGBoost 快速入门）与hvd_trainer.py（Horovod）等。

深入进阶：User Guides 使用指南全景

主文档将 User Guides 定位为常见训练任务的"如何做"指南集合，覆盖以下主题（每篇均配有独立文档）：

• 数据加载与预处理：如何用 Ray Data 摄入、分片与预处理训练数据；
• 使用加速器：CPU/GPU/TPU 资源配置细节；
• 持久化存储：跨节点结果与检查点的可靠存储；
• 监控与日志：训练过程的日志与指标监控；
• 检查点：检查点的保存、管理与加载；
• 异步校验：检查点的异步验证模式；
• 实验追踪：与 MLflow 等工具集成；
• 结果访问：解读Result对象中的指标、检查点与最佳检查点；
• 容错训练：Worker 故障恢复与容错策略；
• 弹性训练：训练中动态增删 Worker；
• 监控你的应用：面向应用层的监控方案；
• 本地模式：num_workers=0时在单进程内调试运行；
• 可复现性：固定随机种子、复现训练结果；
• 超参数优化：与 Ray Tune 集成做超参搜索。

这些指南与主文档相互呼应：例如 hyperparameter-optimization.rst 对应"上报指标与检查点"能力（ray.train.report），fault-tolerance.rst 与 elastic-training.rst 则分别依赖RunConfig(max_failures=...)与ScalingConfig的弹性num_workers元组语法，这些都可在 python/ray/train/v2/api/config.py 的RunConfig定义与ScalingConfig源码中找到对应字段。

更多学习资源：教程、示例、基准与 API

除了入门指南与使用指南，train.md 还提供以下进阶资源：

• Tutorials（教程）：覆盖从视觉到推荐系统等 ML 负载模式的动手实践教程；
• Examples（端到端示例）：仓库 doc/source/train/examples/ 目录下按框架组织了大量可运行示例，例如 PyTorch 的 Fashion MNIST 示例 与回归示例、Transformers 的文本分类 notebook、Lightning 的 MNIST 示例、DeepSpeed 的 GPT-J 微调、JAX 的 intro_to_jax_trainer 等，覆盖模型微调、强化学习（RLHF）、推理等真实场景；
• Benchmarks（基准测试）：benchmarks.rst 提供 Ray Train 的规模基准数据；
• API 参考：doc/source/train/api/api.md 提供所有 Trainer 类、ScalingConfig、RunConfig、Checkpoint、Result等类的完整 API 描述，以及ray.train.report、ray.train.get_context等训练循环工具函数的用法。

总结

本文以 doc/source/train/train.md 为主线，完整覆盖了 Ray Train 的安装方式（pip install -U "ray[train]"）、四大核心概念（训练函数、Worker、ScalingConfig、Trainer）以及 PyTorch、PyTorch Lightning、Hugging Face Transformers、XGBoost、JAX 五大框架的分布式改造实战，并延伸介绍了更多框架集成与全部使用指南。无论你是要把现有单机脚本扩展到集群，还是要训练超大模型、处理超大数据集，Ray Train 都提供了统一、简洁的入口：定义一个训练函数，配好ScalingConfig，选择对应的 Trainer，调用fit()即可获得包含指标与检查点的Result。下一步，建议结合 examples 目录 中的端到端示例动手实践，并在 API 参考 中查阅更多细节。

• 人工智能
• 分布式训练
• 强化学习
• 任务调度
• 模型推理服务