Entrevista de ingeniería de datos: ¿Cómo diagnosticar y solucionar el sesgo de datos (data skew) en Spark?

El escenario y cuándo aplica

Un trabajo diario de Spark SQL realiza un left join entre una tabla de hechos de eventos de 4.8 TB y una dimensión de productos de 180 GB sobre product_id. Tras un despliegue aguas arriba, el 32% de los eventos se normalizan a product_id='UNKNOWN'. El trabajo utiliza 2,000 particiones de shuffle. En la etapa del join, la Spark UI muestra una mediana de lectura de shuffle por tarea de 1.1 GiB, mientras que una tarea lee 720 GiB, realiza spill a disco repetidamente y eventualmente falla con OOM tras varios reintentos. El tiempo de ejecución ha aumentado de 24 a 96 minutos. El negocio necesita que el trabajo termine en menos de 45 minutos, sin descartar los eventos con productos desconocidos ni modificar el resultado del left join.

Los tamaños de tabla, la proporción de la clave, las métricas de partición, los tiempos de ejecución y el SLA son supuestos de la entrevista. La tarea central es distinguir el sesgo de datos (data skew) de la falta de recursos y de la explosión de salida del join mediante evidencia a nivel de partición, para luego elegir una solución que preserve el contrato de datos. Esto pertenece a la categoría de datos porque evalúa planes de ejecución de Spark, particiones de shuffle, distribución de datos y corrección en procesamiento por lotes (batch). La pregunta existente sobre particiones calientes en Kafka se enfoca en claves de mensajes, ordenamiento y offsets de consumidores. Esta pregunta se enfoca en particiones SQL en tiempo de ejecución, estrategia de joins, AQE y conservación de resultados, por lo que la capa de falla y el método de validación son diferentes.

El mismo razonamiento aplica a groupBy, distinct, funciones de ventana y otras dependencias amplias (wide dependencies). Cuando muchos registros para una clave convergen en unas pocas tareas posteriores al shuffle, pueden generar tareas rezagadas (stragglers), spill, presión de GC u OOM. Una buena respuesta no comienza añadiendo memoria. Primero demuestra si la tarea más lenta posee una cantidad desproporcionada de datos y cómputo.

Lo que evalúa el entrevistador

Comienza con la granularidad de la evidencia. Una respuesta sólida avanza desde el trabajo hacia la consulta SQL, luego a una etapa específica y sus tareas individuales. Compara la duración, los registros y bytes leídos en shuffle, el spill, el pico de memoria de ejecución y el tiempo de GC. Una tarea que lee cientos de veces la mediana y sigue siendo lenta cuando se reintenta en otro executor respalda un sesgo de datos determinista. La memoria agregada del executor y el tiempo total de ejecución por sí solos no establecen dicha causa.

El dominio del plan de ejecución importa de igual manera. EXPLAIN FORMATTED confirma el Exchange, el tipo de join y el join físico. EXPLAIN COST y las estadísticas de tiempo de ejecución en la SQL UI exponen los tamaños de datos estimados y observados. Spark AQE utiliza estadísticas en tiempo de ejecución para ajustar el plan. En Spark 4.2.0, la optimización de skew join puede dividir una partición sesgada en un sort-merge join y replicar el lado menor cuando sea necesario. Los valores predeterminados documentados marcan una partición como sesgada solo cuando supera tanto cinco veces la mediana como 256 MiB. El candidato debe inspeccionar la configuración efectiva del entorno, ya que los valores por defecto de la documentación no son hechos inmutables del clúster.

La semántica de los datos se convierte en la línea divisoria. UNKNOWN puede ser un evento "no atribuido" válido o un defecto aguas arriba. Descartar, distribuir aleatoriamente o reescribir esas filas puede alterar el resultado. Un salted join debe replicar únicamente las filas de claves calientes de la dimensión y asignar una sal determinista a las filas de hechos calientes. Replicar toda la dimensión multiplica el volumen de datos, mientras que aleatorizar ambos lados de forma independiente pierde coincidencias.

