import pyarrow as pa
import pandas as pd
import numpy as np

# create pandas dataframe
df = pd.DataFrame({
    'user_id': range(1000000),
    'score': np.random.randn(1000000)
})

# convert to arrow with zero-copy
arrow_table = pa.Table.from_pandas(df, preserve_index=False)

# share with other processes/languages without serialization
# arrow format is ipc-compatible across languages
batch_bytes = pa.ipc.serialize_batch(
    arrow_table.to_batches()[0],
    pa.default_memory_pool()
)

# other process can read without deserialization overhead

from pyspark.sql import SparkSession

# initialize spark with data locality preferences
spark = SparkSession.builder \
    .appName("DataLocalityExample") \
    .config("spark.locality.wait", "10s") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# spark automatically schedules tasks on nodes containing the data
df = spark.read.parquet("hdfs://cluster/large-dataset/")

# transformations execute on data-local nodes
result = df.groupBy("category") \
    .agg({"value": "sum"}) \
    .filter("sum(value) > 1000")

import pandas as pd
import pyarrow.parquet as pq

# bad: csv format (no compression, row-based)
df = pd.read_csv("data.csv")  # 10gb file
# network transfer: 10gb 🔴

# good: parquet format (compressed, columnar)
df = pd.read_parquet("data.parquet")  # 2gb file (5x compression)
# network transfer: 2gb ✅

# write with optimal compression
df.to_parquet(
    "output.parquet",
    compression="snappy",  # fast compression
    engine="pyarrow"
)

# for even better compression (slower write, faster read)
df.to_parquet(
    "output.parquet",
    compression="zstd",  # better compression ratio
    compression_level=3
)

import dask.dataframe as dd
import dask.array as da
from dask.distributed import Client

# connect to dask cluster
client = Client("scheduler-address:8786")

# process 1tb dataset with pandas-like api
df = dd.read_parquet(
    "s3://bucket/large-dataset/*.parquet",
    engine="pyarrow"
)

# operations are lazy—build computation graph
filtered = df[df["score"] > 0.5]
grouped = filtered.groupby("category")["value"].mean()

# trigger computation across cluster
result = grouped.compute()

print(f"Processed {len(df)} rows using {len(client.cluster.workers)} workers")

import torch
import torch.nn as nn
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.utils.data import DataLoader, DistributedSampler

def setup_distributed(rank, world_size):
    """Initialize distributed training environment"""
    dist.init_process_group(
        backend="nccl",  # gpu communication
        init_method="tcp://localhost:12355",
        rank=rank,
        world_size=world_size
    )

class NeuralNetwork(nn.Module):
    def __init__(self):
        super().__init__()
        self.layers = nn.Sequential(
            nn.Linear(1000, 512),
            nn.ReLU(),
            nn.Linear(512, 256),
            nn.ReLU(),
            nn.Linear(256, 10)
        )
    
    def forward(self, x):
        return self.layers(x)

def train_distributed(rank, world_size, dataset):
    """Train model on distributed GPUs"""
    setup_distributed(rank, world_size)
    
    # move model to gpu
    model = NeuralNetwork().to(rank)
    
    # wrap model for distributed training
    ddp_model = DDP(model, device_ids=[rank])
    
    # create distributed sampler
    sampler = DistributedSampler(
        dataset,
        num_replicas=world_size,
        rank=rank,
        shuffle=True
    )
    
    dataloader = DataLoader(
        dataset,
        batch_size=32,
        sampler=sampler
    )
    
    optimizer = torch.optim.Adam(ddp_model.parameters(), lr=0.001)
    criterion = nn.CrossEntropyLoss()
    
    # training loop
    for epoch in range(10):
        sampler.set_epoch(epoch)  # shuffle differently each epoch
        
        for batch_idx, (data, target) in enumerate(dataloader):
            data, target = data.to(rank), target.to(rank)
            
            optimizer.zero_grad()
            output = ddp_model(data)
            loss = criterion(output, target)
            loss.backward()
            
            # gradients automatically synchronized across gpus
            optimizer.step()
        
        if rank == 0:
            print(f"Epoch {epoch}, Loss: {loss.item()}")
    
    dist.destroy_process_group()

# launch distributed training
import torch.multiprocessing as mp

if __name__ == "__main__":
    world_size = 4  # 4 gpus
    mp.spawn(
        train_distributed,
        args=(world_size, dataset),
        nprocs=world_size,
        join=True
    )

# optimize single-machine performance first
import numpy as np
from numba import jit, prange

@jit(parallel=True, nopython=True)
def compute_intensive_task(data):
    """Use Numba for CPU parallelization"""
    result = np.zeros_like(data)
    for i in prange(len(data)):
        result[i] = np.sum(data[i] ** 2)
    return result

