Przykład funkcji agregującej Scala

Zobaczmy, czy jakaś Sztuka ascii nie pomoże. Przyjrzyjmy się sygnaturze typu aggregate:

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

Należy również zauważyć, że A odnosi się do typu zbioru. Tak więc, powiedzmy, że mamy 4 elementy w tej kolekcji, wtedy aggregate może działać tak:

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

Zwróć uwagę na użycie par W poniższym fragmencie kodu. Drugi funkcja przekazywana do agregatu jest to, co nazywa się po obliczeniu poszczególnych sekwencji. Scala jest w stanie to zrobić tylko dla zestawów, które mogą być równoległe.

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

Więc najpierw obliczyłoby to:

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

Tego, co zrobi dalej, nie można przewidzieć (istnieje więcej niż jeden sposób łączenia wyników), ale może to zrobić (jak w powyższym artu ascii): {]}

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

W którym to momencie kończy się

6 + 9 // 15

Co daje ostateczny wynik. Teraz, to jest trochę podobna w strukturze do foldLeft, ale posiada dodatkową funkcję (B, B) => B, której fold nie posiada. Ta funkcja umożliwia jednak pracę równolegle!

Weźmy na przykład, że każde z czterech obliczeń początkowe obliczenia są niezależne od siebie i mogą być wykonywane równolegle. Kolejne dwa (w wyniku 6 i 9) mogą być uruchomione po zakończeniu obliczeń, od których zależą, ale te dwa mogą również działać równolegle.

The 7 obliczeń, równoległych jak wyżej, może zająć tak mało, jak w tym samym czasie 3 szeregowe obliczenia.

Właściwie, przy tak małym zbiorze koszt synchronizacji obliczeń byłby wystarczająco duży, aby wymazać wszelkie zyski. Co więcej, jeśli to złożysz, zajmie to tylko 4 / align = "left" / Gdy jednak Twoje Kolekcje staną się większe, zaczniesz dostrzegać prawdziwe zyski.

Zastanów się, z drugiej strony, foldLeft. Ponieważ nie posiada dodatkowej funkcji, to nie można porównywać żadnych obliczeń:

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

Każdy z nawiasów wewnętrznych musi być obliczony, zanim zewnętrzny będzie mógł kontynuować.

Sygnatura zbioru z elementami typu A wynosi:

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

• {[5] } jest obiektem typu B działającym jako element neutralny. Jeśli chcesz coś policzyć, możesz użyć 0, jeśli chcesz zbudować listę, zacznij od pustej listy, itp.
• {[6] } jest analogiczna do funkcji przekazywanej do metod fold. Wymaga dwóch argumentów, pierwszy jest tego samego typu co element neutralny, który przekazałeś i reprezentuje rzeczy, które zostały już zagregowane na poprzednim iteracja, druga to kolejny element Twojej kolekcji. Wynik musi być również typu B.
• combop: jest funkcją łączącą dwa wyniki w jednym.

W większości zbiorów agregat zaimplementowany jest w TraversableOnce jako:

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

Zatem combop jest ignorowany. Jednak ma to sens dla zbiorów równoległych , ponieważ seqop będzie najpierw stosowane lokalnie równolegle, a następnie wywołane zostanie combop, aby zakończyć agregację.

Więc dla Twojego przykładu, ty można spróbować z krotnie pierwszy:

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))

Potem trzeba znaleźć sposób na zawalenie dwóch multimapów:

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]() ) ) ) 
       } 

Teraz możesz używać agregatu równolegle:

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

Zastosowanie metody " par " do listy, a tym samym użycie zbioru równoległego (scala.kolekcja.równolegle.niezmienny.ParSeq) z listy, aby naprawdę wykorzystać procesory wielordzeniowe. Bez " par " nie będzie żadnego zwiększenia wydajności, ponieważ agregat nie jest wykonywany na kolekcji równoległej.

aggregate jest jak foldLeft, ale może być wykonywany równolegle.

Jak mówi missingfactor , liniowa wersja aggregate(z)(seqop, combop) jest równoważna foldleft(z)(seqop). Jest to jednak niepraktyczne w przypadku równoległym, gdzie musielibyśmy połączyć nie tylko następny element z poprzednim wynikiem (jak w normalnym foldzie), ale chcemy podzielić iterable na sub-iterable, które nazywamy agregatem i musimy połączyć je ponownie. (W porządku od lewej do prawej, ale nie asocjacyjnym, ponieważ moglibyśmy połączyć ostatnie części przed częściami pięści iterowalnych.) To ponowne łączenie się w ogólnie nietrywialne, a zatem potrzeba metody (S, S) => S, aby to osiągnąć.

Definicja w ParIterableLike to:

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

Który rzeczywiście używa combop.

Dla odniesienia, Aggregate jest zdefiniowane jako:

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)
}

Ważną częścią jest merge gdzie combop jest stosowany z dwoma pod-wynikami.

Oto blog o tym, jak agregować włączyć wydajność na procesorze wielordzeniowym z bench mark. http://markusjais.com/scalas-parallel-collections-and-the-aggregate-method/

Oto wideo na temat rozmowy "Scala parallel collections" z "dni Scala 2011". http://days2011.scala-lang.org/node/138/272

Opis na filmie

Scala Parallel Collections

Aleksandar Prokopec

Abstrakcje programowania równoległego stają się coraz większe znaczenie wraz ze wzrostem liczby rdzeni procesora. Model programowania wysokiego poziomu umożliwia programiście skupienie się bardziej na programie, a mniej na niskopoziomowych szczegółach, takich jak synchronizacja i równoważenie obciążenia. Scala parallel collections rozszerza model programowania Scala collection framework, zapewniając równoległe operacje na zbiorach danych. Wykład będzie opisywał architekturę równoległych RAM kolekcji, wyjaśniając ich wdrożenie i decyzje projektowe. Beton zostaną opisane implementacje kolekcji, takie jak parallel hash maps i parallel hash tries. Na koniec zostanie pokazanych kilka przykładowych aplikacji, demonstrujących model programowania w praktyce.

Definicja aggregate w TraversableOnce źródle to:

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

Który nie różni się niczym od prostego foldLeft. combop nie wydaje się być nigdzie używany. Sam jestem zdezorientowany, jaki jest cel tej metody.

Dla wyjaśnienia tych przede mną, w teorii chodzi o to, że agregat powinien działać w ten sposób, (zmieniłem nazwy parametrów, aby były jaśniejsze):

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

Należy logicznie przetłumaczyć na

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)

Ponieważ agregacja i mapowanie są oddzielne, oryginalna lista może teoretycznie być podzielona na różne grupy o różnych rozmiarach i działać równolegle lub nawet na różnych maszynach. W praktyce scala obecna implementacja nie obsługuje ta funkcja domyślnie, ale możesz to zrobić w swoim własnym kodzie.