La selección del remedio revela otro nivel de profundidad. Aumentar spark.sql.shuffle.partitions crea más buckets de hash, pero todas las filas de una clave caliente seguirán cayendo en el mismo bucket. Broadcast es adecuado solo cuando la proyección, el filtrado y estadísticas confiables demuestran que un lado cabe de manera segura en cada executor; forzar un broadcast de una dimensión de 180 GB no es seguro. AQE es la primera opción de baja intrusión. El salting explícito se adapta a claves calientes estables cuando AQE no se activa o aún no cumple el SLA. Las agregaciones suelen utilizar una agregación parcial con sal seguida de una segunda fusión (merge).

Una validación completa cierra la respuesta. Una mejora de rendimiento también debe demostrar que el conteo de filas, los importes de negocio, las claves desconocidas, las tasas de no coincidencia y las tasas de duplicados no han cambiado. Reducir el tiempo de ejecución de 96 a 40 minutos no demuestra la corrección ni muestra si la solución sobrevivirá a una distribución de claves diferente el día de mañana.

Preguntas para clarificar antes de responder

• ¿El cuello de botella está en el scan, en el shuffle write o después del shuffle read? Tareas de scan desiguales pueden deberse a archivos enormes o no divisibles. Este escenario apunta a la ruta de sesgo de clave porque el valor atípico aparece después del shuffle del join.
• ¿El 32% se refiere a registros, bytes comprimidos o costo de procesamiento? Filas anchas, UDFs costosas y la dispersión en la salida pueden generar sesgo de costo incluso cuando el conteo de filas parece modesto. Compara registros, bytes, tiempo y filas de salida.
• ¿Qué significa UNKNOWN para el negocio? Si los productos desconocidos no requieren atributos de dimensión, sepáralos del join principal y rellena los atributos nulos bajo el contrato original. Si deben coincidir con una fila centinela en la dimensión, preserva el join y divide el punto caliente.
• ¿Es product_id único en la dimensión de productos? Varias filas de dimensión con UNKNOWN convierten la clave de hechos caliente en una explosión de salida muchos a muchos. Separa la mala cardinalidad del sesgo de partición antes de optimizar.
• ¿Qué versión de Spark y configuraciones de AQE están efectivas? Revisa la página de Environment y el plan adaptativo final para verificar switches, umbrales, tipo de join y evidencia de que la división por sesgo realmente se ejecutó.
• ¿180 GB es la dimensión original o su entrada proyectada para el join? Si filtrar a la clave y dos atributos la hace apta para broadcast de forma segura, broadcast puede superar a un shuffle bilateral. Las estadísticas en tiempo de ejecución y los presupuestos de memoria del executor deben demostrarlo.
• ¿Qué invariantes definen un resultado equivalente? Como mínimo, especifica filas totales, eventos únicos, métricas de negocio aditivas, filas con claves desconocidas, filas sin coincidencia y la semántica de duplicados permitida.
• ¿Las claves calientes son estables y enumerables? Unas pocas claves estables se adaptan a un salting dirigido. Una cola larga y cambiante favorece AQE, detección dinámica de claves calientes o una corrección semántica aguas arriba.

Marco de respuesta en 30 segundos

"Compararía la lectura de shuffle, el spill, el GC y la ubicación de reintentos de las tareas del join en la SQL UI. Una tarea de 720 GiB frente a una mediana de 1.1 GiB, más un 32% de UNKNOWN, respalda un sesgo de claves; también verificaría la unicidad de la dimensión para descartar una explosión de salida. Primero confirmaría que el skew join de AQE divida la partición grande. Si el tiempo de ejecución sigue superando los 45 minutos, aplicaría salting a UNKNOWN de forma determinista mediante un event_id estable y replicaría solo su fila centinela. Finalmente, conciliaría filas e importes con la línea base, comparando luego el ratio máximo/mediana de entrada por tarea, el tiempo de etapa y el costo."

Respuesta detallada paso a paso

Paso 1: Atribuir los 96 minutos a una etapa y tarea

Conserva el event log y usa el Spark History Server para comparar una ejecución normal y una con regresión para la misma fecha de datos. Sigue la consulta SQL hasta los detalles de su etapa. Registra los percentiles y la duración máxima de tareas, alineados con los registros y bytes leídos en shuffle, el shuffle spill, el pico de memoria de ejecución, el tiempo de GC, el motivo de falla y el executor. En el plan SQL, busca un Exchange hashpartitioning(product_id, 2000) antes del left join, identifica el join físico final y determina si el plan adaptativo se completó.

