reduceByKey: Comment ça marche?

Je suis nouveau à l'Allumage et à la Scala. J'ai été confus au sujet de la façon reduceByKey fonctionne dans Spark. Supposons que nous avons le code suivant:

val lines = sc.textFile("data.txt")
val pairs = lines.map(s => (s, 1))
val counts = pairs.reduceByKey((a, b) => a + b)

La fonction map est clair: s est la clé et il points de la ligne de data.txt et 1 est la valeur.

Cependant, je n'ai pas comment le reduceByKey fonctionne en interne? Ne "une" des points à la clé? Sinon, n'est "qu'un" point "s"? Ensuite, ce qui représente un + b? comment sont-ils remplis?

InformationsquelleAutor user764186 | 2015-05-09

78

Nous allons le décomposer à des méthodes et des types. Qui, généralement, expose les subtilités de nouveaux développeurs:
```
pairs.reduceByKey((a, b) => a + b)
```
devient
```
pairs.reduceByKey((a: Int, b: Int) => a + b)
```
et en renommant les variables rend un peu plus explicite
```
pairs.reduceByKey((accumulatedValue: Int, currentValue: Int) => accumulatedValue + currentValue)
```
Donc, nous pouvons maintenant voir que nous sommes tout simplement en prenant une valeur cumulée pour la clé donnée et en additionnant avec la valeur suivante de la touche. MAINTENANT, nous allons casser, donc nous pouvons comprendre la partie de la clé. Donc, nous allons visualiser la méthode plus comme ceci:
```
pairs.reduce((accumulatedValue: List[(String, Int)], currentValue: (String, Int)) => {
  //Turn the accumulated value into a true key->value mapping
  val accumAsMap = accumulatedValue.toMap   
  //Try to get the key's current value if we've already encountered it
  accumAsMap.get(currentValue._1) match { 
    //If we have encountered it, then add the new value to the existing value and overwrite the old
    case Some(value : Int) => (accumAsMap + (currentValue._1 -> (value + currentValue._2))).toList
    //If we have NOT encountered it, then simply add it to the list
    case None => currentValue :: accumulatedValue 
  }
})
```
Donc, vous pouvez voir que le réduireByKey prend le passe-partout de trouver la clé et de suivi de sorte que vous n'avez pas à vous soucier de la gestion de cette partie.

De plus profond, de plus vrai si vous voulez

Tout cela étant dit, c'est une version simplifiée de ce qui se passe car il y a des optimisations qui sont faits ici. Cette opération est associative, de sorte que l'étincelle, le moteur de réaliser ces réductions localement première (souvent appelée carte côté de réduction) puis, une fois encore, le conducteur. Cela permet d'économiser le trafic réseau; au lieu de les envoyer toutes les données et la réalisation de l'opération, il peut réduire aussi petite que possible et ensuite l'envoyer réduction sur le fil.

InformationsquelleAutor Justin Pihony
40

Une exigence pour la reduceByKey fonction est qu'elle doit être associative. Pour construire une intuition sur la façon reduceByKey fonctionne, nous allons d'abord voir comment associatif associatif de la fonction nous aide dans un calcul parallèle:

Comme nous pouvons le voir, nous pouvons briser une collection originale en morceaux et en appliquant la fonction associative, nous pouvons accumuler un total. Le cas séquentiel est trivial, nous y sommes habitués: 1+2+3+4+5+6+7+8+9+10.

Associativité nous permet de l'utiliser de la même fonction, dans l'ordre et en parallèle. reduceByKey utilise cette propriété pour calculer un résultat d'une EDR, qui est une collection distribuée composé de partitions.

Considérons l'exemple suivant:
```
//collection of the form ("key",1),("key,2),...,("key",20) split among 4 partitions
val rdd =sparkContext.parallelize(( (1 to 20).map(x=>("key",x))), 4)
rdd.reduceByKey(_ + _)
rdd.collect()
> Array[(String, Int)] = Array((key,210))
```
Spark, les données sont réparties dans des partitions. Pour l'illustration suivante, (4) les partitions sont à gauche, enfermés dans des lignes fines. Tout d'abord, nous appliquons la fonction localement pour chaque partition, de façon séquentielle dans la partition, mais nous courons tous les 4 partitions en parallèle. Ensuite, le résultat de chaque calcul sont regroupées en appliquant la même fonction nouveau et enfin arriver à un résultat.

reduceByKey est une spécialisation de aggregateByKey aggregateByKey prend 2 fonctions: une qui est appliquée à chaque partition (séquentielle) et celui qui est appliqué entre les résultats de chaque partition (en parallèle). reduceByKey utilise la même fonction associative sur les deux cas: pour faire un calcul séquentiel sur chaque partition, puis de combiner ces résultats dans un résultat final, comme nous l'avons illustré ici.

InformationsquelleAutor maasg
6

Dans votre exemple de
```
val counts = pairs.reduceByKey((a,b) => a+b)
```
a et b sont à la fois Int accumulateurs pour _2 des tuples dans pairs. reduceKey prendra deux tuples ayant la même valeur s et d'utiliser leurs _2 valeurs a et b, la production d'une nouvelle Tuple[String,Int]. Cette opération est répétée jusqu'à ce qu'il y a un seul tuple pour chaque clé s.

Contrairement à la non-Étincelle (ou, vraiment, non parallèles) reduceByKey où le premier élément est toujours l'accumulateur et la deuxième valeur, reduceByKey fonctionne dans un mode de distribution, c'est à dire chaque nœud sera de réduire, c'est un ensemble de tuples dans une collection de unique assortie tuples et puis de réduire les tuples à partir de plusieurs nœuds jusqu'à ce qu'il y a une dernière unique assortie ensemble de tuples. Cela signifie que les résultats de nœuds sont réduits, a et b représentent déjà réduit les accumulateurs.

InformationsquelleAutor Arne Claassen
0

Étincelle RDD reduceByKey fonction fusionne les valeurs pour chaque clé à l'aide d'un associative fonction de réduction.

La reduceByKey fonction ne fonctionne que sur les Edr et c'est une opération de transformation qui signifie qu'il est paresseusement évalué. Et une fonction associative est passé en paramètre, qui est appliqué à la source de CA et crée un nouveau CA comme un résultat.

Donc dans votre exemple, ca paires a un ensemble de plusieurs éléments appariés comme (s1,1), (s2,1), etc. Et reduceByKey accepte une fonction (accumulateur, n) => (accumulateur + n), ce qui permet d'initialiser la variable accumulateur à la valeur par défaut 0 et ajoute l'élément pour chaque touche et de retourner le résultat rdd compte avoir le nombre total de comptes jumelé avec la clé.

InformationsquelleAutor KayV

Vous devez vous connecter pour publier un commentaire.