Optimizacion Spark — DataFlow Retail SL

Diagnostico de Problemas Criticos

🚨

Data Skew severo en Stage 3 (JOIN orders x product_catalog) Tareas individuales duran 45 min mientras la media es 2.3s. Ratio de skew: ~1170x. El campo product_id tiene distribucion muy desequilibrada (top 20 SKUs = 68% de pedidos).

⚠

AQE desactivado + UDFs Python para calculo de margenes Sin AQE, Spark no puede adaptarse a la distribucion real. Los UDFs Python serializan filas una a una (10-15x mas lento que funciones nativas de Catalyst).

📋

product_catalog (450 MB) se esta haciendo sort-merge join en lugar de broadcast Con autoBroadcastJoinThreshold en 10 MB por defecto, una tabla de 450 MB nunca se hace broadcast. Subirlo a 512 MB eliminaria el shuffle completamente.

Impacto Esperado de la Optimizacion

4h 12m

Duracion actual

antes

32m

Duracion objetivo

-87%

8.2 GB

Spill a disco

OOM en picos

~0 MB

Spill con AQE

eliminado

Distribucion de duracion de tareas — Stage 3

200 tareas en el join orders x product_catalog

ANTES — sin optimizacion

Min: 1.8sAvg: 2.3sMax: 45m 22s

DESPUES — con AQE + broadcast join

Min: 6.2sAvg: 8.1sMax: 10.4s

Script PySpark Optimizado

❌ Configuracion actual (lenta)

session_bad.py

# Configuracion heredada — multiple problemas
spark = (SparkSession.builder
    .appName("daily_sales_aggregation")
    # AQE desactivado — no se adapta
    .config("spark.sql.adaptive.enabled", "false")
    # Broadcast muy limitado
    .config("spark.sql.autoBroadcastJoinThreshold", "10MB")
    # Shuffle partitions fijo — no escala
    .config("spark.sql.shuffle.partitions", "200")
    # Sin Kryo serializer
    .getOrCreate())

# UDF Python para margen — muy lento
from pyspark.sql.functions import udf
from pyspark.sql.types import FloatType

@udf(FloatType())
def calc_margin(price, cost):
    if price and price > 0:
        return (price - cost) / price
    return 0.0

# Sort-merge join — shuffle masivo
result = orders.join(catalog, "product_id")
result = result.withColumn("margin",
    calc_margin(F.col("price"), F.col("cost")))

✅ Configuracion optimizada

session_optimized.py

# Sesion optimizada para DataFlow Retail
spark = (SparkSession.builder
    .appName("daily_sales_aggregation_v2")
    # AQE — adapta particiones y detecta skew
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    # Broadcast para product_catalog (450 MB)
    .config("spark.sql.autoBroadcastJoinThreshold", "512MB")
    # Kryo serializer — 30-40% menos memoria
    .config("spark.serializer",
            "org.apache.spark.serializer.KryoSerializer")
    .config("spark.executor.memory", "24g")
    .config("spark.executor.memoryOverhead", "4g")
    .getOrCreate())

# Funcion nativa — 100% en JVM, sin serializar
result = orders.join(
    F.broadcast(catalog), "product_id", "left"
).withColumn(
    "margin",
    F.when(F.col("price") > 0,
      (F.col("price") - F.col("cost")) / F.col("price")
    ).otherwise(F.lit(0.0))
)

daily_sales_aggregation_v2.py — pipeline completo

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, DoubleType
import datetime