En este caso, la misma tarea sigue leyendo cerca de 720 GiB tras reintentarse en otro executor, mientras que la mayoría lee alrededor de 1.1 GiB. Eso apunta a la partición de entrada en sí. Si una tarea lenta tiene una entrada ordinaria y alto GC o espera de disco en un executor, investiga el nodo primero. Si todas las tareas hacen spill uniformemente, enfócate en el dimensionamiento general de particiones y en los presupuestos de recursos. Si las filas de salida se multiplican repentinamente, inspecciona duplicados en la dimensión y la condición del join.

Paso 2: Demostrar la causa con distribución de claves y cardinalidad del join

Mide las claves calientes tras aplicar exactamente los mismos filtros y normalización de claves usados en producción. Inspeccionar la columna sin procesar es insuficiente porque trim, normalización de mayúsculas/minúsculas, coalesce o una UDF pueden colapsar varios valores en una sola clave. En una tabla muy grande, usa estadísticas existentes, un muestreo controlado o una agregación acotada para que el diagnóstico no se convierta en otro trabajo pesado. La siguiente consulta asume que la tabla de hechos ya contiene o precalcula payload_bytes, permitiendo medir el sesgo tanto en filas como en bytes. Sin esa columna, usa estadísticas de almacenamiento o una estimación controlada de serialización. La consulta expresa la contabilidad requerida:

SELECT
  COALESCE(product_id, '<NULL>') AS join_key,
  COUNT(*) AS row_count,
  SUM(payload_bytes) AS payload_bytes
FROM fact_events
WHERE event_date = DATE '2026-07-17'
GROUP BY COALESCE(product_id, '<NULL>')
ORDER BY row_count DESC
LIMIT 20;

Demuestra también que cada product_id aparece como máximo una vez en la dimensión y compara los conteos de eventos de hechos antes y después del join. Para este left join, una clave de dimensión única significa que cada evento de hechos produce exactamente una fila, incluyendo los eventos no coincidentes. Si UNKNOWN concentra el 32% de las filas de hechos, la dimensión tiene una sola fila centinela y la salida no se multiplica, el particionamiento por hash explica la partición única de 720 GiB.

Paso 3: Abordar la causa semántica antes de elegir una técnica de ejecución

Investiga por qué el despliegue aguas arriba asigna el 32% de los eventos a UNKNOWN. Si es una regresión, revierte o corrige el mapeo y vuelve a ejecutar las particiones afectadas. Eso restaura tanto la calidad de datos como el rendimiento. Si es un valor de negocio válido, la capa de ejecución debe dar soporte a esa distribución.

Los eventos desconocidos válidos que no necesitan atributos de producto pueden tomar una ruta separada: unir las claves frías normalmente, rellenar los atributos de dimensión con nulos para la ruta desconocida bajo el contrato existente y combinar con unionByName. Eso elimina un shuffle sin pérdida de información. Si UNKNOWN debe coincidir con atributos centinela, conserva el join y divide el trabajo con AQE o salting dirigido. La semántica de los resultados decide la rama; las particiones uniformes no son un permiso para descartar datos.

Paso 4: Elegir el remedio menos costoso que funcione

Revisa AQE primero. En Spark 4.2.0, spark.sql.adaptive.enabled y spark.sql.adaptive.skewJoin.enabled están habilitados por defecto, pero un clúster, trabajo o plataforma administrada puede anularlos. Tanto el factor de skew-join como los umbrales absolutos de bytes deben cumplirse. Inspecciona el plan adaptativo final en busca del manejo de sesgo y confirma que las métricas de la etapa muestren la división de la partición grande. Para un sort-merge join compatible, AQE puede dividir el lado grande y replicar el lado menor. Se adapta a distribuciones diarias cambiantes, pero puede añadir costo de shuffle y replicación, y no puede reparar un join muchos a muchos lógicamente incorrecto.

