Hadoop y Spark

📐 Fundamento

Componentes clásicos de Hadoop:

• HDFS (Hadoop Distributed File System): sistema de archivos distribuido.
• MapReduce: modelo de programación para procesamiento distribuido.
• YARN: gestor de recursos del clúster.

HDFS — almacenar archivos enormes:

Archivo de 1 TB
        │
        ▼ se divide en bloques de 128 MB
┌──┬──┬──┬──┬──┬──┬──┐
│B1│B2│B3│B4│B5│B6│B7│ ... (8000 bloques)
└──┴──┴──┴──┴──┴──┴──┘
        │
        ▼ cada bloque se replica 3 veces en distintos nodos
[Node 1: B1, B5]
[Node 2: B1, B2, B6]    ← redundancia para tolerancia a fallos
[Node 3: B2, B3, B7]
...

Si un nodo falla, los bloques siguen disponibles en otros nodos. HDFS detecta y re-replica automáticamente.

MapReduce — procesamiento paralelo:

Problema: contar palabras en un texto de 1 TB

Map (paralelo, ejecuta en cada nodo donde está el bloque):
  texto → [("hola", 1), ("mundo", 1), ("hola", 1), ...]

Shuffle (agrupa por clave):
  "hola" → [1, 1, 1, ...]
  "mundo" → [1, 1, ...]

Reduce (paralelo por clave):
  ("hola", suma de la lista) → ("hola", 1234567)
  ("mundo", ...) → ...

Por qué Hadoop quedó obsoleto:

• MapReduce escribe en disco entre cada paso → muy lento.
• API verbosa (mucho boilerplate Java).
• Spark hace lo mismo en memoria → 10-100x más rápido.

HDFS sigue usándose como almacenamiento, pero es reemplazado por S3/GCS en cloud. La capa de cómputo cambió completamente a Spark.

📐 Fundamento

Arquitectura Spark:

       [Driver Program]
       (tu código Python)
              │
              │ submit job
              ▼
       [Cluster Manager]
       (YARN, Kubernetes, standalone)
              │
              │ provisiona executors
              ▼
   ┌──────────┬──────────┬──────────┐
   ▼          ▼          ▼          ▼
[Executor] [Executor] [Executor] [Executor]
   │          │          │          │
[RAM+CPU]  [RAM+CPU]  [RAM+CPU]  [RAM+CPU]

Tres APIs principales:

Setup PySpark:

pip install pyspark

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("La Esquina Analytics") \
    .config("spark.sql.shuffle.partitions", "200") \
    .getOrCreate()

Leer datos:

# Leer Parquet (más eficiente)
df = spark.read.parquet("s3://la-esquina/ventas/")

# CSV
df = spark.read.option("header", True).csv("s3://la-esquina/ventas.csv")

# JSON
df = spark.read.json("s3://la-esquina/eventos.json")

# Desde PostgreSQL via JDBC
df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://host:5432/esquina") \
    .option("dbtable", "pedidos") \
    .option("user", "user") \
    .option("password", "pass") \
    .load()

# Inspeccionar
df.printSchema()
df.show(5)
df.count()

Transformaciones (lazy, no se ejecutan hasta que pidas un resultado):

from pyspark.sql import functions as F

# Filtrar
abiertos = df.filter(df.estado == "ABIERTO")

# Seleccionar columnas
df_select = df.select("id", "mesa_id", "total")

# Agregar columna calculada
df_con_iva = df.withColumn("total_con_iva", df.total * 1.13)

# Group by + agregación
ventas_por_dia = df.groupBy(
    F.date_trunc("day", "fecha").alias("dia")
).agg(
    F.count("*").alias("num_pedidos"),
    F.sum("total").alias("ingresos"),
    F.avg("total").alias("ticket_promedio")
)

# Joins
pedidos_con_clientes = pedidos_df.join(
    clientes_df,
    pedidos_df.cliente_id == clientes_df.id,
    "inner"  # o "left", "right", "outer"
)

Acciones (gatillan la ejecución):

df.show()              # mostrar primeras filas
df.count()             # contar filas
df.collect()           # traer TODO al driver (¡cuidado, OOM si es grande!)
df.write.parquet("...")  # escribir a disco

Ejemplo end-to-end: top 10 platillos más vendidos:

top_platillos = (
    spark.read.parquet("s3://la-esquina/items_pedido/")
    .filter(F.col("fecha") >= "2026-01-01")
    .groupBy("platillo_id")
    .agg(
        F.sum("cantidad").alias("unidades_vendidas"),
        F.sum(F.col("cantidad") * F.col("precio_unit")).alias("ingresos")
    )
    .join(
        spark.read.parquet("s3://la-esquina/platillos/"),
        on="platillo_id"
    )
    .select("nombre", "categoria", "unidades_vendidas", "ingresos")
    .orderBy(F.desc("ingresos"))
    .limit(10)
)

