Uma task do Spark roda por uma hora enquanto o resto termina em segundos. O que está acontecendo e como você resolve?

Isso é data skew: uma chave concentra uma fatia desproporcional das linhas, então uma única partição faz quase todo o trabalho depois de um shuffle. Confirme olhando a distribuição da chave de join ou de agrupamento. Resolva habilitando o tratamento de skew join do adaptive query execution, fazendo broadcast do lado pequeno se ele couber, aplicando salting na chave quente para espalhá-la por várias partições, ou filtrando e processando as chaves discrepantes separadamente.

Uma cauda longa em uma task quase sempre significa skew: depois do shuffle, uma chave caiu em uma partição e essa partição tem a maioria das linhas. Primeiro eu confirmo em vez de presumir, então um group by na chave de join com um count, e uma olhada no stage na Spark UI onde a duração máxima da task e o shuffle read são muito maiores que a mediana. Jogar executors no problema não faz nada, porque o trabalho não está distribuído. A solução depende do formato. Se o outro lado do join é pequeno, faça broadcast dele e pule o shuffle inteiro. Se ele é realmente grande dos dois lados, o adaptive query execution consegue dividir partições com skew automaticamente, e além disso eu aplico salting: acrescento um sufixo aleatório à chave quente, replico o lado pequeno por esses sufixos, faço o join e depois agrego. O que pega as pessoas são os nulos. Nós tínhamos uma chave de join que era nula em cerca de 40% das linhas, todas com o mesmo hash e na mesma partição, e filtrar os nulos antes do join resolveu de imediato.