Comment répartir uniformément l'ensemble de données pour éviter une jointure biaisée (et des tâches longues) ?

Question

Je suis en train d'écrire une application en utilisant l'API de dataset Spark sur le notebook databricks.

J'ai 2 tables. L'une contient 1,5 milliard de lignes et la seconde 2,5 millions. Les deux tables contiennent des données de télécommunication et la jointure est faite en utilisant le code pays et les 5 premiers chiffres d'un numéro. Le résultat comporte 55 milliards de lignes. Le problème est que j'ai des données déséquilibrées (tâches longues à s'exécuter). Peu importe comment je répartis le dataset, j'ai des tâches longues à s'exécuter en raison de la répartition inégale des clés hachées.

J'ai essayé d'utiliser des jointures de diffusion, j'ai essayé de persister les partitions de la table principale en mémoire, etc...

Quelles sont mes options ici?

Answer 1

Spark répartira les données en fonction de la clé de jointure, donc la répartition avant la jointure ne changera pas le déséquilibre (ajoutera simplement un shuffle inutile)

si vous connaissez la clé qui cause le déséquilibre (généralement quelque chose comme null ou 0 ou ""), divisez vos données en 2 parties - un ensemble de données avec la clé de déséquilibre, et un autre avec le reste

et effectuez la jointure sur les sous-ensembles de données, puis union des résultats

par exemple:

val df1 = ...
val df2 = ...
val skewKey = null

val df1Skew = df1.where($"key" === skewKey)
val df2Skew = df2.where($"key" === skewKey)

val df1NonSkew = df1.where($"key" =!= skewKey)
val df2NonSkew = df2.where($"key" =!= skewKey)

val dfSkew    = df1Skew.join(df2Skew) //c'est une jointure croisée
val dfNonSkew = df1NonSkew.join(df2NonSkew, "key")

val res = dfSkew.union(dfNonSkew)