Jak działa Spark aggregate function-aggregateByKey?

Agregatebykey() różni się od reduceByKey. Dzieje się tak, że reduceByKey jest rodzajem szczególnego przypadku aggregateByKey.

AggregateByKey() połączy wartości dla określonego klucza, a wynikiem takiej kombinacji może być dowolny obiekt, który określisz. Musisz określić, w jaki sposób wartości są łączone ("dodane") wewnątrz jednej partycji (która jest wykonywana w tym samym węźle) i w jaki sposób łączysz wynik z różnych partycji (które mogą znajdować się w różnych węzłach). reduceByKey jest szczególnym przypadkiem, w tym sensie, że wynik kombinacji (np. suma) jest tego samego typu Co wartości, a operacja po połączeniu z różnych partycji jest również taka sama jak operacja po połączeniu wartości wewnątrz partycji.

Przykład: Wyobraź sobie, że masz listę par. Można to porównać:

val pairs = sc.parallelize(Array(("a", 3), ("a", 1), ("b", 7), ("a", 5)))

Teraz chcesz "połączyć" je za pomocą klucza produkującego sumę. W tym przypadku reduceByKey i aggregateByKey są to samo:

val resReduce = pairs.reduceByKey(_ + _) //the same operation for everything
resReduce.collect
res3: Array[(String, Int)] = Array((b,7), (a,9))

//0 is initial value, _+_ inside partition, _+_ between partitions
val resAgg = pairs.aggregateByKey(0)(_+_,_+_)
resAgg.collect
res4: Array[(String, Int)] = Array((b,7), (a,9))

Teraz wyobraź sobie, że chcesz, aby agregacja była zbiorem wartości, czyli innym typem niż wartości, które są liczbami całkowitymi (suma liczb całkowitych jest również liczbami całkowitymi):

import scala.collection.mutable.HashSet
//the initial value is a void Set. Adding an element to a set is the first
//_+_ Join two sets is the  _++_
val sets = pairs.aggregateByKey(new HashSet[Int])(_+_, _++_)
sets.collect
res5: Array[(String, scala.collection.mutable.HashSet[Int])]  =Array((b,Set(7)), (a,Set(1, 5, 3)))