Spark SQL diffusion de jointure de hachage

Je suis en train de réaliser une émission de jointure de hachage sur dataframes à l'aide de SparkSQL, comme indiqué ici: https://docs.cloud.databricks.com/docs/latest/databricks_guide/06%20Spark%20SQL%20%26%20DataFrames/05%20BroadcastHashJoin%20-%20scala.html

Dans cet exemple, la (petite) DataFrame est conservé par saveAsTable et puis, il y a une jointure via spark SQL (via sqlContext.sql("..."))

Le problème que j'ai c'est que j'ai besoin d'utiliser la sparkSQL API pour construire mes SQL (je suis de gauche se joindre à ~50 tables avec une liste d'ID, et ne veux pas écrire du SQL à la main).

How do I tell spark to use the broadcast hash join via the API?  The issue is that if I load the ID list (from the table persisted via `saveAsTable`) into a `DataFrame` to use in the join, it isn't clear to me if Spark can apply the broadcast hash join.

InformationsquelleAutor user1759848 | 2016-05-27

apache-spark apache-spark-sql

19

Vous pouvez marquer explicitement la DataFrame comme suffisamment petit pour la radiodiffusion
à l'aide de broadcast fonction:

Python:
```
from pyspark.sql.functions import broadcast

small_df = ...
large_df = ...

large_df.join(broadcast(small_df), ["foo"])
```
ou de diffusion de l'indice (Spark >= 2.2):
```
large_df.join(small_df.hint("broadcast"), ["foo"])
```
Scala:
```
import org.apache.spark.sql.functions.broadcast

val smallDF: DataFrame = ???
val largeDF: DataFrame = ???

largeDF.join(broadcast(smallDF), Seq("foo"))
```
ou de diffusion de l'indice (Spark >= 2.2):
```
largeDF.join(smallDF.hint("broadcast"), Seq("foo"))
```
SQL

Vous pouvez utiliser les indices (Spark >= 2.2):
```
SELECT /*+ MAPJOIN(small) */ * 
FROM large JOIN small
ON large.foo = small.foo
```
ou
```
SELECT /*+  BROADCASTJOIN(small) */ * 
FROM large JOIN small
ON large.foo = small.foo
```
ou
```
SELECT /*+ BROADCAST(small) */ * 
FROM large JOIN small
ON larger.foo = small.foo
```
R (SparkR):

Avec hint (Spark >= 2.2):
```
join(large, hint(small, "broadcast"), large$foo == small$foo)
```
Avec broadcast (Spark >= 2.3)
```
join(large, broadcast(small), large$foo == small$foo)
```
Note:

Diffusion de jointure est utile si l'une des structures est relativement faible. Sinon, il peut être beaucoup plus cher qu'un plein shuffle.
- Merci! Avec l'expérimentation, je vois que smallDF.join(largeDF) ne pas faire une émission de jointure de hachage, mais largeDF.join(smallDF) n'.
- est-il judicieux de faire smallDF.join(LargeDF, "right_outer") alors?
- peut encore être diffusé automatiquement en fonction spark.sql.autoBroadcastJoinThreshold et broadcast fonction peut être appliquée sur n'importe quelle position: broadcast(largeDF).join(smallDF, Seq("foo"))
- Est il possible que je peux passer de diffusion des allusions à l'aide de sql api tout en faisant trois tables qui sont de deux petites et une grande table. Merci.!
InformationsquelleAutor zero323
5
```
jon_rdd = sqlContext.sql( "select * from people_in_india  p
                            join states s
                            on p.state = s.name")


jon_rdd.toDebugString() /join_rdd.explain() : 
```
shuffledHashJoin :

toutes les données pour l'Inde sera mélangée dans seulement 29 touches pour chacun des états.
Problèmes:
inégale de la fragmentation.
Limitée parallélisme avec 29 sortie des partitions.

broadcaseHashJoin:

diffusion de la petite RDD à tous les nœuds du travailleur.
le parallélisme de la grande rdd est toujours maintenu et shuffle n'est même pas
nécessaire.

PS: l'Image peut laid, mais instructif.

InformationsquelleAutor Vamshavardhan Reddy
2

Avec une diffusion joindre un côté de la jointure équation est matérialisé et de l'envoyer à tous les utilisateurs. Il est donc considéré comme une carte-jointure côté.

Que l'ensemble de données est prise en matérialisée et de les envoyer sur le réseau, il ne fait qu'apporter une amélioration significative des performances, si considérables petit.

Donc, si vous essayez d'effectuer smallDF.join(largeDF)

Attendre..!!! une autre contrainte est qu'il faut également tenir entièrement dans la mémoire de chaque interprète.Il doit également s'insérer dans la mémoire du Pilote!

De diffusion variables sont partagées entre les exécuteurs en utilisant le protocole Torrent je.e.Peer-to-Peer protocole et l'avantage du protocole Torrent est que les pairs, le partage de blocs d'un fichier parmi les uns des autres en ne s'appuyant pas sur une entité centrale chargée de la tenue de tous les blocs.

Exemple mentionné ci-dessus est suffisante pour commencer à jouer de la diffusion de la rejoindre.

Remarque:
Ne peut pas modifier la valeur après la création.
Si vous essayez, le changement va être sur une seule&nœud

InformationsquelleAutor Nayan Sharma

Vous devez vous connecter pour publier un commentaire.