LitData

from PIL import Image
import litdata as ld

# use a local or S3 folder
input_dir = "my_large_images"     # or "s3://my-bucket/my_large_images"
output_dir = "my_resized_images"  # or "s3://my-bucket/my_resized_images"

inputs = [os.path.join(input_dir, f) for f in os.listdir(input_dir)]

# resize the input image
def resize_image(image_path, output_dir):
    output_image_path = os.path.join(output_dir, os.path.basename(image_path))
    Image.open(image_path).resize((224, 224)).save(output_image_path)

ld.map(
    fn=resize_image,
    inputs=inputs,
    output_dir="output_dir",
)

from PIL import Image
import litdata as ld

# use a local or S3 folder
input_dir = "my_large_images"     # or "s3://my-bucket/my_large_images"
output_dir = "my_resized_images"  # or "s3://my-bucket/my_resized_images"

inputs = [os.path.join(input_dir, f) for f in os.listdir(input_dir)]

# resize the input image
def resize_image(image_path, output_dir):
    output_image_path = os.path.join(output_dir, os.path.basename(image_path))
    Image.open(image_path).resize((224, 224)).save(output_image_path)

ld.map(
    fn=resize_image,
    inputs=inputs,
    output_dir="output_dir",
    num_workers=8,
)

from PIL import Image
import litdata as ld, Machine

# use a local or S3 folder
input_dir = "my_large_images"     # or "s3://my-bucket/my_large_images"
output_dir = "my_resized_images"  # or "s3://my-bucket/my_resized_images"

inputs = [os.path.join(input_dir, f) for f in os.listdir(input_dir)]

# resize the input image
def resize_image(image_path, output_dir):
    output_image_path = os.path.join(output_dir, os.path.basename(image_path))
    Image.open(image_path).resize((224, 224)).save(output_image_path)

ld.map(
    fn=resize_image,
    inputs=inputs,
    output_dir="output_dir",
    machine=Machine.DATA_PREP,
    num_nodes=4,
)

import numpy as np
from PIL import Image
import litdata as ld

def random_images(index):
    fake_images = Image.fromarray(np.random.randint(0, 256, (32, 32, 3), dtype=np.uint8))
    fake_labels = np.random.randint(10)
    return {"index": index, "image": fake_images, "class": fake_labels}

if __name__ == "__main__":
    ld.optimize(
        fn=random_images,                       # The function applied to each input
        inputs=list(range(1000)),               # The inputs to the function (here it's a list of numbers)
        output_dir="s3://my-bucket/fast_data",  # Optimized data is stored here
        num_workers=4,                          # The number of workers on the same machine
        chunk_bytes="64MB"                      # Size of each chunk
    )

    train_dataset = ld.StreamingDataset('s3://my-bucket/fast_data', shuffle=True, drop_last=True)
    train_dataloader = ld.StreamingDataLoader(train_dataset)
    for sample in train_dataloader:
        img, cls = sample['image'], sample['class']

import litdata as ld

hf_uri = "hf://datasets/open-thoughts/OpenThoughts-114k/data"

dataset = ld.StreamingDataset(hf_uri)

for batch in dataset:
    print(f"{batch=}")

import datasets, torch, os, warnings
from transformers import pipeline
from litdata import map, Machine

wikipedia = datasets.load_dataset("wikipedia", "20220301.en", cache_dir="data", trust_remote_code=True)['train']

class TextEmbedder:
    def __init__(self):
        self.model = None

    def __call__(self, index, output_dir, device):
        if self.model is None:
            self.model = pipeline("feature-extraction", model="BAAI/bge-small-en-v1.5", device=device)

        article = wikipedia[index]
        for block_start in range(0, len(article['text']), 512):
            article_block = article['text'][block_start: block_start + 512]
            embeddings = self.model(article_block)
            torch.save(embeddings, os.path.join(output_dir, f"{index}_{block_start}.pt"))

map(
    fn=TextEmbedder(),
    inputs=list(range(len(wikipedia))),
    output_dir="s3://my-bucket/wikipedia_embeddings",
)