Apache Spark

Arquitectura Interna de Spark

Para optimizar Spark, primero debes entender cómo funciona por dentro. La arquitectura tiene componentes bien definidos con roles específicos.

Componentes Clave

SparkSession: El punto de entrada

"""
Configuración de SparkSession para Data Engineering
"""
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import *

# Crear SparkSession con configuraciones optimizadas
spark = (SparkSession.builder
    .appName("DataEngineeringPipeline")
    .master("yarn")  # local[*] para dev, yarn para cluster
    
    # ── MEMORIA ────────────────────────────────────────────────────────────
    .config("spark.executor.memory", "8g")        # RAM por executor
    .config("spark.executor.memoryFraction", "0.8")
    .config("spark.driver.memory", "4g")
    .config("spark.memory.offHeap.enabled", "true")
    .config("spark.memory.offHeap.size", "2g")
    
    # ── PARALELISMO ────────────────────────────────────────────────────────
    .config("spark.executor.cores", "4")          # Cores por executor
    .config("spark.sql.shuffle.partitions", "200")  # Particiones después de shuffle
    .config("spark.default.parallelism", "200")
    
    # ── ADAPTIVE QUERY EXECUTION (AQE) - Spark 3.0+ ────────────────────────
    .config("spark.sql.adaptive.enabled", "true")          # AUTO optimiza en runtime
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")  # Reduce particiones pequeñas
    .config("spark.sql.adaptive.skewJoin.enabled", "true")  # Maneja data skew automáticamente
    
    # ── SERIALIZACIÓN ─────────────────────────────────────────────────────
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    
    # ── DELTA LAKE ─────────────────────────────────────────────────────────
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    
    .getOrCreate()
)

spark.sparkContext.setLogLevel("WARN")  # Reducir verbosidad de logs

RDDs: La Abstracción Fundamental

Los Resilient Distributed Datasets (RDDs) son la abstracción de bajo nivel de Spark. Aunque los DataFrames son preferidos en el código moderno, entender RDDs es fundamental para comprender cómo Spark funciona internamente.

"""
RDDs: cuando los necesitas en la práctica (2026)
La mayoría de operaciones se hacen con DataFrames,
pero hay casos donde RDDs son necesarios:
"""
sc = spark.sparkContext  # SparkContext

# ── CREAR RDDs ──────────────────────────────────────────────────────────────
# Desde Python collection
rdd_from_list = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], numSlices=4)

# Desde archivo
rdd_from_file = sc.textFile("s3://bucket/raw/data.txt", minPartitions=8)

# ── TRANSFORMACIONES (LAZY): no se ejecutan hasta una Action ────────────────
# map: 1-a-1
squared_rdd = rdd_from_list.map(lambda x: x ** 2)

# filter: elimina elementos
even_rdd = rdd_from_list.filter(lambda x: x % 2 == 0)

# flatMap: 1-a-N (útil para tokenización)
words_rdd = rdd_from_file.flatMap(lambda line: line.split(" "))

# reduceByKey: agregar por clave (más eficiente que groupByKey)
word_counts = (words_rdd
    .map(lambda word: (word.lower(), 1))
    .reduceByKey(lambda a, b: a + b)
    .sortBy(lambda x: x[1], ascending=False)
)

# ── ACTIONS: Triggerean la ejecución ────────────────────────────────────────
result = word_counts.take(10)          # Top 10 palabras
total = rdd_from_list.count()          # Contar elementos
total_sum = rdd_from_list.reduce(lambda a, b: a + b)  # Reducción
all_data = rdd_from_list.collect()     # ⚠️ Cargar TODO en memoria del driver

# ── PARTICIONAMIENTO ────────────────────────────────────────────────────────
print(f"Particiones actuales: {rdd_from_list.getNumPartitions()}")
rdd_repartitioned = rdd_from_list.repartition(8)    # Aumentar (con shuffle)
rdd_coalesced = rdd_from_list.coalesce(2)            # Reducir (sin shuffle si posible)

# ── PERSISTENCIA / CACHING ──────────────────────────────────────────────────
from pyspark import StorageLevel

# Cuando usas un RDD múltiples veces, cachearlo evita recompute
expensive_rdd = (rdd_from_file
    .filter(lambda x: len(x) > 10)
    .map(lambda x: x.upper())
)
expensive_rdd.persist(StorageLevel.MEMORY_AND_DISK)  # Cache en mem, spill a disco si necesario
# expensive_rdd.cache()  # Equivalente a MEMORY_ONLY

# Usar múltiples veces sin recompute
count = expensive_rdd.count()
sample = expensive_rdd.take(5)
expensive_rdd.unpersist()  # Liberar memoria cuando ya no es necesario