Si la dimensión proyectada tiene estadísticas confiables y es verdaderamente pequeña, evalúa un broadcast hash join para que el lado de los hechos no realice shuffle por la clave de join. Los 180 GB originales quedan muy fuera de un presupuesto de broadcast ordinario. Un hint forzado puede agotar cada executor. Evalúa los bytes proyectados, tareas concurrentes, heap del executor y timeout de broadcast en conjunto.

Cuando AQE no se active o aún no cumpla el SLA, aplica salting manual a las claves calientes estables. El siguiente código asume que event_id es estable y único, product_id es único en la dimensión y solo UNKNOWN necesita dividirse. Las filas de hechos calientes se asignan deterministamente a 32 sales. Solo la fila centinela coincidente de la dimensión se replica 32 veces. Las claves frías permanecen en la sal 0, por lo que la dimensión completa nunca se multiplica.

from pyspark.sql import functions as F

SALT_BUCKETS = 32
HOT_KEYS = ["UNKNOWN"]

events_salted = events.withColumn(
    "salt",
    F.when(
        F.col("product_id").isin(*HOT_KEYS),
        F.pmod(F.xxhash64("event_id"), F.lit(SALT_BUCKETS)).cast("int"),
    ).otherwise(F.lit(0)),
)

salt_values = spark.range(SALT_BUCKETS).select(
    F.col("id").cast("int").alias("salt")
)

products_hot = (
    products.filter(F.col("product_id").isin(*HOT_KEYS))
    .crossJoin(salt_values)
)
products_cold = (
    products.filter(~F.col("product_id").isin(*HOT_KEYS))
    .withColumn("salt", F.lit(0))
)
products_salted = products_cold.unionByName(products_hot)

result = (
    events_salted.join(products_salted, ["product_id", "salt"], "left")
    .drop("salt")
)

Treinta y dos es un candidato inicial en este escenario de entrevista. Deriva el número de buckets a partir de los bytes de la partición caliente, el tamaño objetivo de la tarea, el paralelismo disponible y el costo de replicación del lado menor, probando luego en datos representativos. Muy pocos buckets conservan una cola larga. Demasiados añaden sobrecarga de planificación, archivos y replicación. Simplemente ejecutar repartition(4000, "product_id") seguirá colocando cada fila de UNKNOWN en una sola partición.

Para groupBy(product_id), generalmente no hay una dimensión que replicar. Primero agrega parcialmente por (product_id, salt), luego fusiona los resultados parciales por product_id. Las operaciones que pueden descomponerse de forma segura con fusiones asociativas y conmutativas, como sum, count, min y max, se adaptan a esta técnica. Un median exacto, agregaciones dependientes del orden y estados de UDF no fusionables requieren un algoritmo diferente.

Paso 5: Integrar corrección, rendimiento y costo en una sola compuerta de aceptación

Ejecuta la línea base y la solución candidata sobre una instantánea de entrada inmutable. La corrección va primero: compara el total de filas de salida, event_id únicos, filas con UNKNOWN, filas sin coincidencia y sumas y conteos de negocio por dimensiones significativas. Realiza diffs a nivel de fila para claves calientes, claves frías, nulos y claves de dimensión duplicadas. La garantía de que un evento de hechos produce una fila en el left join depende de la unicidad de la clave de dimensión, por lo que debes monitorear esa restricción por separado.

Para el rendimiento, compara p50, p95 y la duración máxima de tareas en la etapa del join, el ratio máximo/mediana de shuffle read, spill, GC, OOM, reintentos de tareas, tiempo de etapa y tiempo total de ejecución. Para el costo, registra horas-executor, bytes de shuffle y cantidad de archivos de salida. Terminar dentro de los 45 minutos es solo un criterio. Una ejecución que cumple el SLA duplicando el shuffle, alterando resultados o fallando ante un nuevo punto caliente al día siguiente no es aceptable.

Despliega a producción reproduciendo una fecha histórica y luego ejecutando en paralelo (shadowing) una nueva fecha de datos para comparar resultados. Monitorea el conjunto de claves calientes, el ratio de entrada máximo/mediana por tarea y la proporción de claves desconocidas. El salto al 32% de UNKNOWN tras un despliegue aguas arriba también debería activar una alerta de calidad de datos, exponiendo la regresión semántica antes de que retrase el trabajo.