# this can be 100x faster than pure python

from multiprocessing import Pool
import pandas as pd

def process_chunk(chunk):
    """Process data chunk independently"""
    return chunk.apply(expensive_function)

# utilize all cpu cores
with Pool(processes=8) as pool:
    chunks = np.array_split(df, 8)
    results = pool.map(process_chunk, chunks)
    df_processed = pd.concat(results)

import torch

# utilize multiple gpus on single machine
if torch.cuda.device_count() > 1:
    model = nn.DataParallel(model)
    model.to('cuda')

from dask.distributed import Client
from dask_cuda import LocalCUDACluster

# scale to multi-node gpu cluster
cluster = LocalCUDACluster()
client = Client(cluster)

# dask automatically distributes work

import pandas as pd
import gc

# bad: load entire dataset into memory
df = pd.read_csv("huge_file.csv")  # 🔴 out of memory!

# good: process in chunks
chunk_size = 100000
results = []

for chunk in pd.read_csv("huge_file.csv", chunksize=chunk_size):
    processed = process_chunk(chunk)
    results.append(processed)
    
    # explicitly free memory
    del chunk
    gc.collect()

final_result = pd.concat(results, ignore_index=True)

import pandas as pd

# sample data with repeated categories
df = pd.DataFrame({
    'category': ['A', 'B', 'C'] * 1000000,
    'value': range(3000000)
})

print(f"String memory: {df.memory_usage(deep=True)['category'] / 1024**2:.2f} MB")
# output: ~172 mb

# convert to categorical
df['category'] = df['category'].astype('category')
print(f"Categorical memory: {df.memory_usage(deep=True)['category'] / 1024**2:.2f} MB")
# output: ~2.9 mb (60x reduction!)

import numpy as np
import time

data = np.random.randn(1000000)

# bad: python loop
start = time.time()
result = []
for x in data:
    result.append(x ** 2 + 2 * x + 1)
print(f"Loop time: {time.time() - start:.3f}s")
# output: ~0.5s

# good: vectorized numpy
start = time.time()
result = data ** 2 + 2 * data + 1
print(f"Vectorized time: {time.time() - start:.3f}s")
# output: ~0.003s (150x faster!)

import dask.dataframe as dd
import dask.array as da
from dask.distributed import Client
import torch
from torch.utils.data import Dataset, DataLoader
import torchvision.transforms as transforms
from PIL import Image

# step 1: initialize dask cluster
client = Client(n_workers=16, threads_per_worker=2)

# step 2: create distributed dataset catalog
image_metadata = dd.read_parquet(
    "s3://bucket/metadata/*.parquet",
    columns=["image_path", "label", "split"]
)

# step 3: filter and prepare training data
train_data = image_metadata[
    image_metadata["split"] == "train"
].compute()

# step 4: custom dataset with lazy loading
class LazyImageDataset(Dataset):
    def __init__(self, metadata, transform=None):
        self.metadata = metadata
        self.transform = transform
    
    def __len__(self):
        return len(self.metadata)
    
    def __getitem__(self, idx):
        row = self.metadata.iloc[idx]
        
        # load image on-demand
        image = Image.open(row["image_path"])
        label = row["label"]
        
        if self.transform:
            image = self.transform(image)
        
        return image, label

# step 5: data augmentation pipeline
transform = transforms.Compose([
    transforms.Resize(256),
    transforms.RandomCrop(224),
    transforms.RandomHorizontalFlip(),
    transforms.ToTensor(),
    transforms.Normalize(
        mean=[0.485, 0.456, 0.406],
        std=[0.229, 0.224, 0.225]
    )
])

# step 6: create efficient data loader
dataset = LazyImageDataset(train_data, transform=transform)
dataloader = DataLoader(
    dataset,
    batch_size=256,
    shuffle=True,
    num_workers=8,  # parallel data loading
    pin_memory=True  # faster gpu transfer
)

# step 7: define model
class ImageClassifier(torch.nn.Module):
    def __init__(self, num_classes=1000):
        super().__init__()
        self.features = torch.nn.Sequential(
            torch.nn.Conv2d(3, 64, kernel_size=3, padding=1),
            torch.nn.ReLU(),
            torch.nn.MaxPool2d(2),
            torch.nn.Conv2d(64, 128, kernel_size=3, padding=1),
            torch.nn.ReLU(),
            torch.nn.MaxPool2d(2),
            torch.nn.Conv2d(128, 256, kernel_size=3, padding=1),
            torch.nn.ReLU(),
            torch.nn.AdaptiveAvgPool2d((1, 1))
        )
        self.classifier = torch.nn.Linear(256, num_classes)
    
    def forward(self, x):
        x = self.features(x)
        x = x.view(x.size(0), -1)
        x = self.classifier(x)
        return x

