Exemple de la fonction d'agrégation Scala

voyons si l'art ascii n'aide pas. Considérons le type de signature de aggregate :

def aggregate [B] (z: B)(seqop: (B, A) ⇒ B, combop: (B, B) ⇒ B): B

aussi, notez que A se réfère au type de la collection. Donc, disons que nous avons 4 éléments dans cette collection, alors aggregate pourrait fonctionner comme ceci:

z   A   z   A   z   A   z   A
 \ /     \ /seqop\ /     \ /    
  B       B       B       B
    \   /  combop   \   /
      B _           _ B
         \ combop  /
              B

voyons un exemple pratique de cela. Disons que j'ai un GenSeq("This", "is", "an", "example") , et je veux savoir combien il y a de caractères dedans. Je peux écrire le

noter l'utilisation de par dans l'extrait de code ci-dessous. La seconde fonction transmise à aggregate est ce qui est appelé après les séquences individuelles sont calculées. Scala est seulement capable de faire cela pour les ensembles peuvent être parallélisées.

import scala.collection.GenSeq
val seq = GenSeq("This", "is", "an", "example")
val chars = seq.par.aggregate(0)(_ + _.length, _ + _)

donc, d'abord il calculerait ceci:

0 + "This".length     // 4
0 + "is".length       // 2
0 + "an".length       // 2
0 + "example".length  // 7

