Ray Data

import ray
# Read Parquet filesds = ray.data.read_parquet("s3://bucket/data/*.parquet")
# Transform data (lazy execution)ds = ds.map_batches(lambda batch: {"processed": batch["text"].str.lower()})
# Consume datafor batch in ds.iter_batches(batch_size=100):    print(batch)

import rayfrom ray.train import ScalingConfigfrom ray.train.torch import TorchTrainer
# Create datasettrain_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 Raytrainer = TorchTrainer(    train_func,    datasets={"train": train_ds},    scaling_config=ScalingConfig(num_workers=4, use_gpu=True))trainer.fit()

import ray
# Parquet (recommended for ML)ds = ray.data.read_parquet("s3://bucket/data/*.parquet")
# CSVds = ray.data.read_csv("s3://bucket/data/*.csv")
# JSONds = ray.data.read_json("gs://bucket/data/*.json")
# Imagesds = ray.data.read_images("s3://bucket/images/")

# From listds = ray.data.from_items([{"id": i, "value": i * 2} for i in range(1000)])
# From rangeds = ray.data.range(1000000)  # Synthetic data
# From pandasimport pandas as pddf = pd.DataFrame({"col1": [1, 2, 3], "col2": [4, 5, 6]})ds = ray.data.from_pandas(df)

# Batch transformation (fast)def process_batch(batch):    batch["doubled"] = batch["value"] * 2    return batch
ds = ds.map_batches(process_batch, batch_size=1000)

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

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

# Group by columnds = ds.groupby("category").count()
# Custom aggregationds = ds.groupby("category").map_groups(lambda group: {"sum": group["value"].sum()})

# Use GPU for preprocessingdef 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)

# Write to Parquetds.write_parquet("s3://bucket/output/")
# Write to CSVds.write_csv("output/")
# Write to JSONds.write_json("output/")

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

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

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

import ray
# Load modeldef load_model():    # Load once per worker    return MyModel()
# Inference functionclass BatchInference:    def __init__(self):        self.model = load_model()
    def __call__(self, batch):        predictions = self.model(batch["input"])        return {"prediction": predictions}
# Run distributed inferenceds = ray.data.read_parquet("s3://data/")predictions = ds.map_batches(BatchInference, batch_size=32, num_gpus=1)predictions.write_parquet("s3://output/")

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

# Convert to PyTorchtorch_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"]

# Convert to TensorFlowtf_ds = ds.to_tf(feature_columns=["image"], label_column="label", batch_size=32)
for features, labels in tf_ds:    # Train model    pass