Respuesta de muestra de alta calidad

"Primero atribuiría la regresión a una etapa de join específica en la SQL UI. La evidencia actual sugiere fuertemente sesgo: a través de 2,000 tareas, la mediana de lectura de shuffle es de 1.1 GiB, una tarea lee 720 GiB y esa tarea sigue siendo lenta tras moverse a otro executor. Inspeccionaría el plan adaptativo final, el spill y el GC, para luego perfilar las claves tras la normalización exacta de producción. También validaría la unicidad de la clave de dimensión. Múltiples filas de dimensión con UNKNOWN significarían que el síntoma incluye una explosión de salida en el join.

Asumiendo una dimensión única y un 32% de filas de hechos con UNKNOWN, un único bucket de hash-shuffle explica la tarea rezagada. Más particiones crean buckets adicionales pero no dividen esa clave, y más memoria de executor solo pospone el OOM. Primero determinaría si el mapeo aguas arriba es una regresión. Si las filas desconocidas no necesitan atributos de dimensión, las separaría del join y rellenaría los atributos nulos con la semántica original del left join. Si deben coincidir con una fila centinela, verificaría que el skew join de AQE aparezca efectivamente en el plan final, ya que puede dividir una partición de sort-merge join sesgada usando estadísticas en tiempo de ejecución y replicar el lado pequeño.

Si AQE aún deja el tiempo de ejecución por encima de 45 minutos, usaría salting dirigido. Un event_id estable asignaría cada fila de hechos de UNKNOWN de forma determinista a, por ejemplo, una de 32 sales. Replicaría únicamente la fila UNKNOWN de la dimensión a lo largo de esas 32 sales; cada clave fría usaría la sal 0. Cada evento sigue coincidiendo con una fila de dimensión, mientras que varias tareas se reparten el trabajo de la clave caliente. Derivaría el conteo final de buckets a partir de los bytes del hotspot y el tamaño objetivo de la tarea, en lugar de fijar 32 sin mediciones.

Para la validación, compararía la misma entrada inmutable contra la línea base. Las filas totales, eventos únicos, registros desconocidos y no coincidentes, y las sumas de negocio deben coincidir. Luego compararía el ratio máximo/mediana de shuffle read, la cola de duración de tareas, spill, OOM, tiempo de etapa, horas-executor y archivos de salida. Finalmente, reproduciría una fecha histórica, ejecutaría en paralelo una fecha nueva y alertaría sobre la proporción de claves desconocidas y nuevos hotspots. Eso demuestra que el trabajo cumple con los 45 minutos, preserva los resultados y se mantiene observable cuando la distribución aguas arriba vuelva a cambiar."

Errores comunes

• Aumentar las particiones de shuffle de 2,000 a 8,000 inmediatamente → una clave caliente sigue yendo por hash a una sola partición mientras otras tareas se vuelven más pequeñas → mide la distribución de claves, luego divide la clave con AQE, ramificación semántica o salting dirigido.
• Solo aumentar la memoria del executor → eleva la tolerancia de una tarea pero deja intactos los 720 GiB de trabajo y la cola larga → reduce primero la carga de trabajo de la partición máxima, luego dimensiona los recursos según las tareas medidas.
• Declarar sesgo tras ver una sola tarea lenta → un nodo defectuoso, GC, lecturas remotas o una UDF lenta también pueden crear un straggler → compara la entrada de la tarea, la ubicación del reintento, el spill, el GC y el plan de ejecución.
• Aplicar salting aleatorio a los hechos y a la dimensión de forma independiente → las sales no coincidirán y se perderán resultados del join, mientras que los reintentos pueden volverse no deterministas → deriva la sal de los hechos a partir de un ID de fila estable y enumera las mismas sales en la dimensión.
• Replicar la dimensión completa 32 veces → una dimensión de 180 GB genera un costo enorme de red y memoria → replica únicamente las filas confirmadas como claves calientes y mantén las claves frías en sal 0.
• Forzar el broadcast de la dimensión de 180 GB → cada executor debe almacenar los datos difundidos y puede fallar por OOM → proyecta y mide primero; usa broadcast solo después de que los presupuestos de memoria y concurrencia demuestren que es seguro.
• Filtrar UNKNOWN para acelerar el trabajo → la semántica de salida y las métricas aguas abajo cambian → establece el contrato de eventos desconocidos y preserva los resultados del left join incluso al ramificar.
• Comparar solo el tiempo total de ejecución → una aparente aceleración puede provenir de datos descartados, duplicados o mal calculados → demuestra las invariantes de filas y de negocio antes de comparar la distribución de tareas, el costo y el SLA.

