文档教程开发工具【免费下载链接】pysheeetPython Cheat Sheet项目地址https://gitcode.com/gh_mirrors/py/pysheeet点击查看免费下载Ray 是用于将 Python 应用扩展到集群的开源分布式计算框架支持分布式机器学习训练、强化学习、超参数调优与模型服务。在 HPC 环境中常见的做法是用 Slurm 管理节点资源、用 Docker 容器承载运行环境再在容器之上拉起 Ray 集群从而把 Slurm 的资源调度能力与 Ray 的灵活分布式运行时结合起来。读完本文你将掌握 Ray 的 Python API 核心用法远程函数、Actor、对象存储、资源与容错控制、Placement Group学会用仓库提供的 sbatch 脚本在 Slurm 上启动 Docker 化 Ray 集群并通过 Dashboard、raytop 与ray job submit完成监控与作业提交。本文基于 docs/notes/hpc/ray.rst 展开并以仓库中的 src/ray/ray.sbatch、src/ray/Dockerfile 与 src/megatron/recipes/deepseek_v2_lite_pretrain.py 等真实文件作为实现依据。连接 Ray 集群三个默认端口Ray 集群协调依赖三个默认端口理解它们有助于排查连接问题6379— GCSGlobal Control Store集群主协调端口ray.init(addressHEAD_IP:6379)直连方式使用8265— Dashboard 与 REST API 端口ray job submit与 raytop 都通过它工作10001— Ray Client 服务端口供ray.init(addressray://...)使用。连接示例import ray # auto-detect在 Ray 容器内或通过 ray job submit 提交时自动发现 ray.init() # 通过 GCS 直连端口 6379 ray.init(addressHEAD_IP:6379) # 通过 Ray Client 连接端口 10001 ray.init(addressray://HEAD_IP:10001) # 或设置 RAY_ADDRESS 环境变量后直接调用 ray.init() # export RAY_ADDRESSHEAD_IP:6379 # 查看可用资源 print(ray.cluster_resources()) # {CPU: 192.0, GPU: 16.0, ...}远程函数ray.remote与任务的提交/收集ray.remote把普通函数变成分布式任务用.remote()提交用ray.get()收集结果提交的任务可以并行执行。import ray ray.remote def add(a, b): return a b # 并行提交任务 futures [add.remote(i, i) for i in range(4)] print(ray.get(futures)) # [0, 2, 4, 6] # 申请特定资源 ray.remote(num_gpus1) def train_on_gpu(data): import torch device torch.device(cuda) # ... # 申请多卡多核 ray.remote(num_gpus4, num_cpus8) def train_large_model(config): ...对象存储ray.put()共享大对象ray.put()把对象放进共享对象存储并返回 ObjectRef多个任务无需重复序列化即可访问同一份数据特别适合大模型权重或大数据集。import ray import numpy as np # 把大数组放入对象存储返回 ObjectRef data np.random.rand(10000, 10000) data_ref ray.put(data) ray.remote def process(data_ref): data ray.get(data_ref) # 同节点上零拷贝读取 return data.sum() # 所有任务共享同一份拷贝——无需重复序列化 futures [process.remote(data_ref) for _ in range(4)] print(ray.get(futures))ray.wait()按完成顺序处理结果ray.wait()返回已完成的任务而不会阻塞等待全部任务适合边完成边处理结果。import ray ray.remote def slow_task(i): import time time.sleep(i) return i futures [slow_task.remote(i) for i in range(5)] # 至少等待 2 个结果超时 3 秒 ready, not_ready ray.wait(futures, num_returns2, timeout3.0) print(ray.get(ready)) # 逐个处理完成的任务 while futures: ready, futures ray.wait(futures, num_returns1) print(fGot: {ray.get(ready[0])})资源管理集群资源查询与 GPU 分数分配Ray 暴露集群资源信息并支持分数 GPU 分配与自定义资源类型实现细粒度调度。import ray # 查看集群资源 print(ray.cluster_resources()) # 总量 print(ray.available_resources()) # 当前空闲 # 分数 GPU例如 2 个任务共享 1 张 GPU ray.remote(num_gpus0.5) def light_inference(batch): ...容错Actor 自动重启与任务重试长时间运行的分布式训练中Ray 可以自动重启崩溃的 Actor、重试失败的任务而不必让整个作业宕掉。import ray # Actor 失败时最多自动重启 3 次 ray.remote(num_gpus1, max_restarts3) class ResilientWorker: def __init__(self): self.setup_model() def setup_model(self): ... def train(self, batch): ... # 失败任务最多重试 3 次 ray.remote(num_gpus1, max_retries3) def train_step(data): ...Runtime Environments按任务注入依赖Runtime environments 允许为单个任务或 Actor 指定 pip 依赖、环境变量与工作目录而不必把这些打进容器镜像。import ray # 单任务 runtime env ray.remote(runtime_env{pip: [torch2.9.1], env_vars: {NCCL_DEBUG: INFO}}) def train(): ... # 或在 ray.init 时为所有任务统一设置 ray.init(runtime_env{working_dir: /workspace, pip: [vllm0.15.1]})Ray Actors有状态的工作进程Actor 是有状态的 Worker每个 Actor 运行在独立进程中方法调用之间保持状态。import ray ray.remote(num_gpus1) class Worker: def __init__(self, model_path): import torch self.model torch.load(model_path) self.device torch.device(cuda) def predict(self, batch): return self.model(batch.to(self.device)) # 创建 Actor 实例 workers [Worker.remote(/fsx/model.pt) for _ in range(4)] # 调用方法 futures [w.predict.remote(data) for w, data in zip(workers, batches)] results ray.get(futures)Placement Groups把任务钉在指定节点上Placement groups 控制任务与 Actor 在节点间的摆放方式对分布式训练至关重要——例如每个节点一个 placement group 并使用STRICT_PACK将 GPU Worker 固定到特定节点。import ray from ray.util.placement_group import placement_group from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy ray.init() # 创建 placement group8 张 GPU 打包在同一个节点 pg placement_group([{GPU: 1} for _ in range(8)], strategySTRICT_PACK) ray.get(pg.ready()) # 把 Actor 调度到组内特定 bundle ray.remote(num_gpus1) class GpuWorker: def run(self, rank): return frank {rank} on {ray.util.get_node_ip_address()} workers [] for i in range(8): w GpuWorker.options( scheduling_strategyPlacementGroupSchedulingStrategy( placement_grouppg, placement_group_bundle_indexi, ) ).remote() workers.append(w) futures [w.run.remote(i) for i, w in enumerate(workers)] print(ray.get(futures))多节点 placement groups例如跨 2 节点的分布式训练def create_placement_groups(num_nodes, gpus_per_node): 每个节点一个 placement group每组包含 gpus_per_node 个 GPU bundle。 pgs [] for _ in range(num_nodes): bundles [{GPU: 1} for _ in range(gpus_per_node)] pg placement_group(bundles, strategySTRICT_PACK) ray.get(pg.ready()) pgs.append(pg) return pgs # 2 节点 x 8 GPU 16 个 GPU Worker pgs create_placement_groups(num_nodes2, gpus_per_node8)四种放置策略对比STRICT_PACK— 所有 bundle 放在同一节点分布式训练首选PACK— 尽量少节点打包尽力而为STRICT_SPREAD— 每节点一个 bundle追求容错SPREAD— 跨节点尽力分散。实战用 Ray 启动 Megatron 分布式训练Ray placement groups 可以取代torchrun来启动 Megatron 分布式训练每个 Ray actor 对应一个 GPU rankplacement groups 保证 Worker 被钉在正确的节点上。仓库配套的 Megatron 场景是 2 节点 × 8 GPU 的 DeepSeek-V2-Lite 预训练。启动流程先启动 Ray 集群再从 head 容器内运行训练# 先启动 Ray 集群 salloc -N 2 bash examples/ray/ray.sbatch --image /fsx/megatron-lmlatest.tar.gz # 然后在 head 容器内 python3 main.py python3 main.py --hf-path /fsx/models/deepseek-ai/DeepSeek-V2-Lite python3 main.py --nodes 2 --gpus-per-node 8核心启动脚本要点如下与仓库 src/ray/ray.sbatch 的容器化方式同源即每个 Ray actor 一个 GPU rank直接调用torch.distributed.init_process_group()不再需要 torchrun#!/usr/bin/env python3 Launch Megatron Bridge DeepSeek-V2-Lite pretrain on a Ray cluster. Uses Ray placement groups to pin workers to nodes (same pattern as verls Megatron backend). Each Ray actor one GPU rank, calling torch.distributed.init_process_group() directly — no torchrun needed. import argparse import os import ray from ray.util.placement_group import placement_group from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy EFA_ENV { FI_PROVIDER: efa, FI_EFA_USE_DEVICE_RDMA: 1, FI_EFA_FORK_SAFE: 1, NCCL_DEBUG: WARN, NCCL_BUFFSIZE: 8388608, NCCL_P2P_NET_CHUNKSIZE: 524288, CUDA_DEVICE_MAX_CONNECTIONS: 1, OMP_NUM_THREADS: 1, } def build_config(hf_pathNone): from megatron.bridge.recipes.deepseek.deepseek_v2 import ( deepseek_v2_lite_pretrain_config, ) cfg deepseek_v2_lite_pretrain_config( **({hf_path: hf_path} if hf_path else {}), tensor_model_parallel_size8, pipeline_model_parallel_size1, expert_model_parallel_size2, sequence_parallelTrue, seq_length4096, train_iters500, global_batch_size64, micro_batch_size1, eval_interval100, lr_warmup_iters50, save_interval0, ) cfg.model.moe_permute_fusion False return cfg ray.remote(num_cpus0, num_gpus1) class Worker: def run(self, rank, local_rank, world_size, master_addr, master_port, hf_path): # 必须在任何 torch/CUDA 导入之前重置 CUDA_VISIBLE_DEVICES gpu_ids ray.get_gpu_ids() cuda_device int(gpu_ids[0]) if gpu_ids else local_rank os.environ.pop(CUDA_VISIBLE_DEVICES, None) os.environ.update(EFA_ENV) os.environ.update( { RANK: str(rank), LOCAL_RANK: str(cuda_device), WORLD_SIZE: str(world_size), MASTER_ADDR: master_addr, MASTER_PORT: str(master_port), } ) import megatron.core.jit as _jit if not hasattr(_jit, disable_jit_fuser): _jit.disable_jit_fuser lambda: None from megatron.bridge.training.gpt_step import forward_step from megatron.bridge.training.pretrain import pretrain cfg build_config(hf_path) pretrain(configcfg, forward_step_funcforward_step) def create_placement_groups(num_nodes, gpus_per_node): Create one placement group per node, each with gpus_per_node GPU bundles. pgs [] for _ in range(num_nodes): bundles [{GPU: 1} for _ in range(gpus_per_node)] pg placement_group(bundles, strategySTRICT_PACK) ray.get(pg.ready()) pgs.append(pg) return pgs def resolve_master_and_sort_pgs(pgs): Resolve node IPs for each PG, sort by IP, return (sorted_pgs, master_addr). ray.remote(num_cpus0) def _get_ip(): return ray.util.get_node_ip_address() ip_futures [] for pg in pgs: ref _get_ip.options( scheduling_strategyPlacementGroupSchedulingStrategy( placement_grouppg, placement_group_bundle_index0, ) ).remote() ip_futures.append(ref) ips ray.get(ip_futures) pg_ip_pairs sorted(zip(pgs, ips), keylambda x: x[1]) sorted_pgs [p for p, _ in pg_ip_pairs] master_addr pg_ip_pairs[0][1] return sorted_pgs, master_addr def main(): parser argparse.ArgumentParser() parser.add_argument(--hf-path, defaultdeepseek-ai/DeepSeek-V2-Lite) parser.add_argument(--nodes, typeint, default2) parser.add_argument(--gpus-per-node, typeint, default8) parser.add_argument(--master-port, typeint, default29500) args parser.parse_args() world_size args.nodes * args.gpus_per_node ray.init() pgs create_placement_groups(args.nodes, args.gpus_per_node) sorted_pgs, master_addr resolve_master_and_sort_pgs(pgs) # 每 GPU 一个 Worker钉在 placement groups 上 futures [] rank 0 for pg in sorted_pgs: for local_rank in range(args.gpus_per_node): worker Worker.options( scheduling_strategyPlacementGroupSchedulingStrategy( placement_grouppg, placement_group_bundle_indexlocal_rank, ) ).remote() futures.append( worker.run.remote( rank, local_rank, world_size, master_addr, args.master_port, args.hf_path, ) ) rank 1 ray.get(futures) if __name__ __main__: main()关键细节从源码看 rank 编排与 EFA 环境上面的脚本中有一个容易忽略但至关重要的细节Worker.run在导入任何 torch/CUDA 代码之前先从ray.get_gpu_ids()拿到本 Actor 被分配的真实 GPU ID并删除CUDA_VISIBLE_DEVICES再把LOCAL_RANK设为真实 CUDA 设备号。这是因为 Ray 与 Megatron 对 GPU 编号的约定不同若不重置会导致进程落到错误的设备上。此外脚本通过resolve_master_and_sort_pgs在每个 PG 的 bundle 0 上调度一个零资源任务取回节点 IP按 IP 排序后取第一个作为MASTER_ADDR保证 rank 编号与节点顺序稳定一致。整套EFA_ENVFI_PROVIDERefa、FI_EFA_USE_DEVICE_RDMA1、NCCL_P2P_NET_CHUNKSIZE524288等与 src/ray/Dockerfile 中构建 EFA/NCCL 时的 ENV 完全对齐确保多节点 NCCL 走 EFA 高速链路。仓库中还有一份等价的配置源src/megatron/recipes/deepseek_v2_lite_pretrain.py它把训练配置tensor_model_parallel_size8、expert_model_parallel_size2、seq_length4096、global_batch_size64等封装为configure()函数可供 Ray 场景直接复用也支持通过 Hydra 风格参数覆盖如model.tensor_model_parallel_size4。若走传统srun enroot路线非 Ray可参考 src/megatron/srun.sh 中srun --container-image ... --mpipmix --ntasks-per-node8的启动方式。在 Slurm 上启动 Ray 集群下面的 sbatch 脚本会在 Slurm 管理的多个节点上于 Docker 容器内拉起一个 Ray 集群在第一个分配节点上启动 Ray head在其余节点上启动 Ray worker然后轮询直到所有节点注册进集群。仓库中的完整实现见 src/ray/ray.sbatch。# 分配节点并以交互方式启动 salloc -N 2 bash ray.sbatch # 作为批处理作业提交 sbatch -N 2 ray.sbatch # 使用自定义容器镜像 sbatch -N 4 ray.sbatch --image /fsx/raylatest.tar.gz脚本会依次处理镜像加载tar.gz 归档或 registry pull、head/worker 启动并通过ray status做健康检查。从 src/ray/ray.sbatch 源码可看到几个值得关注的实现点镜像加载若--image以.tar.gz结尾用pigz -dc ... | tar -xf - -O manifest.json解析出镜像名再srun --ntasks-per-node1在每节点docker load否则走docker pull端口与节点发现RAY_PORT$((6379 (SLURM_JOB_ID % 1000)))用作业 ID 推导端口避免冲突HEAD_NODE${NODES[0]}取首个分配节点HEAD_IP由getent ahostsv4解析容器参数docker run --gpus ... --privileged --utshost --ipchost --nethost --ulimit memlock-1 --security-opt seccompunconfined并挂载/dev/infiniband下的uverbs*设备与 GDR 驱动保证 GPU、IB/EFA 直通head/worker 启动ray start --head --port${RAY_PORT} --dashboard-host0.0.0.0 --num-gpus$(nvidia-smi ... | wc -l) --num-cpus$(nproc) --disable-usage-stats --blockworker 则用ray start --address${HEAD_IP}:${RAY_PORT}加入集群健康检查最多重试 30 次、每次间隔 5 秒在 head 容器内执行ray status | grep -c node_直到活跃节点数达到$NUM_NODES。查看集群状态# 从 head 节点查看集群状态 docker exec ray-head ray status # 在容器内直接查看 ray status # 通过 REST API 获取集群信息 curl http://${HEAD_IP}:8265/api/cluster_status # 通过 REST API 列出节点 curl http://${HEAD_IP}:8265/api/v0/nodes # 列出运行中的作业 curl http://${HEAD_IP}:8265/api/jobs/Ray Dashboard 与 SSH 端口转发Ray 在 head 节点的8265端口内置 Web Dashboard。HPC 集群通常在防火墙后面可用 SSH 端口转发访问# 把 Ray dashboard 转发到本机 ssh -L 8265:${HEAD_IP}:8265 login-node # 然后在浏览器打开 # http://localhost:8265用 raytop 做终端实时监控raytop 是一个 Ray 集群的实时 TUI 监控工具类似分布式 GPU 训练界的htop提供集群级 CPU/GPU/内存利用率、每节点明细、基于 Prometheus 指标的每 GPU 利用率、运行作业状态与实时 Actor 数量等统一视图。安装与使用cargo install raytop # 指向 Ray dashboard 端点 raytop --master http://${HEAD_IP}:8265raytop 的数据来自多个 Ray API/api/cluster_status— 集群级 CPU/GPU/内存分配/api/v0/nodes— 每节点信息与状态/api/prometheus/sd— 实时每 GPU 利用率与显存占用/api/jobs/— 运行中的作业/api/v0/actors— 每节点 Actor 数量。交互快捷键j/k或方向键在节点间导航Enter打开详情面板Tab在作业与节点间切换焦点t切换主题q退出。对长期运行的分布式训练如 verl RLHF、Megatron 预训练raytop 可以在不打开浏览器的情况下快速确认所有节点活跃、GPU 满载、无 Actor 崩溃。提交 Ray 作业sbatch 脚本成功结束时会在日志中打印 head 地址# sbatch 脚本成功时输出 # [info] Ray cluster ready at HEAD_IP:RAY_PORT # 或在 head 容器内查询 docker exec ray-head ray status # 或查看 Slurm 作业日志 cat slurm-JOB_ID.out | grep Ray cluster ready使用ray job submit提交作业RAY_HEADhttp://HEAD_IP:8265 # 提交作业 ray job submit --address ${RAY_HEAD} -- python train.py # 提交 Megatron 示例 ray job submit --address ${RAY_HEAD} \ --runtime-env-json{working_dir: /workspace} \ -- python main.py --nodes 2 --gpus-per-node 8 # 带 runtime env 提交 ray job submit --address ${RAY_HEAD} \ --runtime-env-json{working_dir: /workspace} \ -- python train.py # 列出作业 ray job list --address ${RAY_HEAD} # 查看作业状态 ray job status --address ${RAY_HEAD} JOB_ID # 获取作业日志 ray job logs --address ${RAY_HEAD} JOB_ID # 停止作业 ray job stop --address ${RAY_HEAD} JOB_ID常用环境变量HPC 集群上运行 Ray 的常见环境变量# Ray 设置 export RAY_DEDUP_LOGS0 # 关闭日志去重 export RAY_USAGE_STATS_ENABLED0 # 关闭使用统计 # 多节点 GPU 通信的 NCCL 设置 export NCCL_DEBUGINFO export NCCL_SOCKET_IFNAME^docker,lo,veth export NCCL_P2P_NET_CHUNKSIZE524288 # EFA 设置AWS export FI_PROVIDERefa export FI_EFA_USE_DEVICE_RDMA1 export FI_EFA_FORK_SAFE1这些变量与 src/ray/Dockerfile 中内置的ENV NCCL_SOCKET_IFNAME^docker,lo,veth、ENV FI_PROVIDERefa、ENV NCCL_P2P_NET_CHUNKSIZE524288等保持一致说明容器镜像已为 EFA NCCL 多节点通信做好预配置其中NCCL_SOCKET_IFNAME^docker,lo,veth用于排除 docker、loopback 与 veth 虚拟网卡确保 NCCL 走真实高速网络。小结把 Slurm、Docker 与 Ray 三层叠加就能在 HPC 集群上获得Slurm 管资源、Docker 管环境、Ray 管分布式执行的完整链路。仓库中的 src/ray/ray.sbatch 是可直接套用的集群启动脚本src/ray/Dockerfile 提供了含 EFA/NCCL/NVSHMEM/vLLM/Megatron-Bridge 的完整训练镜像src/megatron/recipes/deepseek_v2_lite_pretrain.py 则给出了 Megatron 训练配置的复用入口。结合 Placement Group 的节点钉扎、raytop 的实时监控与ray job submit的作业管理即可稳定支撑多节点、多 GPU 的分布式训练作业。赞分享文档教程开发工具【免费下载链接】pysheeetPython Cheat Sheet项目地址https://gitcode.com/gh_mirrors/py/pysheeet点击查看免费下载相关推荐Horovod on Ray 实战指南用 RayExecutor 在 Ray 集群上运行分布式训练Horovod on Ray 实战指南用 RayExecutor 在 Ray 集群上运行分布式训练 导读 本文基于 Horovod 官方文档中 Horovo深度学习机器学习分布式训练Ludwig 分布式训练实战使用 Ray Job Submission 在远程 Ray 集群上运行训练任务Ludwig 分布式训练实战使用 Ray Job Submission 在远程 Ray 集群上运行训练任务 本篇技术指南围绕 Ludwig 官方示例 exam人工智能深度学习机器学习大模型预训练微调LoRA多模态NLP计算机视觉模型推理服务Ray集群部署教程在Kubernetes上运行分布式计算Ray是一个强大的分布式计算框架专为大规模数据处理和机器学习任务而设计。本教程将指导您如何在Kubernetes集群上部署Ray分布式计算框架实现从单机到集人工智能分布式训练强化学习任务调度模型推理服务后端上一篇Vulhub 环境复现YApi 开放注册导致的 Mock 沙箱逃逸远程代码执行RCE下一篇Agent Zero 的 document_query 工具文档加载、解析、索引与问答的插件化实现解析创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考