Deep Learning y Big Data

import time

def matmul_python(A, B):
    """Multiplicación de matrices con bucles for puros.
    Complejidad: O(n³). Extremadamente lento para n > 500.
    """
    n = len(A)
    m = len(B[0])
    k = len(B)
    C = [[0.0] * m for _ in range(n)]
    for i in range(n):
        for j in range(m):
            for p in range(k):
                C[i][j] += A[i][p] * B[p][j]
    return C

# Benchmark: matrices 512×512
n = 512
A = [[float(i + j) for j in range(n)] for i in range(n)]
B = [[float(i * j) for j in range(n)] for i in range(n)]

start = time.time()
C = matmul_python(A, B)
print(f"Python puro: {time.time() - start:.2f}s")
# → Python puro: ~45-120 segundos (¡minutos para n=1024!)

import numpy as np
import time

def matmul_numpy(A, B):
    """NumPy usa BLAS optimizado (OpenBLAS/MKL) internamente.
    Operaciones vectorizadas en C/Fortran con SIMD y multi-core.
    """
    return A @ B  # equivalente a np.matmul(A, B)

# Benchmark: matrices 512×512
n = 512
A = np.random.randn(n, n).astype(np.float32)
B = np.random.randn(n, n).astype(np.float32)

start = time.time()
C = matmul_numpy(A, B)
print(f"NumPy (CPU): {time.time() - start:.4f}s")
# → NumPy (CPU): ~0.005-0.02 segundos (¡~2000-10000× más rápido!)

# ¿Por qué? NumPy llama a BLAS (Basic Linear Algebra Subprograms)
# implementado en C/Fortran con instrucciones SIMD (AVX-512)
# y multi-threading automático.

# Para matrices más grandes:
n = 4096
A = np.random.randn(n, n).astype(np.float32)
B = np.random.randn(n, n).astype(np.float32)
start = time.time()
C = A @ B
print(f"NumPy 4096×4096: {time.time() - start:.3f}s")
# → ~1-3 segundos (4096³ = 68B operaciones)

import cupy as cp
import time

def matmul_cupy(A, B):
    """CuPy: API idéntica a NumPy pero ejecuta en GPU.
    Internamente usa cuBLAS (NVIDIA's BLAS para CUDA).
    """
    return A @ B

# Benchmark: matrices 4096×4096 en GPU
n = 4096
A_gpu = cp.random.randn(n, n, dtype=cp.float32)
B_gpu = cp.random.randn(n, n, dtype=cp.float32)

# Warm-up (primera ejecución compila kernels)
_ = A_gpu @ B_gpu
cp.cuda.Stream.null.synchronize()

start = time.time()
C_gpu = matmul_cupy(A_gpu, B_gpu)
cp.cuda.Stream.null.synchronize()  # Esperar a que GPU termine
print(f"CuPy (GPU): {time.time() - start:.4f}s")
# → CuPy (GPU): ~0.005-0.02 segundos para 4096×4096
# ¡~100-500× más rápido que NumPy para matrices grandes!

# Transferencia CPU ↔ GPU (el coste oculto)
A_cpu = np.random.randn(n, n).astype(np.float32)

start = time.time()
A_gpu = cp.asarray(A_cpu)       # CPU → GPU
elapsed_to_gpu = time.time() - start

start = time.time()
A_back = cp.asnumpy(A_gpu)     # GPU → CPU
elapsed_to_cpu = time.time() - start

print(f"CPU→GPU: {elapsed_to_gpu:.4f}s, GPU→CPU: {elapsed_to_cpu:.4f}s")
# ¡La transferencia puede ser más lenta que el cómputo!

import torch

# ═══════════════════════════════════════════════════
# Detectar devices disponibles
# ═══════════════════════════════════════════════════
print(f"CUDA disponible: {torch.cuda.is_available()}")
print(f"MPS disponible: {torch.backends.mps.is_available()}")  # Apple Silicon
print(f"Nº GPUs CUDA: {torch.cuda.device_count()}")
if torch.cuda.is_available():
    print(f"GPU actual: {torch.cuda.get_device_name(0)}")

# ═══════════════════════════════════════════════════
# Crear tensores en devices específicos
# ═══════════════════════════════════════════════════
x_cpu = torch.randn(1000, 1000)                    # CPU (por defecto)
x_gpu = torch.randn(1000, 1000, device='cuda')     # GPU directamente
x_gpu2 = torch.randn(1000, 1000, device='cuda:1')  # Segunda GPU

# ═══════════════════════════════════════════════════
# Mover tensores entre devices
# ═══════════════════════════════════════════════════
x_to_gpu = x_cpu.to('cuda')      # CPU → GPU
x_to_cpu = x_gpu.cpu()           # GPU → CPU
x_to_gpu_fp16 = x_cpu.to('cuda', dtype=torch.float16)  # + cambio de tipo

# ═══════════════════════════════════════════════════
# Mover modelos completos
# ═══════════════════════════════════════════════════
model = torch.nn.Linear(1000, 100)
model = model.to('cuda')         # Todos los parámetros a GPU
# ⚠️ Los datos de entrada también deben estar en el mismo device
output = model(x_to_gpu)

# ═══════════════════════════════════════════════════
# Device-agnostic code (buena práctica)
# ═══════════════════════════════════════════════════
device = torch.device(
    'cuda' if torch.cuda.is_available()
    else 'mps' if torch.backends.mps.is_available()
    else 'cpu'
)
model = model.to(device)
data = data.to(device)

import tensorflow as tf