# ── LINEAGE: Spark guarda cómo reconstruir el RDD ─────────────────────────
print(expensive_rdd.toDebugString().decode())  # Muestra el DAG

DataFrames y Spark SQL

"""
DataFrames: la API moderna de Spark
Equivalente a un DataFrame de Pandas pero distribuido
"""
from pyspark.sql import functions as F
from pyspark.sql.types import *
from pyspark.sql.window import Window

# ── LECTURA DE DATOS ────────────────────────────────────────────────────────
# Parquet (columnar, comprimido, particionado)
orders_df = spark.read \
    .option("mergeSchema", "true") \
    .parquet("s3://bucket/silver/orders/")

# CSV con schema explícito (mejor performance, sin inferencia)
schema = StructType([
    StructField("order_id",    LongType(),    nullable=False),
    StructField("customer_id", IntegerType(), nullable=False),
    StructField("amount",      DoubleType(),  nullable=True),
    StructField("order_date",  TimestampType(), nullable=False),
    StructField("status",      StringType(),  nullable=True),
])
df = spark.read.schema(schema).option("header", "true").csv("s3://bucket/raw/orders.csv")

# Delta Lake
delta_df = spark.read.format("delta").load("s3://bucket/delta/orders/")

# ── TRANSFORMACIONES BÁSICAS ────────────────────────────────────────────────
clean_df = (orders_df
    # Filtrar
    .filter(F.col("status").isin("delivered", "shipped"))
    .filter(F.col("amount") > 0)
    
    # Transformar columnas
    .withColumn("order_year",  F.year(F.col("order_date")))
    .withColumn("order_month", F.month(F.col("order_date")))
    .withColumn("amount_usd",  F.round(F.col("amount"), 2))
    .withColumn("is_large",    F.when(F.col("amount") > 500, True).otherwise(False))
    
    # Renombrar
    .withColumnRenamed("order_id", "id")
    
    # Eliminar duplicados
    .dropDuplicates(["order_id"])
    
    # Seleccionar columnas
    .select("id", "customer_id", "amount_usd", "order_date", "order_year", "order_month", "is_large")
    
    # Repartition para escritura eficiente
    .repartition(F.col("order_year"), F.col("order_month"))
)

# ── AGREGACIONES ────────────────────────────────────────────────────────────
agg_df = (orders_df
    .filter(F.col("order_date") >= "2026-01-01")
    .groupBy(F.date_trunc("month", F.col("order_date")).alias("month"), F.col("status"))
    .agg(
        F.count("*").alias("order_count"),
        F.countDistinct("customer_id").alias("unique_customers"),
        F.sum("amount").alias("total_revenue"),
        F.avg("amount").alias("avg_amount"),
        F.percentile_approx("amount", 0.5).alias("median_amount"),
        F.max("amount").alias("max_amount"),
    )
    .orderBy("month", "status")
)

# ── JOINS ────────────────────────────────────────────────────────────────────
customers_df = spark.read.parquet("s3://bucket/silver/customers/")

# Inner Join
result_df = orders_df.join(customers_df, on="customer_id", how="inner")

# Broadcast Join: cuando una tabla es pequeña (< 10MB default)
# Evita shuffle completamente
small_categories = spark.read.parquet("s3://bucket/dim/categories/")
result_broadcast = orders_df.join(
    F.broadcast(small_categories),  # Copia a todos los executors
    on="category_id",
    how="left"
)

# ── SPARK SQL: SQL directo ──────────────────────────────────────────────────
orders_df.createOrReplaceTempView("orders")
customers_df.createOrReplaceTempView("customers")

# Puedes mezclar SQL y DataFrame API libremente
result_sql = spark.sql("""
    SELECT 
        DATE_TRUNC('month', o.order_date)   AS month,
        c.customer_segment,
        COUNT(*)                             AS orders,
        SUM(o.amount)                        AS revenue,
        AVG(o.amount)                        AS avg_ticket
    FROM orders o
    JOIN customers c ON o.customer_id = c.customer_id
    WHERE o.status = 'delivered'
      AND o.order_date >= '2026-01-01'
    GROUP BY 1, 2
    ORDER BY 1, revenue DESC
""")

# El resultado de spark.sql() es un DataFrame normal
result_sql.show(20)  # Mostrar en consola
result_sql.explain(True)  # Ver el query plan físico

Transformaciones Avanzadas

"""
Operaciones avanzadas de PySpark para Data Engineering
"""
from pyspark.sql.window import Window

# ── WINDOW FUNCTIONS en Spark ──────────────────────────────────────────────
# Definir la ventana
window_by_customer = Window.partitionBy("customer_id").orderBy("order_date")
window_monthly = Window.partitionBy(F.date_trunc("month", F.col("order_date")))