Preguntas de seguimiento y cómo responderlas

AQE está habilitado. ¿Por qué no se dividió el join sesgado?

Inspecciona el plan adaptativo final y las configuraciones efectivas para spark.sql.adaptive.enabled, el switch de skew-join, el umbral de factor sobre la mediana y el umbral absoluto de bytes. Ambos umbrales deben cumplirse. Confirma que el join físico siga una ruta compatible con AQE, que las estadísticas en tiempo de ejecución estén disponibles y que un hint o anulación de la plataforma no limite el plan. Prueba cambios de umbrales u optimización forzada de sesgo en datos representativos midiendo el shuffle adicional. Si el plan no puede beneficiarse, usa salting dirigido. Un valor true en un archivo de configuración no demuestra que el plan ejecutado haya dividido la partición.

Si la proyección reduce la dimensión a 6 GiB, ¿se puede aplicar broadcast?

Seis GiB aún requieren una evaluación del heap del executor, tareas concurrentes, tamaño serializado, timeout de broadcast y estabilidad del clúster. Broadcast puede eliminar el shuffle por la clave de join en el lado grande y así evitar el hotspot, pero también copia la dimensión a los executors. Usa estadísticas para demostrar los bytes reales, luego observa el pico de memoria y GC en una prueba de carga con forma de producción antes de añadir un hint. "Mucho más pequeña que la tabla de hechos" no es un criterio de broadcast suficiente.

¿Qué pasa si las claves calientes cambian a diario y HOT_KEYS no se puede mantener manualmente?

Prefiere la respuesta en tiempo de ejecución de AQE. Si el salting explícito sigue siendo necesario, genera una tabla acotada de claves calientes antes del trabajo principal usando umbrales de registros, bytes o costo, y luego difunde (broadcast) esa pequeña tabla para seleccionar la ruta con sal. Versiona la lista por fecha de datos y asígnale un umbral, límite de conteo y mecanismo de fallback. Esto añade una etapa de planificación y estado operativo, por lo que las ganancias estables en el SLA deben justificar la complejidad.

Si la operación lenta es groupBy, ¿se sigue replicando una dimensión?

No. Para una agregación fusionable, aplica sal a las filas calientes, calcula agregados parciales por (key, salt) y luego fusiona esos parciales por key. Eso distribuye la entrada de una clave caliente entre tareas, mientras que la segunda etapa procesa un número pequeño de parciales. Especifica si el agregado es fusionable de forma segura. Estados ordenados globalmente o no fusionables no pueden usar esta técnica sin un algoritmo diferente.

¿Cómo eliges 32 buckets de sal?

Divide los bytes de la partición caliente entre la entrada objetivo por tarea para obtener un límite inferior, y luego considera los núcleos disponibles, la replicación del lado pequeño, la sobrecarga del scheduler y las restricciones de archivos de salida. Si 720 GiB deben reducirse a unos 32 GiB por tarea, el límite inferior teórico es aproximadamente 23, por lo que 32 es un experimento razonable en este escenario. Compara varias opciones en cuanto a entrada máxima por tarea, tiempo de etapa y horas-executor, manteniendo un margen limitado para el crecimiento del hotspot.

¿Puede la ejecución especulativa solucionar este straggler?

Un duplicado de una tarea sesgada determinista seguirá leyendo la misma partición de 720 GiB, repitiendo generalmente un trabajo costoso en dos executors. La especulación es más útil para un nodo intermitentemente lento o fluctuaciones transitorias. Verifica si la misma tarea sigue siendo lenta tras reintentarse en otro lugar. Si su entrada sigue siendo el valor atípico, divide el trabajo de datos en lugar de duplicarlo.