Data Processing Ray Data 转换

# Ray Data 数据转换 Ray Data 中数据转换的完整指南。 ## 核心操作 ### 映射批次 (向量化) python # 推荐用于性能优化 def process_batch(batch): # batch 是 numpy 数组或 pandas Series 的字典 batch["doubled"] = batch["value"] * 2 return batch ds = ds.map_batches(process_batch, batch_size=1000) **性能**：比逐行处理快 10-100 倍 ### 映射 (逐行) python # 仅在无法向量化时使用 def process_row(row): row["squared"] = row["value"] ** 2 return row ds = ds.map(process_row) ### 过滤 python # 移除行 ds = ds.filter(lambda row: row["score"] > 0.5) ### 平铺映射 (Flat map) python # 一行 → 多行 def expand_row(row): return [{"value": row["value"] + i} for i in range(3)] ds = ds.flat_map(expand_row) ## GPU 加速转换 python def gpu_transform(batch): import torch data = torch.tensor(batch["data"]).cuda() # GPU 处理 result = data * 2 return {"processed": result.cpu().numpy()} ds = ds.map_batches(gpu_transform, num_gpus=1, batch_size=64) ## 分组操作 python # 按列分组 grouped = ds.groupby("category") # 聚合 result = grouped.count() # 自定义聚合 result = grouped.map_groups(lambda group: { "sum": group["value"].sum(), "mean": group["value"].mean() }) ## 最佳实践 1. **优先使用 map_batches 而非 map** - 快 10-100 倍 2. **调整 batch_size** - 更大通常更快（需平衡内存占用） 3. **使用 GPU 进行繁重计算** - 如图像/音频预处理 4. **流式处理大数据集** - 对于超过内存的数据使用 iter_batches