Ray Data: Scalable Data Processing Guide
Table of Contents
1. Introduction to Ray Data
2. Core Abstractions: Datasets and Blocks
3. Installation and Setup
4. Reading and Writing Data
5. Transformations and Map/Reduce
6. Efficient IO and File Formats
7. Integration with Ray Train and ML Workflows
8. Performance Tuning
9. Fault Tolerance and Checkpointing
10. Examples and Recipes
11. Migrating from Pandas/Spark
12. Appendix: API Reference
1. Introduction to Ray Data
Ray Data is a distributed data processing library built on Ray. It offers an easy-to-use API for loading,
processing, and persisting large datasets across a cluster with minimal code changes from single-node
pandas workflows.
Ray Data focuses on efficiency, interoperability, and scalability while integrating closely with other Ray
libraries such as Train, Tune, and Serve.
2. Core Abstractions
Datasets
A [Link] is a distributed collection of data divided into logical blocks. Operations on datasets
are lazily executed and optimized by Ray.
Blocks
Blocks are the unit of parallelism and can be in-memory or stored on disk. They may be Arrow tables,
pandas DataFrames, or custom objects.
3. Installation and Setup
Install Ray Data with Ray core:
pip install ray[default]
Start Ray:
import ray
[Link]()
4. Reading and Writing Data
Read CSV
from [Link] import read_csv
ds = read_csv("s3://bucket/data/*.csv")
Read Parquet
from [Link] import read_parquet
ds = read_parquet("s3://bucket/data/")
Read JSON, Avro, and Custom Sources
Ray Data supports multiple formats; custom connectors can be implemented.
Write Data
ds.write_parquet("s3://bucket/output/")
5. Transformations and Map/Reduce
Map and Map Batches
ds = [Link](preprocess)
ds = ds.map_batches(batch_fn, batch_size=1024)
GroupBy and Aggregations
[Link]("user_id").sum("amount")
Join
ds_left.join(ds_right, on="id", how="inner")
6. Efficient IO and File Formats
Parquet for columnar storage and predicate pushdown
Arrow for in-memory efficient columnar representation
Use partitioning and predicate pushdown for selective reads
7. Integration with Ray Train and ML Workflows
Ray Data integrates with Ray Train for data-parallel training loops. Use dataset iterators inside training
steps to get scalable input pipelines.
8. Performance Tuning
Tune parallelism to match cluster resources
Use map_batches for minimized serialization overhead
Use columnar formats and predicate pushdown
9. Fault Tolerance and Checkpointing
Ray Data supports checkpointing datasets to persistent stores and resuming from partial progress.
10. Examples and Recipes
ETL pipelines reading from S3, transforming, and writing to Parquet
Preprocessing pipelines for ML training with augmentation
Windowed aggregations for time-series data
11. Migrating from Pandas or Spark
Replace pandas read_csv with [Link].read_csv
Convert DataFrame workflows to map_batches for batching
Use repartition and coalesce to control parallelism
12. Appendix: API Reference
(Contains function signatures, types, and code examples for key APIs.)
(Extended content: advanced block management, plugin connectors, cross-cluster data exchange,
examples converting complex pandas pipelines, handling skewed data, and tuning strategies to optimize
memory and IO.)
Tutorials
Tutorial 1 — ETL: Read CSV from S3, Transform, and Write Parquet
1. Install Ray and start a local cluster or connect to your remote cluster:
pip install ray[default]
python -c "import ray; [Link]()"
2. Create [Link] :
from [Link] import read_csv
ds = read_csv("s3://my-bucket/logs/*.csv")
def clean_batch(df):
df['timestamp'] = df['timestamp'].astype('datetime64[ms]')
df = [Link](subset=['user_id'])
return df
ds = ds.map_batches(clean_batch, batch_format="pandas")
ds.write_parquet("s3://my-bucket/cleaned/", file_format="parquet")
3. Run the pipeline:
python [Link]
Tutorial 2 — Preprocessing for Training with map_batches
1. Use map_batches to produce TF/PyTorch-ready batches:
import ray
from [Link] import read_parquet
[Link]()
ds = read_parquet("s3://my-bucket/cleaned/")
def to_torch(batch):
import torch
X = [Link](batch['features'].tolist())
y = [Link](batch['label'].tolist())
return {"X": X, "y": y}
batched = ds.map_batches(to_torch, batch_size=256, batch_format="pandas")
for batch in batched.iter_batches(batch_format="python"):
# send batch to training loop
train_step(batch['X'], batch['y'])
2. Notes:
Use repartition to control parallelism and avoid tiny partitions.
Use map_batches(..., batch_format='pandas') for efficient vectorized preprocessing.
These tutorials illustrate simple, repeatable patterns for real-world pipelines using Ray Data.