# ── 1. SESION OPTIMIZADA ─────────────────────────────────────────────────────
spark = (SparkSession.builder
    .appName("daily_sales_aggregation_v2")
    # AQE: adapta particiones, detecta y divide particiones skewed
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
    # Broadcast: product_catalog 450MB < 512MB threshold
    .config("spark.sql.autoBroadcastJoinThreshold", "512m")
    # Memoria ejecutores r5.4xlarge (128GB, 16 cores → 3 ejecutores)
    .config("spark.executor.memory", "24g")
    .config("spark.executor.memoryOverhead", "4g")
    .config("spark.executor.cores", "5")
    .config("spark.memory.fraction", "0.6")
    .config("spark.memory.storageFraction", "0.4")
    # Serializacion Kryo: 30-40% menos memoria en shuffle
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.execution.arrow.pyspark.enabled", "true")
    # Particiones optimas para 47 GB de datos (128 MB/partition ~ 376)
    .config("spark.sql.shuffle.partitions", "auto")
    .config("spark.sql.files.maxPartitionBytes", "128m")
    # Compresion LZ4 para shuffle (mas rapida que snappy)
    .config("spark.io.compression.codec", "lz4")
    .config("spark.shuffle.compress", "true")
    .getOrCreate())

processing_date = datetime.date.today().strftime("%Y-%m-%d")
S3_BASE = "s3://dataflow-retail-datalake"

# ── 2. LECTURA EFICIENTE (predicate pushdown + column pruning) ────────────────
orders = (spark.read
    .format("parquet")
    .option("mergeSchema", "false")
    .load(f"{S3_BASE}/raw/orders/ingestion_date={processing_date}/")
    .select("order_id", "store_id", "customer_id",
            "order_ts", "channel"))

order_items = (spark.read
    .format("parquet")
    .option("mergeSchema", "false")
    .load(f"{S3_BASE}/raw/order_items/ingestion_date={processing_date}/")
    .select("order_id", "product_id", "quantity", "unit_price", "unit_cost"))

# Tablas pequenas: broadcast sin shuffle
product_catalog = (spark.read
    .parquet(f"{S3_BASE}/reference/product_catalog/")
    .select("product_id", "category", "subcategory", "brand"))

store_meta = (spark.read
    .parquet(f"{S3_BASE}/reference/store_metadata/")
    .select("store_id", "region", "city", "store_type"))

# ── 3. TRANSFORMACIONES (sin UDFs Python) ────────────────────────────────────
# Enriquecer order_items con catalog (broadcast — sin shuffle)
items_enriched = order_items.join(
    F.broadcast(product_catalog), "product_id", "left"
).withColumns({
    "revenue":     F.col("quantity") * F.col("unit_price"),
    "cost_total":  F.col("quantity") * F.col("unit_cost"),
    # Margen calculado con Catalyst (antes era UDF Python)
    "margin_pct":  F.when(
        (F.col("unit_price") > 0),
        (F.col("unit_price") - F.col("unit_cost")) / F.col("unit_price")
    ).otherwise(F.lit(0.0))
})

# Unir con orders y metadata de tienda
sales_full = (items_enriched
    .join(orders, "order_id", "inner")
    .join(F.broadcast(store_meta), "store_id", "left"))

# ── 4. AGREGACIONES EFICIENTES ───────────────────────────────────────────────
daily_summary = (sales_full
    .groupBy("region", "channel", "category", "brand")
    .agg(
        F.count("order_id").alias("num_orders"),
        F.sum("quantity").alias("units_sold"),
        F.sum("revenue").alias("total_revenue"),
        F.sum("cost_total").alias("total_cost"),
        F.avg("margin_pct").alias("avg_margin"),
        F.approx_count_distinct("customer_id").alias("unique_customers")
    )
    .withColumn("processing_date", F.lit(processing_date))
    .withColumn("gross_margin",
        F.col("total_revenue") - F.col("total_cost")))

# ── 5. ESCRITURA PARTICIONADA ─────────────────────────────────────────────────
(daily_summary.write
    .format("parquet")
    .option("compression", "snappy")
    .partitionBy("processing_date", "region")
    .mode("overwrite")
    .save(f"{S3_BASE}/curated/daily_sales_summary/"))

spark.stop()

Patrones de Optimizacion Aplicados

1

Broadcast Join — product_catalog

product_catalog (450 MB) se distribuye a cada ejecutor en lugar de shuffle masivo. Elimina completamente el Stage 3 que causaba 45 min de straggler.

critico F.broadcast() autoBroadcastJoinThreshold=512m

2

AQE — Adaptive Query Execution

Con AQE activado, Spark detecta y divide automaticamente particiones skewed en runtime. Coalesce inteligente de particiones despues del shuffle.