df_with_windows = orders_df.withColumn(
    # Número de orden por cliente (running count)
    "order_number", F.row_number().over(window_by_customer)
).withColumn(
    # Revenue acumulado por cliente
    "cumulative_revenue", F.sum("amount").over(
        window_by_customer.rowsBetween(Window.unboundedPreceding, Window.currentRow)
    )
).withColumn(
    # Valor del pedido anterior
    "prev_order_amount", F.lag("amount", 1).over(window_by_customer)
).withColumn(
    # % del total del mes
    "pct_monthly_revenue", 
    F.col("amount") / F.sum("amount").over(window_monthly) * 100
).withColumn(
    # Rank dentro del mes
    "monthly_rank", F.dense_rank().over(
        Window.partitionBy(F.date_trunc("month", F.col("order_date"))).orderBy(F.desc("amount"))
    )
)

# ── UDFs: Custom transformaciones ──────────────────────────────────────────
# ⚠️ Los UDFs Python son lentos (rompen el optimizador de Spark)
# Úsalos solo cuando no haya función nativa equivalente
from pyspark.sql.types import StringType

@F.udf(returnType=StringType())
def categorize_order(amount: float) -> str:
    if amount is None:
        return "unknown"
    elif amount >= 1000:
        return "enterprise"
    elif amount >= 500:
        return "large"
    elif amount >= 100:
        return "medium"
    return "small"

# Aplicar UDF
df.withColumn("order_category", categorize_order(F.col("amount")))

# ✅ MEJOR: Usar Pandas UDF (Vectorized - mucho más rápido)
import pandas as pd
from pyspark.sql.functions import pandas_udf

@pandas_udf(StringType())
def categorize_order_vectorized(amounts: pd.Series) -> pd.Series:
    """Pandas UDF: opera sobre Series completas, no elemento a elemento."""
    return pd.cut(
        amounts,
        bins=[-float('inf'), 100, 500, 1000, float('inf')],
        labels=['small', 'medium', 'large', 'enterprise']
    ).astype(str)

df.withColumn("order_category", categorize_order_vectorized(F.col("amount")))

# ── PROCESAMIENTO DE JSON COMPLEJO ─────────────────────────────────────────
from pyspark.sql.types import *

# Schema del JSON anidado
event_schema = StructType([
    StructField("event_id", LongType()),
    StructField("type", StringType()),
    StructField("data", StructType([
        StructField("user_id", IntegerType()),
        StructField("items", ArrayType(StructType([
            StructField("sku", StringType()),
            StructField("qty", IntegerType()),
            StructField("price", DoubleType())
        ])))
    ]))
])

events_df = spark.read.schema(event_schema).json("s3://bucket/raw/events/")

# Explotar array de items (1 fila por item)
items_df = events_df.withColumn("item", F.explode("data.items")) \
    .select(
        "event_id",
        "data.user_id",
        "item.sku",
        "item.qty",
        "item.price",
        (F.col("item.qty") * F.col("item.price")).alias("line_total")
    )

Structured Streaming

"""
Spark Structured Streaming: procesar streams como DataFrames
Paradigma: stream = tabla unbounded que crece infinitamente
"""
from pyspark.sql.streaming import *

# ── LEER DESDE KAFKA ────────────────────────────────────────────────────────
kafka_df = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka-1:9092,kafka-2:9092")
    .option("subscribe", "orders,returns")           # Múltiples topics
    .option("startingOffsets", "latest")             # Solo mensajes nuevos
    # .option("startingOffsets", "earliest")         # Replay desde inicio
    .option("maxOffsetsPerTrigger", 10000)           # Control de throughput
    .load()
)

# Kafka produce: key y value como binary
# Parsear el schema del message
from pyspark.sql.functions import from_json, col

order_schema = StructType([
    StructField("order_id", LongType()),
    StructField("customer_id", IntegerType()),
    StructField("amount", DoubleType()),
    StructField("timestamp", TimestampType()),
])

orders_stream = (kafka_df
    .selectExpr("CAST(value AS STRING) as json_value", "timestamp as kafka_timestamp", "topic")
    .withColumn("order", from_json(col("json_value"), order_schema))
    .select("order.*", "kafka_timestamp", "topic")
    .withWatermark("timestamp", "10 minutes")  # Latencia máxima tolerada
)

# ── TRANSFORMACIONES EN STREAMING ──────────────────────────────────────────
# Windowed aggregation: métricas cada 5 minutos, slides cada 1 minuto
windowed_metrics = (orders_stream
    .groupBy(
        F.window("timestamp", "5 minutes", "1 minute"),  # Tumbling/Sliding window
        F.col("topic")
    )
    .agg(
        F.count("order_id").alias("order_count"),
        F.sum("amount").alias("window_revenue"),
        F.avg("amount").alias("avg_amount")
    )
)