ce qu'il fait ensuite ne peut pas être prédit (il ya plus d'une façon de combiner les résultats), mais il pourrait faites ceci (comme dans l'art ascii ci-dessus):

4 + 2 // 6
2 + 7 // 9

où il se termine par

6 + 9 // 15

qui donne le résultat final. Maintenant, c'est un peu similaire dans la structure à foldLeft , mais il a une fonction supplémentaire (B, B) => B , que fold n'a pas. Cette fonction lui permet toutefois de travailler en parallèle!

considèrent, par exemple, que chacun des quatre calculs les calculs initiaux sont indépendants les uns des autres et peuvent être effectués en parallèle. Les deux suivants (résultant en 6 et 9) peuvent être commencés une fois que leurs calculs sur lesquels ils dépendent sont finis, mais ces deux peuvent aussi exécuter en parallèle.

les 7 calculs, mis en parallèle comme ci-dessus, pouvaient prendre aussi peu que le même temps 3 calculs en série.

en fait, avec une si petite collection Le coût de synchronisation des calculs serait assez grand pour effacer tout gain. En outre, si vous avez plié ceci, il ne faudrait 4 calculs total. Une fois que vos collections sont plus grandes, cependant, vous commencez à voir des gains réels.

Considèrent, en revanche, foldLeft . Parce qu'il n'a pas la fonction supplémentaire, il ne peut pas paralléliser n'importe quel calcul:

(((0 + "This".length) + "is".length) + "an".length) + "example".length

chacune des parenthèses intérieures doit être calculée avant la parenthèse extérieure pouvez procéder.

la signature pour une collection avec des éléments de type A est:

def aggregate [B] (z: B)(seqop: (B, A) ⇒ B, combop: (B, B) ⇒ B): B 

• z est un objet de type B agissant comme un élément neutre. Si vous voulez compter quelque chose, vous pouvez utiliser 0, si vous voulez construire une liste, commencer avec une liste vide, etc.
• segop est analogue à la fonction que vous passez aux méthodes fold . Elle prend deux arguments, la première est du même type que l'élément neutre vous passé et représentent la substance qui a déjà été agrégé sur itération précédente, le second est l'élément suivant de votre collection. Le résultat doit également être de type B .
• combop est une fonction combinant deux résultats en un.

dans la plupart des collections, aggregate est mis en œuvre dans TraversableOnce comme:

  def aggregate[B](z: B)(seqop: (B, A) => B, combop: (B, B) => B): B 
    = foldLeft(z)(seqop)

ainsi combop est ignoré. Cependant, il est logique pour collections parallèles , parce que seqop sera d'abord appliqué localement en parallèle, puis combop est appelé pour terminer l'agrégation.

Donc, pour votre exemple, vous pouvez essayer avec un pli d'abord:

val seqOp = 
  (map:Map[String,Set[String]],tuple: (String,String)) => 
    map + ( tuple._1 -> ( map.getOrElse( tuple._1, Set[String]() ) + tuple._2 ) )


list.foldLeft( Map[String,Set[String]]() )( seqOp )
// returns: Map(one -> Set(i, 1), two -> Set(2, ii), four -> Set(iv))

, Alors vous devez trouver un moyen de s'effondrer deux multimaps:

val combOp = (map1: Map[String,Set[String]], map2: Map[String,Set[String]]) =>
       (map1.keySet ++ map2.keySet).foldLeft( Map[String,Set[String]]() ) { 
         (result,k) => 
           result + ( k -> ( map1.getOrElse(k,Set[String]() ) ++ map2.getOrElse(k,Set[String]() ) ) ) 
       } 

Maintenant, vous pouvez utiliser aggregate en parallèle:

list.par.aggregate( Map[String,Set[String]]() )( seqOp, combOp )
//Returns: Map(one -> Set(i, 1), two -> Set(2, ii), four -> Set(iv))

Application de la méthode "égalité" pour liste, utilisant ainsi la collection parallèle (scala.collection.parallèle.immuable.ParSeq) de la liste pour vraiment profiter des processeurs multi core. Sans "par", il n'y aura pas de gain de performance puisque l'agrégat n'est pas fait sur la collecte parallèle.

aggregate est comme foldLeft mais peut être exécuté en parallèle.

Comme missingfactor dit , la version linéaire de aggregate(z)(seqop, combop) est équivalent à foldleft(z)(seqop) . Cela n'est toutefois pas possible dans le cas parallèle, où nous aurions besoin de combiner non seulement l'élément suivant avec le résultat précédent (comme dans un pli normal), mais nous voulons diviser le itérable en sous-itérables sur lesquels nous appelons agrégat et avons besoin de combiner à nouveau ceux-ci. (De gauche à droite, mais pas associatif que nous pourrions avoir combiné les dernières pièces avant que le poing parties de l'itératif.) Cette re-combinaison en général non-trivial, et donc, on a besoin d'une méthode (S, S) => S pour accomplir cela.

la définition de ParIterableLike est la suivante:

def aggregate[S](z: S)(seqop: (S, T) => S, combop: (S, S) => S): S = {
  executeAndWaitResult(new Aggregate(z, seqop, combop, splitter))
}

qui utilise en effet combop .

pour référence, Aggregate est défini comme:

protected[this] class Aggregate[S](z: S, seqop: (S, T) => S, combop: (S, S) => S, protected[this] val pit: IterableSplitter[T])
  extends Accessor[S, Aggregate[S]] {
    @volatile var result: S = null.asInstanceOf[S]
    def leaf(prevr: Option[S]) = result = pit.foldLeft(z)(seqop)
    protected[this] def newSubtask(p: IterableSplitter[T]) = new Aggregate(z, seqop, combop, p)
    override def merge(that: Aggregate[S]) = result = combop(result, that.result)
}

La partie importante est merge où combop est appliquée avec deux sous-résultats.

voici le blog sur la façon dont aggregate activer la performance sur le multi core processeur avec bench mark. http://markusjais.com/scalas-parallel-collections-and-the-aggregate-method/

Voici une vidéo sur la "Scala parallèle des collections" parler de "Scala Jours 2011". http://days2011.scala-lang.org/node/138/272

la description sur la vidéo

Collections Parallèles De Scala

Aleksandar Prokopec

les abstractions de programmation en parallèle prennent de plus en plus d'importance au fur et à mesure que le nombre de cœurs des processeurs augmente. Une programmation de haut niveau permet au programmeur de se concentrer davantage sur le programme, et moins sur les détails de bas niveau comme la synchronisation et l'équilibrage de charge. Les collections parallèles de Scala étendent le modèle de programmation du cadre de collecte de Scala, en fournissant des opérations parallèles sur les ensembles de données. La présentation décrira architecture du cadre de collecte parallèle, expliquant leurs décisions de mise en œuvre et de conception. Des applications concrètes de collecte telles que des cartes de hachage parallèles et des essais de hachage parallèles seront décrites. Enfin, plusieurs exemples d'applications seront présentés, démontrant le modèle de programmation dans la pratique.

la définition de aggregate dans TraversableOnce source est:

def aggregate[B](z: B)(seqop: (B, A) => B, combop: (B, B) => B): B = 
  foldLeft(z)(seqop)

qui n'est pas différent d'un simple foldLeft . combop ne semble être utilisé nulle part. Je suis moi-même confus quant à ce que le but de cette méthode est.

juste pour clarifier les explications de ceux qui sont devant moi, en théorie l'idée est que total devrait fonctionner comme ça, (j'ai changé les noms des paramètres pour les rendre plus claire):

Seq(1,2,3,4).aggragate(0)(
     addToPrev = (prev,curr) => prev + curr, 
     combineSums = (sumA,sumB) => sumA + sumB)

devrait logiquement se traduire par

Seq(1,2,3,4)
    .grouped(2) // split into groups of 2 members each
    .map(prevAndCurrList => prevAndCurrList(0) + prevAndCurrList(1))
    .foldLeft(0)(sumA,sumB => sumA + sumB)

parce que l'agrégation et la cartographie sont séparées, la liste originale pourrait théoriquement être divisée en différents groupes de tailles différentes et fonctionner en parallèle ou même sur des machines différentes. Dans l'implémentation actuelle de la pratique scala ne supporte pas cette fonctionnalité par défaut mais vous pouvez le faire dans votre propre code.