Tuning Spark

En Spark, una etapa termina cuando acaba su tarea más lenta. Por eso una media razonable puede ocultar una cola extrema: si la mediana dura 18 segundos y una tarea tarda 11 minutos, añadir workers no elimina la clave caliente que concentra datos. La evidencia útil está en la pestaña Stages de Spark UI: duración máxima frente a mediana, registros de entrada, shuffle read y tamaño de cada tarea.

Antes de aplicar salting, confirma la causa. Un join con una clave nula dominante, un cliente desproporcionado o una partición temporal demasiado amplia producen remedios distintos. AQE puede dividir particiones sesgadas en joins compatibles; si la distribución forma parte del negocio, conviene además aislar claves calientes o rediseñar la agregación y medir el efecto con el mismo conjunto de datos.

from pyspark.sql import functions as F

key_profile = (
    orders.groupBy("customer_id")
    .count()
    .orderBy(F.desc("count"))
)

key_profile.show(20, truncate=False)
orders.where(F.col("customer_id").isNull()).count()

Spark derrama datos cuando una operación no puede mantener en memoria sus estructuras intermedias. El spill a disco añade serialización e I/O; un volumen pequeño puede ser normal, pero spill masivo junto con garbage collection, tareas largas o executor lost indica que el plan y la forma de los datos no caben de manera eficiente. Revisa la etapa concreta, el operador y la distribución antes de cambiar la máquina.

Las palancas tienen costes distintos. Reducir el ancho de fila proyectando sólo columnas necesarias, filtrar antes del shuffle o sustituir una UDF por funciones nativas reduce trabajo. Reparticionar puede distribuir mejor la carga, mientras que workers con más memoria ayudan cuando cada partición es legítimamente grande. Aumentar `spark.sql.shuffle.partitions` sin medir puede crear miles de tareas diminutas.

from pyspark.sql import functions as F

daily_net = (
    events
    .where(F.col("event_date") >= F.date_sub(F.current_date(), 30))
    .select("event_date", "store_id", "net_amount")
    .groupBy("event_date", "store_id")
    .agg(F.sum("net_amount").alias("net_amount"))
)

daily_net.explain("formatted")

AQE usa estadísticas disponibles después de exchanges para cambiar una estrategia sort-merge a broadcast hash, combinar particiones post-shuffle demasiado pequeñas, dividir particiones sesgadas y propagar relaciones vacías. En Databricks está habilitado por defecto para consultas batch compatibles con exchanges o subconsultas. El plan adaptativo final puede diferir del plan inicial mostrado antes de ejecutar.

AQE no reordena dinámicamente todos los joins ni arregla un modelo de datos deficiente. Una relación que parece pequeña en catálogo puede superar el límite real, y determinados tipos de join no admiten broadcast en uno de sus lados. Usa `explain('formatted')`, el plan final de Spark UI y métricas de ejecución para demostrar qué regla se aplicó; evita copiar configuraciones antiguas que anulen los defaults optimizados.

print(spark.conf.get("spark.sql.adaptive.enabled"))

result = (
    facts.join(dimensions, "product_id")
    .groupBy("category")
    .count()
)

result.explain("formatted")
result.count()  # materializa el plan para revisarlo en Spark UI

Un broadcast hash join replica la relación pequeña y evita repartir la grande por la clave. Es excelente para una dimensión realmente pequeña, pero peligroso si las estadísticas están obsoletas o si el lado difundido crece: la transferencia y la tabla hash consumen memoria en cada executor. Los hints expresan una preferencia al optimizador, no corrigen una semántica de join incompatible.

Para joins grandes, el sort-merge distribuye ambos lados y paga shuffle y ordenación. Reduce primero las filas y columnas, conserva estadísticas y observa si hay claves calientes. En SQL y DataFrames, el criterio no es 'broadcast siempre es más rápido', sino coste total, compatibilidad del tipo de join y evidencia estable en ejecuciones representativas.

from pyspark.sql import functions as F

active_products = (
    spark.table("prod.ref.products")
    .where("is_active = true")
    .select("product_id", "category")
)

enriched = orders.join(F.broadcast(active_products), "product_id", "left")
enriched.explain("formatted")

Las funciones SQL y PySpark nativas permanecen visibles para Catalyst y pueden beneficiarse de Photon, generación de código, pushdown y simplificación de expresiones. Una UDF escalar de Python serializa datos entre JVM y Python, oculta parte de la lógica al optimizador y puede provocar fallback de Photon. Antes de crearla, busca funciones integradas para arrays, mapas, strings, fechas y tipos complejos.

Cuando la lógica no existe, una pandas UDF o APIs basadas en Arrow pueden procesar lotes y reducir el coste por fila, pero siguen requiriendo medición, tipos explícitos y pruebas de nulos. La optimización correcta incluye mantenibilidad: una expresión nativa legible suele ser más fácil de gobernar y portar que una UDF opaca.

from pyspark.sql import functions as F

normalized = customers.withColumn(
    "email_domain",
    F.lower(F.element_at(F.split(F.trim("email"), "@"), -1)),
).withColumn(
    "is_company_email",
    ~F.col("email_domain").isin("gmail.com", "outlook.com", "yahoo.com"),
)