apache-spark
Joint
Recherche…
Remarques
Une chose à noter est vos ressources par rapport à la taille des données que vous rejoignez. C'est là que votre code Spark Join peut échouer, ce qui entraîne des erreurs de mémoire. Pour cette raison, assurez-vous de bien configurer vos travaux Spark en fonction de la taille des données. Voici un exemple de configuration pour une jointure de 1,5 million à 200 millions.
Utiliser Spark-Shell
spark-shell --executor-memory 32G --num-executors 80 --driver-memory 10g --executor-cores 10
Utiliser Spark Submit
spark-submit --executor-memory 32G --num-executors 80 --driver-memory 10g --executor-cores 10 code.jar
Broadcast Hash Rejoignez Spark
Une jointure de diffusion copie les petites données vers les nœuds de travail, ce qui conduit à une jointure extrêmement efficace et extrêmement rapide. Lorsque nous rejoignons deux jeux de données et que l'un des jeux de données est beaucoup plus petit que l'autre (par exemple, lorsque le petit jeu de données peut tenir dans la mémoire), nous devons utiliser une jointure par hachage de diffusion.
L'image suivante visualise une jointure par hachage de diffusion lorsque le petit ensemble de données est diffusé sur chaque partition du grand ensemble de données.
Vous trouverez ci-dessous un exemple de code que vous pouvez facilement implémenter si vous avez un scénario similaire de jointure de jeu de données volumineux et petit.
case class SmallData(col1: String, col2:String, col3:String, col4:Int, col5:Int)
val small = sc.textFile("/datasource")
val df1 = sm_data.map(_.split("\\|")).map(attr => SmallData(attr(0).toString, attr(1).toString, attr(2).toString, attr(3).toInt, attr(4).toInt)).toDF()
val lg_data = sc.textFile("/datasource")
case class LargeData(col1: Int, col2: String, col3: Int)
val LargeDataFrame = lg_data.map(_.split("\\|")).map(attr => LargeData(attr(0).toInt, attr(2).toString, attr(3).toInt)).toDF()
val joinDF = LargeDataFrame.join(broadcast(smallDataFrame), "key")