Convertir le Tableau[(String,String)] type de RDD[(String,String)] type spark

Je suis nouvelle étincelle.

Voici mon code:

val Data = sc.parallelize(List(
      ("I", "India"), 
      ("U", "USA"), 
      ("W", "West"))) 

val DataArray = sc.broadcast(Data.collect)

val FinalData = DataArray.value

Ici FinalData est de Array[(String, String)] type.
Mais je veux données dans le formulaire de RDD[(String, String)] type.

Puis-je convertir des FinalData à RDD[(String, String)] type.

Plus De Détail:

Je veux rejoindre les Deux RDD, de Sorte à optimiser la condition de jointure(Pour le point de vue des performances)
Je suis de radiodiffusion petit RDD à tous cluster de sorte que les données de brassage sera de moins en moins.(Et indirectement la performance aurez amélioré)
Donc, pour tout cela, j'ai écrit quelque chose comme ceci:

//Big Data
val FirstRDD = sc.parallelize(List(****Data of first table****))

//Small Data
val SecondRDD = sc.parallelize(List(****Data of Second table****))

Donc sans hésiter je vais diffuser Petit jeu de Données(moyens SecondRDD)

val DataArray = sc.broadcast(SecondRDD.collect)

val FinalData = DataArray.value

//Ici, il va donner d'erreur que

val Join = FirstRDD.leftOuterJoin(FinalData)

Trouvé Tableau requis RDD

C'est pourquoi je suis à la recherche pour le Tableau de RDD de conversion.

InformationsquelleAutor Darshan | 2016-09-21

apache-spark scala

Les émissions sont en effet très utile pour améliorer les performances d'une JOINTURE entre un grand RDD et une plus petite. Lorsque vous faites cela, la diffusion (avec map ou mapPartitions) remplace le rejoindre, il n'est pas utilisé dans une jointure, et, par conséquent, en aucun cas, vous aurez besoin de "transformer une diffusion dans un EDR".

Voici à quoi il devrait ressembler:

val largeRDD = sc.parallelize(List(
  ("I", "India"),
  ("U", "USA"),
  ("W", "West")))

val smallRDD = sc.parallelize(List(
  ("I", 34),
  ("U", 45)))

val smaller = sc.broadcast(smallRDD.collectAsMap())

//using "smaller.value" inside the function passed to RDD.map ->
//on executor side. Broadcast made sure it's copied to each executor (once!)
val joinResult = largeRDD.map { case (k, v) => (k, v, smaller.value.get(k)) }

joinResult.foreach(println)
//prints:
//(I,India,Some(34))
//(W,West,None)
//(U,USA,Some(45))

Voir une solution similaire (à l'aide de mapPartitions), qui pourrait être plus efficace ici.

InformationsquelleAutor Tzach Zohar

3

Le réel problème, c'est que vous êtes la création d'un Broadcast variable, par la collecte de l' RDD (notez que cette action convertit le RDD dans un Array). Donc, ce que je dis, c'est que vous avez déjà un RDD, qui est Data, et cette variable a exactement les mêmes valeurs que FinalData, mais dans la forme que vous voulez RDD[(String, String)].

Vous pouvez vérifier cela dans la sortie suivante.
```
Data: org.apache.spark.rdd.RDD[(String, String)] = ParallelCollectionRDD[2] at parallelize at <console>:32
DataArray: org.apache.spark.broadcast.Broadcast[Array[(String, String)]] = Broadcast(1)
FinalData: Array[(String, String)] = Array((I,India), (U,USA), (W,West))
```
Bien, je ne comprends pas votre démarche, Vous avez juste besoin de paralléliser le Broadcast's valeur.
```
//You already have this data stored in `Data`, so it's useless repeat this process.
val DataCopy = sc.parallelize(DataArray.value)
```
MODIFIER

Après la lecture de votre question, je crois que le problème est presque la même. Vous essayez de join un RDD avec un Broadcast et ce n'est pas permis. Toutefois, si vous avez lu la documentation que vous pouvez remarquer qu'il est possible de rejoindre les deux RDDs (voir code ci-dessous).
```
val joinRDD = FirstRDD.keyBy(_._1).join(SecondRDD.keyBy(_._1))
```
- Mais Après la diffusion, j'ai à nouveau convertir FinalData de RDD. Comme FinalData est de type Tableau.
- vous êtes peut-être pas de remarquer le fait que Data a type RDD[(String, String)] - exactement ce que vous semblez être à la recherche de...
- Il y a une fonction que je veux utiliser. Mais cette fonction est en RDD pas dans le Tableau.C'est pourquoi
- Zohar Oui Data a type RDD[(String, String)]
- Et quelle est la différence entre Data et la RDD vous essayez de créer?
- Zohar, Mais je veux utiliser la diffusion également. Ainsi, après la diffusion, je pourrai variable diffusé type. Je vais obtenir la valeur de Diffusé variable à l'aide de DataArray.value Et ce sera de retour en tant que Array Mais je le veux en RDD.Alors que l'avenir de l'opération, je peux effectuer sur RDD.
- Afin de vous intentionnellement souhaitez modifier les données avant et arrière 4 fois entre le conducteur de l'application et des nœuds de cluster, juste à la fin avec un identique de la valeur?
- Veuillez vérifier Nouvellement ajouter du code. Il va donner une brève idée sur ce que je suis à la recherche d'
- J'ai mis à jour ma réponse
InformationsquelleAutor Alberto Bonsanto

Vous devez vous connecter pour publier un commentaire.

MODIFIER