Ray Data 分布式机器学习数据处理库 - 支持GPU加速和多模态数据

ray-data by orchestra-research/ai-research-skills

74 周安装量

5,600 GitHub Stars

GitHub

安装命令

npx skills add https://github.com/orchestra-research/ai-research-skills --skill ray-data

AI/机器学习分布式系统数据处理

🇨🇳中文介绍

Ray Data - 可扩展的机器学习数据处理

用于机器学习和人工智能工作负载的分布式数据处理库。

何时使用 Ray Data

在以下情况下使用 Ray Data：

为机器学习训练处理大型数据集（>100GB）
需要在集群上进行分布式数据预处理
构建批量推理流水线
加载多模态数据（图像、音频、视频）
将数据处理从笔记本电脑扩展到集群

关键特性：

流式执行：处理大于内存的数据
GPU 支持：使用 GPU 加速转换
框架集成：PyTorch、TensorFlow、HuggingFace
多模态：图像、Parquet、CSV、JSON、音频、视频

使用替代方案的情况：

Pandas：单机上的小数据（<1GB）
Dask：表格数据，类 SQL 操作
Spark：企业级 ETL，SQL 查询

快速开始

安装

pip install -U 'ray[data]'

加载和转换数据

import ray

# 读取 Parquet 文件
ds = ray.data.read_parquet("s3://bucket/data/*.parquet")

# 转换数据（惰性执行）
ds = ds.map_batches(lambda batch: {"processed": batch["text"].str.lower()})

# 消费数据
for batch in ds.iter_batches(batch_size=100):
    print(batch)

广告位招租

在这里展示您的产品或服务

触达数万 AI 开发者，精准高效

联系我们

与 Ray Train 集成

import ray
from ray.train import ScalingConfig
from ray.train.torch import TorchTrainer

# 创建数据集
train_ds = ray.data.read_parquet("s3://bucket/train/*.parquet")

def train_func(config):
    # 在训练中访问数据集
    train_ds = ray.train.get_dataset_shard("train")

    for epoch in range(10):
        for batch in train_ds.iter_batches(batch_size=32):
            # 在批次上训练
            pass

# 使用 Ray 进行训练
trainer = TorchTrainer(
    train_func,
    datasets={"train": train_ds},
    scaling_config=ScalingConfig(num_workers=4, use_gpu=True)
)
trainer.fit()

import ray

# Parquet（推荐用于机器学习）
ds = ray.data.read_parquet("s3://bucket/data/*.parquet")

# CSV
ds = ray.data.read_csv("s3://bucket/data/*.csv")

# JSON
ds = ray.data.read_json("gs://bucket/data/*.json")

# 图像
ds = ray.data.read_images("s3://bucket/images/")

从 Python 对象读取

# 从列表
ds = ray.data.from_items([{"id": i, "value": i * 2} for i in range(1000)])

# 从范围
ds = ray.data.range(1000000)  # 合成数据

# 从 pandas
import pandas as pd
df = pd.DataFrame({"col1": [1, 2, 3], "col2": [4, 5, 6]})
ds = ray.data.from_pandas(df)

批量映射（向量化）

# 批量转换（快速）
def process_batch(batch):
    batch["doubled"] = batch["value"] * 2
    return batch

ds = ds.map_batches(process_batch, batch_size=1000)

# 逐行处理（较慢）
def process_row(row):
    row["squared"] = row["value"] ** 2
    return row

ds = ds.map(process_row)

# 过滤行
ds = ds.filter(lambda row: row["value"] > 100)

# 按列分组
ds = ds.groupby("category").count()

# 自定义聚合
ds = ds.groupby("category").map_groups(lambda group: {"sum": group["value"].sum()})

# 使用 GPU 进行预处理
def preprocess_images_gpu(batch):
    import torch
    images = torch.tensor(batch["image"]).cuda()
    # GPU 预处理
    processed = images * 255
    return {"processed": processed.cpu().numpy()}

ds = ds.map_batches(
    preprocess_images_gpu,
    batch_size=64,
    num_gpus=1  # 请求 GPU
)

# 写入 Parquet
ds.write_parquet("s3://bucket/output/")

# 写入 CSV
ds.write_csv("output/")

# 写入 JSON
ds.write_json("output/")

# 控制并行度
ds = ds.repartition(100)  # 100 个块用于 100 核集群

# 更大的批次 = 更快的向量化操作
ds.map_batches(process_fn, batch_size=10000)  # 对比 batch_size=100

# 处理大于内存的数据
ds = ray.data.read_parquet("s3://huge-dataset/")
for batch in ds.iter_batches(batch_size=1000):
    process(batch)  # 流式处理，不加载到内存

import ray

# 加载模型
def load_model():
    # 每个工作器加载一次
    return MyModel()

# 推理函数
class BatchInference:
    def __init__(self):
        self.model = load_model()

    def __call__(self, batch):
        predictions = self.model(batch["input"])
        return {"prediction": predictions}

# 运行分布式推理
ds = ray.data.read_parquet("s3://data/")
predictions = ds.map_batches(BatchInference, batch_size=32, num_gpus=1)
predictions.write_parquet("s3://output/")

