Particionamiento, shuffle y datos sesgados en Spark
Un DataFrame se divide en particiones: bloques de filas que los ejecutores procesan en paralelo. El número de particiones fija el grado de paralelismo. Mientras los datos caben dentro de una partición y se procesan donde están, todo va rápido; el problema aparece cuando una operación necesita reagrupar filas por clave y estas viven en nodos distintos.
Qué es una partición
La partición es la unidad de trabajo en paralelo: una tarea por partición. Muy pocas particiones dejan núcleos ociosos; demasiadas generan un overhead de planificación que ahoga al scheduler. El valor por defecto de spark.sql.shuffle.partitions (200) es un punto de partida, no un acierto garantizado.
- Cada partición se procesa en una tarea, idealmente en el nodo donde ya residen sus datos.
- El objetivo es mover cómputo a los datos, no los datos al cómputo.
- El tamaño equilibrado de partición evita tareas rezagadas (stragglers).
El shuffle: cuando los datos cambian de nodo
Operaciones como groupBy, join o repartition rompen la localidad: para reunir todas las filas de una misma clave, Spark debe barajar (shuffle) datos entre ejecutores, escribiendo salidas a disco, transfiriéndolas por red y volviendo a particionarlas por clave.
Por qué el shuffle cuesta tanto
- Escribe y lee datos en disco en cada extremo.
- Satura la red del clúster al mover tablas completas.
- Serializa y deserializa objetos entre procesos.
- Ordena y agrupa, añadiendo CPU que ya no está donde estaban los datos.
Datos sesgados (data skew)
El skew ocurre cuando una clave concentra muchísimas más filas que las demás: una sola partición de salida se vuelve enorme mientras el resto termina rápido. Esa tarea lentísima es el straggler que define el tiempo total del job, y suele terminar en memoria insuficiente (OOM) o volcado a disco (spill).
| Síntoma | Causa probable | Remediación |
|---|---|---|
| Una tarea del stage dura 100x más | Clave con skew extremo (o nulas) | Salted key o AQE skew join |
| Executor muere por OOM | Partición de salida gigantesca | Aumentar particiones, aislar la clave |
| Spill a disco constante | Poca memoria para el agregado | Memorizar/optimizar, filtrar antes |
| Job lento sin skew claro | Demasiadas o muy pocas particiones | Ajustar shuffle.partitions, coalesce |
Estrategias para evitar o aliviar el shuffle
- Broadcast join: para una tabla pequeña, repártela completa a cada nodo y elimina el shuffle del join.
- Salted keys: añade un prefijo aleatorio a la clave sesgada para dispersarla en varias particiones.
- Filtra y proyecta temprano: menos filas y columnas antes del groupBy o join.
- Cache/persist: guarda resultados intermedios que se reutilizan para no recalcular el shuffle.
- AQE (Adaptive Query Execution): en Spark 3 detecta skew y ajusta join y particiones en tiempo de ejecución.
Ajustar el paralelismo: repartition vs coalesce
repartition(n) redistribuye por completo (hace shuffle) y sirve para aumentar o balancear particiones. coalesce(n) reduce el número de particiones sin shuffle completo, ideal para juntar muchas particiones pequeñas antes de escribir. Usar repartition donde bastaba coalesce añade un shuffle costoso e innecesario.
Errores frecuentes
- Dejar las 200 particiones por defecto en datos de gigabytes o de megabytes, sin medir.
- Joins con claves nulas que caen todas en la misma partición.
- Reintentar agrandando memoria en vez de atacar el skew de la clave.
Ejemplo resuelto: join sesgado por clave nula
Un join de millones de filas se atasca en una sola tarea. Diagnóstico y corrección por etapas:
- Inspeccionar la UI: un task del stage de join domina la duración.
- Confirmar el skew: contar filas por clave y detectar una dominante (o nulas).
- Separar las claves nulas y procesarlas aparte o rellenarlas con un prefijo aleatorio.
- Aplicar salted key o habilitar AQE skew join sobre la clave grande.
- Repetir la medición y verificar que el straggler desapareció.
¿Para qué sirve en la realidad?
La mayoría de los jobs de producción que "se ponen lentos de repente" no cambian de datos: cambian de distribución. Reconocer el shuffle y el skew permite arreglar el pipeline sin tirar de presupuesto en más clúster.
¿Cuál es normalmente la operación más costosa en un job distribuido de Spark?
Usar coalesce(n) para reducir particiones suele ser más barato que repartition(n) porque evita un shuffle completo.
Cuando una sola clave concentra la mayoría de las filas y vuelve una tarea un straggler, decimos que hay sesgo de (skew).
Clasifica cada medida según si reduce o provoca tráfico de shuffle.
Reduce shuffle
Provoca shuffle
Une cada término con su definición.
Shuffle y skew en Spark
Rendimiento distribuido: partición, shuffle y skew
# Spark a escala ## Particion ### unidad de paralelismo ### equilibrio y tamano ## Shuffle ### groupBy y join ### disco + red + orden ## Skew ### clave dominante ### straggler y OOM ## Mitigacion ### broadcast join ### salted key ### AQE y cache
Spark SQL: tuning de rendimiento (documentación oficial)
Referencia oficial de particionado, broadcast, AQE y configuración de shuffle.
Adaptive Query Execution en Spark 3 (Databricks)
Cómo Spark detecta skew y reoptimiza joins y particiones en tiempo de ejecución.
Tienes un job de join que antes corría en 10 minutos y ahora tarda 3 horas sin que haya crecido el volumen total. ¿Qué revisarías primero en la UI de Spark y qué hipótesis sobre la distribución de claves pondrías a prueba antes de tocar memoria o particiones?