Problèmes de mémoire Spark

Lorsque les exécuteurs n'ont plus de mémoire, Spark les supprime et tente à nouveau d'exécuter leurs tâches. Le job ralentit et, après plusieurs échecs, l'étape et le job échouent.

Utilisez cette page lorsque des jobs échouent ou que des exécuteurs sont supprimés et que le log des événements du compute ne montre pas de pertes d’instance spot ou préemptible ou d’autoscaling. Pour consulter le log des événements, voir Jobs échoués ou exécuteurs supprimés. Si l’étape se termine mais qu’elle est lente et montre du spill, voir Skew and spill.

Vérification d'un problème de mémoire

Les problèmes de mémoire causent souvent des erreurs telles que :

Output

SparkException: Job aborted due to stage failure: Task 3 in stage 0.0 failed 4 times, most recent failure: Lost task 3.3 in stage 0.0 (TID 30) (10.139.64.114 executor 4): ExecutorLostFailure (executor 4 exited caused by one of the running tasks) Reason: Remote RPC client disassociated. Likely due to containers exceeding thresholds, or network issues. Check driver logs for WARN messages.

Ces erreurs sont souvent génériques et d'autres problèmes peuvent les provoquer. Si vous suspectez un problème de mémoire, vérifiez-le en doublant la mémoire par cœur pour voir si cela résout le problème.

Par exemple, si le type de worker possède 4 cœurs et 16 Go de mémoire, passez à un type de worker avec 4 cœurs et 32 Go de mémoire. Cela vous donne 8 Go par cœur au lieu de 4 Go. Le ratio mémoire/cœurs compte, pas la mémoire totale. Si le job met plus de temps à échouer ou n'échoue pas du tout, la mémoire peut en être la cause.

Si une mémoire plus importante résout le problème, conservez le type de worker le plus grand. Si cela ne fonctionne pas, ou si le coût supplémentaire est trop élevé, vérifiez les causes possibles dans la section suivante.

Causes possibles

Les problèmes de mémoire ont de nombreuses causes possibles. Start par ces causes courantes :

• Trop peu de partitions de shuffle. Définissez spark.sql.shuffle.partitions sur auto afin que Spark choisisse le nombre de partitions de shuffle. Voir la configuration AQE.
• Une grande diffusion. Recherchez les indices de jointure de diffusion sur les grandes tables et vérifiez le threshold de diffusion d'exécution. Voir configuration AQE.
• UDF. Chaque tâche qui exécute une UDF peut charger toutes les données de sa partition en mémoire. Réécrivez l'UDF à l'aide de fonctions natives ou repartitionnez avant de l'appliquer. Voir UDF lentes.
• Une fonction de fenêtre sans PARTITION BY. Ajoutez une clause PARTITION BY afin que la fenêtre ne couvre pas tout le dataset. Voir Étape Spark longue avec une tâche.
• Asymétrie. Vérifiez l'asymétrie dans les métriques de synthèse de l'étape. Voir Asymétrie. Pour les jointures, voir Gérer dynamiquement la jointure par asymétrie.
• État du streaming. Pour les queries qui conservent un état de streaming, utilisez le magasin d'état RocksDB et plafonnez son utilisation de la mémoire. Voir Plafonner l'utilisation de la mémoire RocksDB.