top_platillos.show()
top_platillos.write.mode("overwrite").parquet("s3://la-esquina/reportes/top_platillos/")

📐 Fundamento

DataFrames se pueden consultar con SQL nativo:

df.createOrReplaceTempView("pedidos")

resultado = spark.sql("""
    SELECT 
        DATE_TRUNC('day', fecha) AS dia,
        COUNT(*) AS num_pedidos,
        SUM(total) AS ingresos
    FROM pedidos
    WHERE fecha >= '2026-01-01'
    GROUP BY DATE_TRUNC('day', fecha)
    ORDER BY dia
""")

resultado.show()

UDF (User-Defined Function) — funciones custom:

from pyspark.sql.types import StringType

def categorizar_ticket(monto: float) -> str:
    if monto < 5: return "pequeño"
    elif monto < 20: return "mediano"
    else: return "grande"

categorizar_udf = F.udf(categorizar_ticket, StringType())

df_con_categoria = df.withColumn("categoria_ticket", categorizar_udf(df.total))

Mejor: usar funciones built-in cuando sea posible (más rápidas):

df_con_categoria = df.withColumn("categoria_ticket",
    F.when(F.col("total") < 5, "pequeño")
     .when(F.col("total") < 20, "mediano")
     .otherwise("grande")
)

UDFs en Python son ~100x más lentas que funciones nativas (cruzan boundary Python ↔ JVM).

📐 Fundamento

1. Particionamiento al escribir:

# Particiona los datos por año/mes para queries posteriores eficientes
df.write \
    .partitionBy("año", "mes") \
    .parquet("s3://la-esquina/ventas/")

2. Broadcast joins (para tablas pequeñas):

from pyspark.sql.functions import broadcast

# Si platillos_df es chico (< 100 MB), broadcast lo replica a todos los executors
# evitando shuffle costoso
resultado = pedidos_df.join(broadcast(platillos_df), on="platillo_id")

3. Cache para datos reusados:

df = spark.read.parquet("...")
df.cache()  # mantener en memoria

# Si vas a usar df múltiples veces:
df.filter(...).count()
df.filter(...).show()
df.groupBy(...).agg(...)

df.unpersist()  # liberar cuando termines

4. Predicate pushdown:

Spark automáticamente "empuja" filtros hacia el storage cuando es posible.

# Spark internamente lee solo las particiones necesarias
df = spark.read.parquet("s3://la-esquina/ventas/").filter("año = 2026")
# Solo lee s3://la-esquina/ventas/año=2026/

5. Coalescer para reducir archivos pequeños:

# Sin esto: cada partición → un archivo. 200 archivos tiny si tenés 200 partitions.
df.coalesce(10).write.parquet("...")  # combina en 10 archivos más grandes

6. Configuración importante:

spark = SparkSession.builder \
    .config("spark.sql.shuffle.partitions", "200") \  # default 200, ajustar según volumen
    .config("spark.sql.adaptive.enabled", "true") \   # AQE: optimización automática (Spark 3+)
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .getOrCreate()

Ver el plan de ejecución:

df.explain(True)  # muestra plan lógico, físico, etc.

📐 Fundamento

Plataformas para correr Spark:

Submit de un job:

spark-submit \
    --master yarn \
    --deploy-mode cluster \
    --executor-memory 4G \
    --executor-cores 2 \
    --num-executors 10 \
    mi_script.py --fecha 2026-05-01

Patrón ETL típico:

def main(fecha):
    spark = SparkSession.builder.getOrCreate()
    
    # EXTRACT
    raw = spark.read.json(f"s3://la-esquina/raw/{fecha}/")
    
    # TRANSFORM
    limpio = (raw
        .filter(F.col("evento_tipo").isNotNull())
        .withColumn("timestamp", F.to_timestamp("fecha_iso"))
        .dropDuplicates(["evento_id"])
    )
    
    # LOAD
    (limpio
        .repartition(50)
        .write
        .mode("overwrite")
        .partitionBy("año", "mes", "dia")
        .parquet("s3://la-esquina/clean/"))
    
    spark.stop()

if __name__ == "__main__":
    import sys
    main(sys.argv[1])

Ejecutar este job diariamente con Airflow:

# DAG de Airflow
from airflow import DAG
from airflow.providers.amazon.aws.operators.emr import EmrSubmitJobOperator

dag = DAG('etl_diario', schedule='@daily', ...)

procesar = EmrSubmitJobOperator(
    job_flow_id="j-XXX",
    steps=[{
        "Name": f"Procesar {{ ds }}",
        "ActionOnFailure": "CONTINUE",
        "HadoopJarStep": {
            "Jar": "command-runner.jar",
            "Args": ["spark-submit", "s3://scripts/etl.py", "{{ ds }}"]
        }
    }],
    dag=dag
)