数据预处理流水线

# 多步骤流水线
ds = (
    ray.data.read_parquet("s3://raw/")
    .map_batches(clean_data)
    .map_batches(tokenize)
    .map_batches(augment)
    .write_parquet("s3://processed/")
)

与机器学习框架集成

# 转换为 PyTorch
torch_ds = ds.to_torch(label_column="label", batch_size=32)

for batch in torch_ds:
    # batch 是包含张量的字典
    inputs, labels = batch["features"], batch["label"]

# 转换为 TensorFlow
tf_ds = ds.to_tf(feature_columns=["image"], label_column="label", batch_size=32)

for features, labels in tf_ds:
    # 训练模型
    pass

支持的数据格式

格式	读取	写入	使用场景
Parquet	✅	✅	机器学习数据（推荐）
CSV	✅	✅	表格数据
JSON	✅	✅	半结构化数据
图像	✅	❌	计算机视觉
NumPy	✅	✅	数组
Pandas	✅	❌	数据框

扩展性（处理 100GB 数据）：

1 个节点（16 核）：约 30 分钟
4 个节点（64 核）：约 8 分钟
16 个节点（256 核）：约 2 分钟

GPU 加速（图像预处理）：

仅 CPU：1,000 张图像/秒
1 个 GPU：5,000 张图像/秒
4 个 GPU：18,000 张图像/秒

Pinterest：模型训练的最后一公里数据处理
ByteDance：使用多模态大语言模型扩展离线推理
Spotify：用于批量推理的机器学习平台

转换操作指南 - 映射、过滤、分组操作
集成指南 - Ray Train、PyTorch、TensorFlow

🇺🇸English

Ray Data - Scalable ML Data Processing

Distributed data processing library for ML and AI workloads.

When to use Ray Data

Use Ray Data when:

Processing large datasets (>100GB) for ML training
Need distributed data preprocessing across cluster
Building batch inference pipelines
Loading multi-modal data (images, audio, video)
Scaling data processing from laptop to cluster

Key features :

Streaming execution : Process data larger than memory
GPU support : Accelerate transforms with GPUs
Framework integration : PyTorch, TensorFlow, HuggingFace
Multi-modal : Images, Parquet, CSV, JSON, audio, video

Use alternatives instead :

Pandas : Small data (<1GB) on single machine
Dask : Tabular data, SQL-like operations
Spark : Enterprise ETL, SQL queries

Quick start

Installation

pip install -U 'ray[data]'

Load and transform data

import ray

# Read Parquet files
ds = ray.data.read_parquet("s3://bucket/data/*.parquet")

# Transform data (lazy execution)
ds = ds.map_batches(lambda batch: {"processed": batch["text"].str.lower()})

# Consume data
for batch in ds.iter_batches(batch_size=100):
    print(batch)

Integration with Ray Train

import ray
from ray.train import ScalingConfig
from ray.train.torch import TorchTrainer

# Create dataset
train_ds = ray.data.read_parquet("s3://bucket/train/*.parquet")

def train_func(config):
    # Access dataset in training
    train_ds = ray.train.get_dataset_shard("train")

    for epoch in range(10):
        for batch in train_ds.iter_batches(batch_size=32):
            # Train on batch
            pass

# Train with Ray
trainer = TorchTrainer(
    train_func,
    datasets={"train": train_ds},
    scaling_config=ScalingConfig(num_workers=4, use_gpu=True)
)
trainer.fit()

Reading data

From cloud storage

import ray

# Parquet (recommended for ML)
ds = ray.data.read_parquet("s3://bucket/data/*.parquet")

# CSV
ds = ray.data.read_csv("s3://bucket/data/*.csv")

# JSON
ds = ray.data.read_json("gs://bucket/data/*.json")

# Images
ds = ray.data.read_images("s3://bucket/images/")

From Python objects

# From list
ds = ray.data.from_items([{"id": i, "value": i * 2} for i in range(1000)])

# From range
ds = ray.data.range(1000000)  # Synthetic data

# From pandas
import pandas as pd
df = pd.DataFrame({"col1": [1, 2, 3], "col2": [4, 5, 6]})
ds = ray.data.from_pandas(df)

Transformations

Map batches (vectorized)

# Batch transformation (fast)
def process_batch(batch):
    batch["doubled"] = batch["value"] * 2
    return batch

ds = ds.map_batches(process_batch, batch_size=1000)

Row transformations

# Row-by-row (slower)
def process_row(row):
    row["squared"] = row["value"] ** 2
    return row

ds = ds.map(process_row)

Filter

# Filter rows
ds = ds.filter(lambda row: row["value"] > 100)

Group by and aggregate

# Group by column
ds = ds.groupby("category").count()

# Custom aggregation
ds = ds.groupby("category").map_groups(lambda group: {"sum": group["value"].sum()})

GPU-accelerated transforms

