Una tarea de Spark corre durante una hora mientras el resto termina en segundos. ¿Qué está pasando y cómo lo arreglas?

Eso es data skew: una clave concentra una proporción desmedida de las filas, así que una sola partición hace casi todo el trabajo después del shuffle. Confírmalo mirando la distribución de la clave de join o de agrupación. Arréglalo activando el manejo de skew join de adaptive query execution, haciendo broadcast del lado pequeño si cabe, salando la clave caliente entre varias particiones, o filtrando y procesando aparte las claves atípicas.

Una cola larga en una sola tarea casi siempre significa skew: después del shuffle, una clave cayó en una partición y esa partición tiene la mayoría de las filas. Primero lo confirmo en vez de asumirlo, así que hago un group by sobre la clave de join con un count, y miro la etapa en la Spark UI donde la duración máxima de tarea y el tamaño de lectura de shuffle aplastan a la mediana. Tirarle ejecutores no hace nada, porque el trabajo no está distribuido. El arreglo depende de la forma. Si el otro lado del join es pequeño, hazle broadcast y te saltas el shuffle por completo. Si de verdad es grande por ambos lados, adaptive query execution puede dividir automáticamente las particiones sesgadas, y más allá de eso salo: añado un sufijo aleatorio a la clave caliente, replico el lado pequeño entre esos valores de sal, hago el join y luego agrego. El que pilla a la gente es el de los nulos. Teníamos una clave de join que era nula en torno al 40% de las filas, todas las cuales hasheaban a la misma partición, y filtrar los nulos antes del join lo arregló de golpe.