# step 8: training with mixed precision for speed
model = ImageClassifier(num_classes=1000).cuda()
optimizer = torch.optim.AdamW(model.parameters(), lr=0.001)
scaler = torch.cuda.amp.GradScaler()  # mixed precision

# step 9: training loop
for epoch in range(10):
    model.train()
    total_loss = 0
    
    for batch_idx, (images, labels) in enumerate(dataloader):
        images, labels = images.cuda(), labels.cuda()
        
        # mixed precision training
        with torch.cuda.amp.autocast():
            outputs = model(images)
            loss = torch.nn.functional.cross_entropy(outputs, labels)
        
        optimizer.zero_grad()
        scaler.scale(loss).backward()
        scaler.step(optimizer)
        scaler.update()
        
        total_loss += loss.item()
        
        if batch_idx % 100 == 0:
            print(f"Epoch {epoch}, Batch {batch_idx}, Loss: {loss.item():.4f}")
    
    avg_loss = total_loss / len(dataloader)
    print(f"Epoch {epoch} completed, Average Loss: {avg_loss:.4f}")

# step 10: save trained model
torch.save(model.state_dict(), "model_checkpoint.pth")

import torch.distributed as dist

# use nccl for gpu communication (optimized for nvidia)
dist.init_process_group(
    backend="nccl",  # nvidia collective communications library
    init_method="tcp://master:23456",
    world_size=8,
    rank=rank
)

# for cpu communication, use gloo
dist.init_process_group(
    backend="gloo",
    init_method="tcp://master:23456",
    world_size=8,
    rank=rank
)

# enable rdma for high-performance networking
import os
os.environ["NCCL_IB_DISABLE"] = "0"  # enable infiniband
os.environ["NCCL_NET_GDR_LEVEL"] = "5"  # gpu direct rdma

import prometheus_client as prom
from prometheus_client import Counter, Histogram, Gauge
import time

# define metrics
requests_total = Counter(
    'data_processing_requests_total',
    'Total data processing requests'
)

processing_duration = Histogram(
    'data_processing_duration_seconds',
    'Time spent processing data',
    buckets=[0.1, 0.5, 1.0, 5.0, 10.0, 30.0, 60.0]
)

gpu_utilization = Gauge(
    'gpu_utilization_percent',
    'GPU utilization percentage',
    ['gpu_id']
)

# instrument your code
@processing_duration.time()
def process_data_batch(batch):
    requests_total.inc()
    # process data
    result = expensive_computation(batch)
    return result

# monitor gpu utilization
import pynvml

pynvml.nvmlInit()
device_count = pynvml.nvmlDeviceGetCount()

for i in range(device_count):
    handle = pynvml.nvmlDeviceGetHandleByIndex(i)
    util = pynvml.nvmlDeviceGetUtilizationRates(handle)
    gpu_utilization.labels(gpu_id=i).set(util.gpu)

# expose metrics endpoint
prom.start_http_server(8000)

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def fetch_remote_data(url):
    """Retry failed downloads with exponential backoff"""
    response = requests.get(url, timeout=30)
    response.raise_for_status()
    return response.json()

# checkpointing for long-running jobs
def train_with_checkpoints(model, dataloader, checkpoint_dir):
    start_epoch = load_checkpoint_if_exists(checkpoint_dir)
    
    for epoch in range(start_epoch, num_epochs):
        for batch in dataloader:
            train_step(model, batch)
        
        # save checkpoint every epoch
        save_checkpoint(model, epoch, checkpoint_dir)

import pandera as pa
from pandera import Column, Check

# define schema for data validation
schema = pa.DataFrameSchema({
    "user_id": Column(pa.Int, Check.greater_than(0)),
    "score": Column(pa.Float, Check.in_range(0, 1)),
    "timestamp": Column(pa.DateTime),
    "category": Column(pa.String, Check.isin(["A", "B", "C"]))
})

# validate data at pipeline boundaries
@pa.check_input(schema)
def process_validated_data(df):
    return df.groupby("category")["score"].mean()

# bad: skewed data distribution
df.groupBy("country").count()
# usa: 9,000,000 records
# other countries: 1,000 records each
# result: one worker processes 90% of data! 🔴

# good: repartition before expensive operations
df.repartition(100, "country") \
  .groupBy("country") \
  .agg(...)

# bad: multiple shuffles
df.groupBy("col1").sum() \
  .join(other_df) \
  .groupBy("col2").mean()  # each operation shuffles data 🔴

# good: minimize shuffles
df_agg = df.groupBy("col1", "col2").agg(sum("val1"), mean("val2"))
result = df_agg.join(other_df)