# Use GPU for preprocessing
def preprocess_images_gpu(batch):
    import torch
    images = torch.tensor(batch["image"]).cuda()
    # GPU preprocessing
    processed = images * 255
    return {"processed": processed.cpu().numpy()}

ds = ds.map_batches(
    preprocess_images_gpu,
    batch_size=64,
    num_gpus=1  # Request GPU
)

Writing data

# Write to Parquet
ds.write_parquet("s3://bucket/output/")

# Write to CSV
ds.write_csv("output/")

# Write to JSON
ds.write_json("output/")

Performance optimization

Repartition

# Control parallelism
ds = ds.repartition(100)  # 100 blocks for 100-core cluster

Batch size tuning

# Larger batches = faster vectorized ops
ds.map_batches(process_fn, batch_size=10000)  # vs batch_size=100

Streaming execution

# Process data larger than memory
ds = ray.data.read_parquet("s3://huge-dataset/")
for batch in ds.iter_batches(batch_size=1000):
    process(batch)  # Streamed, not loaded to memory

Common patterns

Batch inference

import ray

# Load model
def load_model():
    # Load once per worker
    return MyModel()

# Inference function
class BatchInference:
    def __init__(self):
        self.model = load_model()

    def __call__(self, batch):
        predictions = self.model(batch["input"])
        return {"prediction": predictions}

# Run distributed inference
ds = ray.data.read_parquet("s3://data/")
predictions = ds.map_batches(BatchInference, batch_size=32, num_gpus=1)
predictions.write_parquet("s3://output/")

Data preprocessing pipeline

# Multi-step pipeline
ds = (
    ray.data.read_parquet("s3://raw/")
    .map_batches(clean_data)
    .map_batches(tokenize)
    .map_batches(augment)
    .write_parquet("s3://processed/")
)

Integration with ML frameworks

PyTorch

# Convert to PyTorch
torch_ds = ds.to_torch(label_column="label", batch_size=32)

for batch in torch_ds:
    # batch is dict with tensors
    inputs, labels = batch["features"], batch["label"]

TensorFlow

# Convert to TensorFlow
tf_ds = ds.to_tf(feature_columns=["image"], label_column="label", batch_size=32)

for features, labels in tf_ds:
    # Train model
    pass

Supported data formats

Format	Read	Write	Use Case
Parquet	✅	✅	ML data (recommended)
CSV	✅	✅	Tabular data
JSON	✅	✅	Semi-structured
Images	✅	❌	Computer vision
NumPy	✅	✅	Arrays
Pandas	✅	❌	DataFrames

Performance benchmarks

Scaling (processing 100GB data):

1 node (16 cores): ~30 minutes
4 nodes (64 cores): ~8 minutes
16 nodes (256 cores): ~2 minutes

GPU acceleration (image preprocessing):

CPU only: 1,000 images/sec
1 GPU: 5,000 images/sec
4 GPUs: 18,000 images/sec

Use cases

Production deployments :

Pinterest : Last-mile data processing for model training
ByteDance : Scaling offline inference with multi-modal LLMs
Spotify : ML platform for batch inference

References

Transformations Guide - Map, filter, groupby operations
Integration Guide - Ray Train, PyTorch, TensorFlow

Resources

Docs : https://docs.ray.io/en/latest/data/data.html
GitHub : https://github.com/ray-project/ray ⭐ 36,000+
Version : Ray 2.40.0+
Examples : https://docs.ray.io/en/latest/data/examples/overview.html

Weekly Installs

Repository

orchestra-resea…h-skills

GitHub Stars

5.6K

First Seen

Feb 7, 2026

Security Audits

Gen Agent Trust HubWarn SocketPass SnykWarn

Installed on

opencode65

codex64

gemini-cli63

cursor63

github-copilot62

kimi-cli58

AI 代码实施计划编写技能 | 自动化开发任务分解与 TDD 流程规划工具

50,900 周安装

Ray Data 分布式机器学习数据处理库 - 支持GPU加速和多模态数据

🇨🇳中文介绍

Ray Data - 可扩展的机器学习数据处理

何时使用 Ray Data

快速开始

安装

加载和转换数据

相关 Skills

与 Ray Train 集成

读取数据

从云存储读取

从 Python 对象读取

转换操作

批量映射（向量化）

行级转换

过滤

分组和聚合

GPU 加速转换

写入数据

性能优化

重分区

批次大小调优

流式执行

常见模式

批量推理

数据预处理流水线

与机器学习框架集成

PyTorch

TensorFlow

支持的数据格式

性能基准测试

使用案例

参考

资源

🇺🇸English

Ray Data - Scalable ML Data Processing

When to use Ray Data

Quick start

Installation

Load and transform data

Integration with Ray Train

Reading data

From cloud storage

From Python objects

Transformations

Map batches (vectorized)

Row transformations

Filter

Group by and aggregate

GPU-accelerated transforms

Writing data

Performance optimization

Repartition

Batch size tuning

Streaming execution

Common patterns

Batch inference

Data preprocessing pipeline

Integration with ML frameworks

PyTorch

TensorFlow

Supported data formats

Performance benchmarks

Use cases

References

Resources

最新 Skills