# Ray Data 工作开发帮助开发人员在 SRP 构建、测试、部署和调试 Ray Data 作业。 Ray Data 专为具有分布式 CPU/GPU 协调的大规模 ML 工作负载而设计。 ## 何时使用此技能在以下情况下使用此技能： * 为批量推理、数据预处理或 ETL 编写新的 Ray Data 作业 * 在本地或 GPU 集群上测试 Ray Data 代码 * 将作业部署到生产 RayCluster * 调试性能问题或故障 * 在 Airflow 中设置计划作业 ## 开发工作流程 ### 1\. A10 的本地开发 **环境设置：** # 连接A10开发机 ssh oci-dev2.ssh.buildagi.us # A102 开发机 # 或 ssh oci-dev3.ssh.buildagi.us # A103 开发机 # 创建虚拟环境 python3 -m venv my_venv 源 my_venv/bin/activate # 安装 Ray Data 和依赖项 pip install ray[数据] pip install Transformers Pillow torch # 根据需要添加其他依赖项 **开发技巧：** * 使用VSCode/Cursor进行远程编辑 * 始终使用 virtualenv 以避免依赖冲突 * 将代码存储在 `/data/srp/raydata/code/` 中以获得 RayCluster 可见性 * 首先使用小数据集进行测试 **本地运行：** # 简单的Python执行 python /data/srp/raydata/code/your-job.py **何时使用本地开发：** * 初步代码编写和调试 * 不需要GPU的工作 * 适合 A10 GPU 内存 (~24GB) 的型号 ### 2\. Slurm H100/H200 测试当模型对于 A10 来说太大或需要更多 GPU 内存时： # 连接到 Slurm 集群 ssh -p 2222 zhuguangbin@129.80.180.16 # 使用 H100/H200 GPU 启动 apptainer sapptainer -c 20 -m 200G -g 1 -p h100 -i /data/srp/apptainer/ray_2.52.0-py310-gpu.sif # 在 apptainer 中运行你的工作 python /data/srp/raydata/code/your-job.py **使用 Slurm 测试用于：** * 大型型号（例如 Qwen3-VL-8B-Instruct-FP8） * GPU 密集型工作负载 * 生产前的最终测试 ### 3\. RayCluster 上的生产部署 **设置 Ray 客户端：** # 在您的 Mac 或任何安装了 Ray 的机器上 conda 创建-n 射线 python=3.10 -y 康达激活射线 pip install -U“射线[全部]” # 设置RayCluster地址导出 RAY_ADDRESS="https://ray.g.yesy.online" **提交工作：** 射线作业提交\ --runtime-env-json='{ “pip”：[“火炬”，“火炬视觉”，“变形金刚”]， “env_vars”：{ "INPUT_PATH": "s3://bucket/path/to/input", "OUTPUT_PATH": "/data/srp/project/output" } }' \ -- python /data/srp/raydata/code/your-job.py **运行时环境参数：** * `pip`：要安装的Python包列表 * `env_vars`：环境变量（输入/输出路径、配置） **重要说明：** * 代码必须位于 `/data/srp/` 中（通过 JuiceFS 安装） * RayCluster 提供自动扩展、故障转移和作业调度 * 仪表板： ### 4\. Airflow 中的预定作业对于定期执行，请使用 Airflow BashOperator：运算符：airflow.operators.bash.BashOperator bash_命令：|- echo“正在安装射线[数据]...” pip install ray[数据] ~/.local/bin/ray 作业提交 \ --runtime-env-json='{"pip": ["torch"], "env_vars": {"INPUT_PATH": "..."}}' \ -- python /data/srp/raydata/code/your-job.py 环境： RAY_ADDRESS：“https://ray.g.yesy.online” 依赖项： - 开始 **参考：** ## 代码结构 ### 基本射线数据管道导入操作系统进口射线 # 1. 初始化 Ray 会话 ray.init() # 2.加载数据 #来自S3 ds = ray.data.read_parquet("s3://bucket/path/*.parquet") # 来自 JuiceFS ds = ray.data.read_parquet("/data/srp/project/data/*.parquet") # 图片 ds = ray.data.read_images("s3://bucket/images/", mode="RGB") # 3. 转换数据 ds = ds.map_batches( 您的处理器，计算=ray.data.ActorPoolStrategy(size=1), num_GPU=1，批量大小=16 ） # 4. 保存结果 ds.write_parquet(“/data/srp/project/output/”) ### 处理器类模式 **简单函数UDF：** def process_batch(批处理): # 处理批量数据批次[“结果”] = [计算批次[“输入”]中项目的(项目)] 退货批次 ds = ds.map_batches(process_batch,batch_size=32) **基于模型的处理器（推荐）：** 从输入导入字典将 numpy 导入为 np 从变压器进口管道从 PIL 导入图像批次大小 = 16 图像分类器类： def __init__(自身): # 在 __init__ 中加载模型一次 self.分类器 = 管道( “图像分类”，模型 =“google/vit-base-patch16-224”， device=0 # GPU 设备） def __call__(self, 批处理: Dict[str, np.ndarray]): # 将 numpy 数组转换为 PIL 图像 images = [Image.fromarray(img) for img in batch["image"]] # 运行推理输出 = self.classifier( 图像，顶部_k=1, 批量大小=BATCH_SIZE ） # 将结果添加到批次中 batch["score"] = [out[0]["score"] for out in 输出] batch["label"] = [out[0]["label"] for out in 输出] 退货批次 # 与 ActorPoolStrategy 一起使用以提高 GPU 效率预测= ds.map_batches( 图像分类器， compute=ray.data.ActorPoolStrategy(size=1), # GPU 副本数量 num_gpus=1, # 每个副本的 GPU 数批量大小=BATCH_SIZE ） ### 使用 vLLM 进行批量推理从 ray.data.llm 导入 LLMPredictor、vLLMEngineProcessorConfig model_id =“meta-llama/Llama-3.1-8B-Instruct” 处理器 = vLLMEngineProcessorConfig( model_id=model_id, 张量并行大小=1，最大模型长度=2048， max_num_batched_tokens=4096 ） ds = ds.map_batches( 法学硕士预测器， fn_constructor_kwargs={"处理器": 处理器}, num_GPU=1，批量大小=32 ） ## 最佳实践 ### 性能优化 1. **批量大小：** * 使用适合 GPU 内存的最大批量大小 * 更大的批次 = 更好的 GPU 利用率 * 从batch_size=16开始，增加直到OOM 2. **并发处理：** * 将CPU预处理与GPU推理分开 * 使用多个`map_batches`操作 * 在推理运行时启用并行预处理 3. **GPU利用率：** *为每个演员设置“num_gpus=1” * 对 N 个 GPU 副本使用 `ActorPoolStrategy(size=N)` * 将副本与集群中可用的 GPU 相匹配 4. **内存管理：** * 在 Ray 仪表板中监控内存使用情况 * 如果遇到OOM则减少batch_size * 使用更小的模型或更强大的 GPU ### 错误处理 # 设置AWS区域以避免端点解析问题导入操作系统如果“AWS_DEFAULT_REGION”不在 os.environ 中： os.environ["AWS_DEFAULT_REGION"] = "us-east-1" # 使用环境变量作为路径 input_path = os.environ.get("INPUT_PATH", "默认/路径") output_path = os.environ.get("OUTPUT_PATH", "默认/输出") # 在处理器中添加错误处理鲁棒处理器类： def __call__(自我，批处理)：尝试： # 处理批次退货批次除了异常 e： print(f"处理批次时出错：{e}") # 添加错误列或跳过批处理[“错误”] = str(e) 退货批次 ## 监控与调试 ### 雷仪表板访问仪表板： **关键指标：** * **作业状态：** 正在运行/已完成/失败的作业 * **集群节点：** 可用 GPU、CPU、内存 * **任务时间轴：** 每个任务的执行时间 * **操作员指标：** 吞吐量、批处理时间 * **资源使用情况：** GPU利用率、内存压力 ### 查看作业日志 # 列出职位雷工作清单 # 获取作业日志射线作业日志 # 实时关注日志 ray 作业日志 --follow ### 常见问题问题 |原因 |解决方案 ---|---|--- **GPU OOM** |批次太大 |减少“batch_size” **CPU OOM** |并发演员过多 |增加每个参与者的“num_cpus” **预处理速度慢** |顺序处理 |分为不同的“map”操作 **GPU 利用率低** |批量太小|增加“batch_size” **模型加载失败** |缺少依赖项 |添加到 `runtime-env-json` pip 列表 **S3 访问错误** |缺少 AWS 区域 |设置 `AWS_DEFAULT_REGION` 环境变量 ### 调试模式 # 以本地模式运行进行调试 ray.init(local_mode=True) # 单进程执行 # 取小样本进行测试 Sample = ds.take(10) # 获取 10 个项目打印（样本） # 限制数据以进行快速测试 ds = ds.limit(100) # 仅处理 100 个项目 ## 工作流程示例 ### 1\.图像分类导入操作系统进口射线从变压器进口管道从 PIL 导入图像 ray.init() ds = ray.data.read_images( os.environ.get("INPUT_PATH", "s3://bucket/images/"), 模式=“RGB” ）图像分类器类： def __init__(自身): self.分类器 = 管道( “图像分类”，模型 =“google/vit-base-patch16-224”，设备=0 ） def __call__(自我，批处理)： images = [Image.fromarray(img) for img in batch["image"]] 输出= self.classifier（图像，top_k = 1，batch_size = 16） batch["label"] = [out[0]["label"] for out in 输出] batch["score"] = [out[0]["score"] for out in 输出] 退货批次预测= ds.map_batches( 图像分类器，计算=ray.data.ActorPoolStrategy(size=1), num_GPU=1，批量大小=16 ）预测.write_parquet( os.environ.get("OUTPUT_PATH", "/data/srp/output/") ） ### 2\. Vision-Language Model (Qwen-VL) import os import ray import torch from transformers import AutoProcessor, AutoModelForVision2Seq from PIL import Image from qwen_vl_utils import process_vision_info ray.init() ds = ray.data.read_images( os.environ.get("INPUT_PATH", "s3://bucket/images/"), mode="RGB" ) PROMPT = os.environ.get( "PROMPT", "请详细描述这张图片中的内容，包括主要对象、场景和任何值得注意的细节。" ) class QwenVLAnalyzer: def __init__(self): model_name = "Qwen/Qwen3-VL-8B-Instruct-FP8" self.processor = AutoProcessor.from_pretrained(model_name) self.model = AutoModelForVision2Seq.from_pretrained( model_name, torch_dtype=torch.float16, device_map="auto" ) self.model.eval() def __call__(self, batch): images = [Image.fromarray(img) for img in batch["image"]] messages_list = [ [{ "role": "user", "content": [ {"type": "image", "image": img}, {"type": "text", "text": PROMPT} ] }] for img in images ] image_inputs, video_inputs = process_vision_info(messages_list) texts = self.processor.apply_chat_template( messages_list, tokenize=False, add_generation_prompt=True ) inputs = self.processor( text=texts, images=image_inputs, videos=video_inputs, padding=True, return_tensors="pt" ).to(self.model.device) with torch.no_grad(): generated_ids = self.model.generate(**inputs, max_new_tokens=512) generated_ids_trimmed = [ out[len(inp):] for inp, out in zip(inputs.input_ids, generated_ids) ] output_texts = self.processor.batch_decode( generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False ) batch["description"] = output_texts return batch predictions = ds.map_batches( QwenVLAnalyzer, compute=ray.data.ActorPoolStrategy(size=1), num_gpus=1, batch_size=20 ) predictions.write_parquet( os.environ.get("OUTPUT_PATH", "/data/srp/output/") ) ### 3\. Batch Inference with vLLM Reference: ### 4\. HTTP API Inference Reference: ## Resources ### Official Documentation * Ray Data Quickstart: * Working with LLMs: * Batch Inference: * Loading Data: * Transforming Data: * Working with Images: * Runtime Environments: ### SRP Resources * RayData Wiki: * ray-data-etl Project: * RayCluster Dashboard: * Airflow Examples: ## Quick Reference ### 常用命令 # 本地开发 ssh oci-dev2.ssh.buildagi.us python3 -m venv venv && 源 venv/bin/activate pip install ray[数据] # 泥浆测试 ssh -p 2222 zhuguangbin@129.80.180.16 sapptainer -c 20 -m 200G -g 1 -p h100 -i /data/srp/apptainer/ray_2.52.0-py310-gpu.sif # 制作提交导出 RAY_ADDRESS="https://ray.g.yesy.online" ray 作业提交 --runtime-env-json='...' -- python /data/srp/raydata/code/job.py # 监控雷工作清单 ray 作业日志 --follow ### 文件位置 * 代码存储：`/data/srp/raydata/code/` * 输出存储：`/data/srp/project/output/` * Apptainer 图像：`/data/srp/apptainer/` ## 实施步骤在帮助用户处理 Ray Data 作业时，请遵循以下工作流程： 1. **了解要求：** * 什么类型的处理？（推理、ETL、预处理） * 输入数据来源及格式 * 模型要求（GPU 内存、批量大小） * 预期输出格式 2. **选择开发环境：** * 从A10本地开发开始进行测试 * 如果需要较大型号，请移至 Slurm H100/H200 * 部署到 RayCluster 以实现生产规模 3. **编写代码：** * 使用处理器类模式 * 包括正确的错误处理 * 使用环境变量作为路径 *从小批量开始 4. **迭代测试：** * 首先使用 `.take(10)` 或 `.limit(100)` 进行测试 * 验证输出格式 * 在仪表板中检查资源使用情况 * 优化批量大小和并发量 5. **部署到生产：** * 使用适当的运行时环境提交到 RayCluster * 通过仪表板监控 * 如果需要，设置气流调度 6. **调试问题：** * 检查 Ray 仪表板是否有错误 * 查看作业日志 * 调整批量大小或资源 * 查阅故障排除表

射线数据

包含技能