alto impacto adaptive.enabled=true skewJoin.enabled=true

3

Eliminar UDFs Python

calc_margin() UDF Python reemplazado por F.when() + operaciones nativas. Catalyst puede optimizar y code-gen el calculo. 10-15x mas rapido.

alto impacto F.when() arrow.enabled=true

4

Memory Tuning — r5.4xlarge

Configuracion ajustada para 128 GB RAM, 16 cores: 3 ejecutores x 24 GB + 4 GB overhead. Elimina los spills a disco de 8.2 GB.

medio executor.memory=24g KryoSerializer

5

Column Pruning + Predicate Pushdown

.select() explicito justo despues del .read: Parquet solo lee las columnas necesarias. Filtro por ingestion_date en ruta S3 evita escanear historico.

medio mergeSchema=false partition pruning

6

approx_count_distinct

Sustituye SELECT DISTINCT customer_id + count() (shuffle) por HyperLogLog nativo. Error <2%, sin shuffle extra. Para 50M clientes es fundamental.

medio approx_count_distinct() HyperLogLog

Configuracion de Produccion — DataFlow Retail

Parametro Spark	Valor anterior	Valor nuevo	Razon
spark.sql.adaptive.enabled	false	true	Optimizacion en runtime
spark.sql.adaptive.skewJoin.enabled	false	true	Divide particiones skewed
spark.sql.autoBroadcastJoinThreshold	10MB	512MB	Cubre product_catalog 450MB
spark.serializer	JavaSerializer	KryoSerializer	-35% memoria shuffle
spark.executor.memory	8g	24g	Elimina spills 8.2 GB
spark.executor.memoryOverhead	1g	4g	PySpark + Arrow
spark.executor.cores	4	5	3 ejec/nodo x 16 cores
spark.sql.shuffle.partitions	200	auto (AQE)	Ajuste dinamico
spark.sql.execution.arrow.pyspark.enabled	false	true	Pandas UDFs 10x mas rapidos
spark.io.compression.codec	snappy	lz4	Mas rapido en shuffle

Checklist de Implementacion

Mejoras inmediatas (dia 1)

✖
Activar AQE en spark-submit / EMR config
spark.sql.adaptive.enabled=true
✖
Subir autoBroadcastJoinThreshold a 512m
Elimina shuffle en join con product_catalog
✖
Reemplazar calc_margin UDF
Sustituir por F.when() en el script
✖
Activar KryoSerializer
En configuracion de cluster EMR

Mejoras semana 2 — Delta Lake

⌛
Migrar orders a Delta Lake
Upserts eficientes + OPTIMIZE + Z-ORDER por product_id
⌛
Bucketing en orders (200 buckets, customer_id)
Elimina shuffle en joins frecuentes con clientes
⌛
Checkpoint en lineage complejo
Truncar linaje en DAG de +10 stages
⌛
Monitoreo de skew con check_partition_skew()
Alert si skew ratio > 3x en cualquier stage

Roadmap de Implementacion

Semana 1

Optimizaciones criticas

AQE + broadcast join + eliminar UDFs. Reduccion esperada: 4h → 45 min

Semana 2

Delta Lake + Z-Order

Migracion de orders a Delta, compactacion y Z-ORDER por product_id. Objetivo: 45 → 32 min

Semana 3

Bucketing + Monitoring

Bucket joins para queries recurrentes, dashboard de metricas Spark UI + alertas

📈

Ahorro estimado en coste AWS EMR 4h 12m → 32 min = -87% de duracion. Con 10 workers r5.4xlarge (~3.60€/h cada uno), el ahorro por ejecucion es aprox. 128 € / dia (~46.700 €/ano).

Resumen de mejoras

Duracion del job 4h 12m → 32m

Spill a disco 8.2 GB → ~0 MB

Skew ratio Stage 3 1170x → <2x

Coste diario EMR ~151 € → ~23 €

Fallos OOM en picos frecuentes → eliminados

Pipeline daily_sales_aggregation — DataFlow Retail SL