# ── ESCRIBIR RESULTADOS ────────────────────────────────────────────────────
# Opción 1: A Delta Lake (Lakehouse pattern)
query_delta = (windowed_metrics.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "s3://bucket/checkpoints/streaming-metrics/")
    .partitionBy("topic")
    .trigger(processingTime="1 minute")  # Micro-batch cada 1 min
    .start("s3://bucket/silver/streaming-metrics/")
)

# Opción 2: A Kafka (enrich and re-publish)
query_kafka = (orders_stream
    .select(
        F.to_json(F.struct("*")).alias("value"),
        F.col("order_id").cast("string").alias("key")
    )
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("topic", "orders-enriched")
    .option("checkpointLocation", "s3://bucket/checkpoints/orders-enriched/")
    .start()
)

# Esperar a que todos los streams terminen
spark.streams.awaitAnyTermination()

Optimización de Spark

# Detectar skew: algunas tasks duran 10x más que otras
# Spark UI > Stage > Task Distribution

# SOLUCIÓN 1: Salting (artificialmente aumentar cardinalidad)
import random

# Antes del JOIN, agregar salt al key grande
orders_salted = orders_df.withColumn(
    "salted_customer_id",
    F.concat(F.col("customer_id"), F.lit("_"), (F.rand() * 10).cast("int"))
)

customers_replicated = customers_df.withColumn(
    "salt", F.explode(F.array([F.lit(i) for i in range(10)]))
).withColumn(
    "salted_customer_id",
    F.concat(F.col("customer_id"), F.lit("_"), F.col("salt"))
)

# JOIN con salted key
result = orders_salted.join(customers_replicated, "salted_customer_id", "inner")

# SOLUCIÓN 2: AQE Skew Join (Spark 3.0+, automático)
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")
# Spark automáticamente divide particiones sesgadas en el join

# El problema: miles de archivos pequeños = overhead masivo de metadata
# (Common en streaming donde cada micro-batch escribe archivos)

# Detectar: ¿cuántos archivos y cuál es su tamaño promedio?
import subprocess
result = subprocess.run(
    ["aws", "s3", "ls", "--recursive", "s3://bucket/silver/orders/"],
    capture_output=True, text=True
)

# SOLUCIÓN 1: Repartition antes de escribir
# Regla: ~128-256MB por archivo en Parquet
target_file_size_mb = 128
estimated_size_gb = 10  # Tamaño estimado del dataset
num_partitions = max(1, int(estimated_size_gb * 1024 / target_file_size_mb))

df.repartition(num_partitions).write.parquet("s3://bucket/output/")

# SOLUCIÓN 2: Coalesce (sin shuffle, para reducir)
df.coalesce(50).write.parquet("s3://bucket/output/")

# SOLUCIÓN 3: OPTIMIZE en Delta Lake
spark.sql("OPTIMIZE delta.`s3://bucket/silver/orders/` WHERE order_year = 2026")
# Compacta automáticamente los archivos pequeños

from pyspark import StorageLevel

# ── REGLAS DE CACHE ─────────────────────────────────────────────────────────
# CACHEAR cuando:
# - Usas el mismo DataFrame más de una vez en el job
# - La recomputación es costosa (reads de S3, joins complejos)
# - Tienes suficiente memoria

# NO CACHEAR cuando:
# - Solo usas el DF una vez
# - El DF es muy grande (llenaría la memoria)
# - La recomputación es barata (simple transformación)

# Niveles de storage:
customer_stats.cache()                                   # MEMORY_ONLY
customer_stats.persist(StorageLevel.MEMORY_AND_DISK)    # Spill a disco si no cabe
customer_stats.persist(StorageLevel.DISK_ONLY)          # Solo disco (para DFs enormes)
customer_stats.persist(StorageLevel.MEMORY_ONLY_SER)   # Serializado en memoria (menos GC)

# SIEMPRE unpersist cuando ya no necesitas el cache
customer_stats.unpersist()

# ── BROADCAST VARIABLES: compartir datos read-only eficientemente ──────────
# Para lookup tables pequeñas que se usan en UDFs
country_map = {"US": "United States", "MX": "Mexico", "BR": "Brazil"}
broadcast_country_map = spark.sparkContext.broadcast(country_map)

@pandas_udf(StringType())
def expand_country_code(codes: pd.Series) -> pd.Series:
    mapping = broadcast_country_map.value  # Acceso local a la copia del executor
    return codes.map(lambda c: mapping.get(c, c))

# ── TUNING DE JVM ────────────────────────────────────────────────────────────
# Reducir GC overhead: usar G1GC en lugar de ParallelGC
# spark.executor.extraJavaOptions=-XX:+UseG1GC -XX:G1HeapRegionSize=16M

Guía Rápida de Optimización