# ═══════════════════════════════════════════════════
# Detectar devices disponibles
# ═══════════════════════════════════════════════════
print("Devices:", tf.config.list_physical_devices())
gpus = tf.config.list_physical_devices('GPU')
print(f"GPUs: {gpus}")
tpus = tf.config.list_physical_devices('TPU')
print(f"TPUs: {tpus}")

# ═══════════════════════════════════════════════════
# Configurar memoria GPU
# ═══════════════════════════════════════════════════
# TF por defecto reserva TODA la memoria GPU.
# Para permitir crecimiento dinámico:
for gpu in gpus:
    tf.config.experimental.set_memory_growth(gpu, True)

# O limitar a una cantidad fija:
# tf.config.set_logical_device_configuration(
#     gpus[0],
#     [tf.config.LogicalDeviceConfiguration(memory_limit=4096)]  # 4 GB
# )

# ═══════════════════════════════════════════════════
# Placement explícito de operaciones
# ═══════════════════════════════════════════════════
with tf.device('/CPU:0'):
    a = tf.random.normal([1000, 1000])  # En CPU

with tf.device('/GPU:0'):
    b = tf.random.normal([1000, 1000])  # En GPU

# TF automáticamente mueve datos entre devices si es necesario
c = tf.matmul(a, b)  # a se copia a GPU automáticamente

# ═══════════════════════════════════════════════════
# TPU setup (Google Cloud / Colab)
# ═══════════════════════════════════════════════════
# resolver = tf.distribute.cluster_resolver.TPUClusterResolver()
# tf.config.experimental_connect_to_cluster(resolver)
# tf.tpu.experimental.initialize_tpu_system(resolver)
# strategy = tf.distribute.TPUStrategy(resolver)

# ═══════════════════════════════════════════════════
# PyTorch DDP — Data Parallelism distribuido
# ═══════════════════════════════════════════════════
import torch
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.utils.data import DataLoader, DistributedSampler

def train(rank, world_size):
    # 1. Inicializar proceso distribuido
    dist.init_process_group("nccl", rank=rank, world_size=world_size)
    torch.cuda.set_device(rank)
    
    # 2. Crear modelo y wrappear con DDP
    model = MyModel().cuda(rank)
    model = DDP(model, device_ids=[rank])
    
    # 3. DataLoader con DistributedSampler
    sampler = DistributedSampler(dataset, num_replicas=world_size, rank=rank)
    dataloader = DataLoader(dataset, batch_size=32, sampler=sampler)
    
    optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4)
    
    for epoch in range(num_epochs):
        sampler.set_epoch(epoch)  # Shuffle diferente cada epoch
        for batch in dataloader:
            inputs = batch['input'].cuda(rank)
            targets = batch['target'].cuda(rank)
            
            output = model(inputs)
            loss = criterion(output, targets)
            
            optimizer.zero_grad()
            loss.backward()        # DDP sincroniza gradientes automáticamente
            optimizer.step()
    
    dist.destroy_process_group()

# Lanzar con: torchrun --nproc_per_node=4 train.py
# O: python -m torch.distributed.launch --nproc_per_node=4 train.py

# ═══════════════════════════════════════════════════
# TensorFlow MirroredStrategy — Data Parallelism
# ═══════════════════════════════════════════════════
import tensorflow as tf

# 1. Crear strategy (auto-detecta GPUs)
strategy = tf.distribute.MirroredStrategy()
print(f"Número de replicas: {strategy.num_replicas_in_sync}")

# 2. Crear modelo dentro del scope de la strategy
with strategy.scope():
    model = tf.keras.Sequential([
        tf.keras.layers.Dense(512, activation='relu'),
        tf.keras.layers.Dense(10)
    ])
    model.compile(
        optimizer=tf.keras.optimizers.Adam(1e-4),
        loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
        metrics=['accuracy']
    )

# 3. Crear dataset distribuido
batch_size_per_replica = 32
global_batch_size = batch_size_per_replica * strategy.num_replicas_in_sync

dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))
dataset = dataset.shuffle(10000).batch(global_batch_size).prefetch(tf.data.AUTOTUNE)

# 4. Entrenar — TF distribuye automáticamente
model.fit(dataset, epochs=10)

# Para multi-nodo:
# strategy = tf.distribute.MultiWorkerMirroredStrategy()

# ═══════════════════════════════════════════════════
# PyTorch FSDP — Fully Sharded Data Parallel
# ═══════════════════════════════════════════════════
import torch
from torch.distributed.fsdp import (
    FullyShardedDataParallel as FSDP,
    MixedPrecision,
    ShardingStrategy,
)
from torch.distributed.fsdp.wrap import transformer_auto_wrap_policy
from transformers import AutoModelForCausalLM

# 1. Configurar mixed precision
mp_policy = MixedPrecision(
    param_dtype=torch.bfloat16,
    reduce_dtype=torch.bfloat16,
    buffer_dtype=torch.bfloat16,
)

# 2. Definir qué capas se fragmentan
auto_wrap_policy = transformer_auto_wrap_policy(
    transformer_layer_cls={transformers.models.llama.modeling_llama.LlamaDecoderLayer}
)

# 3. Crear modelo FSDP
model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-2-7b-hf")
model = FSDP(
    model,
    sharding_strategy=ShardingStrategy.FULL_SHARD,  # ZeRO-3
    mixed_precision=mp_policy,
    auto_wrap_policy=auto_wrap_policy,
    device_id=torch.cuda.current_device(),
)

# 4. Training loop normal — FSDP gestiona fragmentación automáticamente
optimizer = torch.optim.AdamW(model.parameters(), lr=2e-5)
for batch in dataloader:
    loss = model(**batch).loss
    loss.backward()
    optimizer.step()
    optimizer.zero_grad()

# Lanzar: torchrun --nproc_per_node=8 --nnodes=4 train_fsdp.py