Unterschied zwischen Reduzieren und Falten Links / Falten in der funktionalen Programmierung (insbesondere Scala- und Scala-APIs)?

Warum haben Scala und Frameworks wie Spark und Scalding beide reduceund foldLeft? Was ist dann der Unterschied zwischen reduceund fold?

reduzieren gegen foldLeft

Ein großer Unterschied, der in keiner anderen Stackoverflow-Antwort zu diesem Thema eindeutig erwähnt wird, besteht darin, dass reduceein kommutatives Monoid angegeben werden sollte , dh eine Operation, die sowohl kommutativ als auch assoziativ ist. Dies bedeutet, dass die Operation parallelisiert werden kann.

Diese Unterscheidung ist sehr wichtig für Big Data / MPP / Distributed Computing und der gesamte Grund, warum es reduceüberhaupt existiert. Die Sammlung kann zerhackt werden und die reduceDose kann an jedem Block arbeiten, dann reducekann die Dose an den Ergebnissen jedes Blocks arbeiten - tatsächlich muss die Chunking-Ebene nicht eine Ebene tiefer anhalten. Wir könnten auch jedes Stück zerhacken. Aus diesem Grund ist das Summieren von Ganzzahlen in einer Liste O (log N), wenn eine unendliche Anzahl von CPUs angegeben wird.

Wenn Sie sich nur die Signaturen ansehen, gibt es keinen Grund reducezu existieren, weil Sie reducemit einem alles erreichen können, was Sie können foldLeft. Die Funktionalität von foldLeftist größer als die Funktionalität von reduce.

Aber man kann eine nicht parallelisieren foldLeft, so dass ihre Laufzeit ist immer O (N) (auch wenn Sie in einem kommutativen Monoid füttern). Dies liegt daran, dass angenommen wird, dass die Operation kein kommutatives Monoid ist und der kumulierte Wert daher durch eine Reihe aufeinanderfolgender Aggregationen berechnet wird.

foldLeftnimmt weder Kommutativität noch Assoziativität an. Es ist die Assoziativität, die es ermöglicht, die Sammlung zu zerlegen, und die Kommutativität, die das Kumulieren erleichtert, da die Reihenfolge nicht wichtig ist (es spielt also keine Rolle, in welcher Reihenfolge die einzelnen Ergebnisse aus den einzelnen Blöcken aggregiert werden sollen). Genau genommen ist Kommutativität für die Parallelisierung nicht erforderlich, beispielsweise für verteilte Sortieralgorithmen. Sie erleichtert lediglich die Logik, da Sie Ihren Chunks keine Reihenfolge geben müssen.

Wenn Sie sich die Spark-Dokumentation ansehen, reduceheißt es speziell "... kommutativer und assoziativer Binäroperator".

http://spark.apache.org/docs/1.0.0/api/scala/index.html#org.apache.spark.rdd.RDD

Hier ist ein Beweis, der reduceNICHT nur ein Sonderfall von istfoldLeft

scala> val intParList: ParSeq[Int] = (1 to 100000).map(_ => scala.util.Random.nextInt()).par

scala> timeMany(1000, intParList.reduce(_ + _))
Took 462.395867 milli seconds

scala> timeMany(1000, intParList.foldLeft(0)(_ + _))
Took 2589.363031 milli seconds

gegen falten reduzieren

Hier kommt es den FP / mathematischen Wurzeln etwas näher und es ist etwas schwieriger zu erklären. Reduzieren wird formal als Teil des MapReduce-Paradigmas definiert, das sich mit geordneten Sammlungen (Multisets) befasst. Falten wird formal als Rekursion definiert (siehe Katamorphose) und nimmt daher eine Struktur / Sequenz zu den Sammlungen an.

foldIn Scalding gibt es keine Methode, da wir sie unter dem (strengen) Map Reduce-Programmiermodell nicht definieren können, foldda Chunks keine Reihenfolge haben und foldnur Assoziativität und keine Kommutativität erfordern.

Einfach ausgedrückt, reducefunktioniert ohne eine Reihenfolge der Kumulierung, folderfordert eine Reihenfolge der Kumulierung und es ist diese Reihenfolge der Kumulierung, die einen Nullwert erfordert, NICHT die Existenz des Nullwerts, der sie unterscheidet. Genau genommen reduce sollte dies für eine leere Sammlung funktionieren, da ihr Nullwert abgeleitet werden kann, indem ein beliebiger Wert genommen xund dann gelöst x op y = xwird. Dies funktioniert jedoch nicht mit einer nicht kommutativen Operation, da es einen unterschiedlichen linken und rechten Nullwert geben kann (dh x op y != y op x). Natürlich macht sich Scala nicht die Mühe, herauszufinden, was dieser Nullwert ist, da dies etwas Mathematik erfordern würde (die wahrscheinlich nicht berechenbar ist), also löst sie einfach eine Ausnahme aus.

Es scheint (wie es in der Etymologie häufig der Fall ist), dass diese ursprüngliche mathematische Bedeutung verloren gegangen ist, da der einzige offensichtliche Unterschied in der Programmierung die Signatur ist. Das Ergebnis ist, dass reducees zu einem Synonym für foldMapReduce geworden ist , anstatt die ursprüngliche Bedeutung von MapReduce beizubehalten. Heutzutage werden diese Begriffe häufig synonym verwendet und verhalten sich in den meisten Implementierungen gleich (wobei leere Sammlungen ignoriert werden). Die Seltsamkeit wird durch Besonderheiten wie in Spark verschärft, auf die wir jetzt eingehen werden.

Spark hat also eine fold, aber die Reihenfolge, in der Unterergebnisse (eines für jede Partition) kombiniert werden (zum Zeitpunkt des Schreibens), ist dieselbe Reihenfolge, in der Aufgaben erledigt werden - und somit nicht deterministisch. Vielen Dank an @CafeFeed für den Hinweis auf die foldVerwendung runJob. Nachdem ich den Code gelesen hatte, stellte ich fest, dass er nicht deterministisch ist. Weitere Verwirrung wird dadurch erzeugt, dass Spark ein treeReduceaber nein hat treeFold.

Fazit

Es gibt einen Unterschied zwischen reduceund foldauch bei Anwendung auf nicht leere Sequenzen. Ersteres wird als Teil des MapReduce-Programmierparadigmas für Sammlungen mit beliebiger Reihenfolge definiert ( http://theory.stanford.edu/~sergei/papers/soda10-mrc.pdf ), und man sollte davon ausgehen, dass Operatoren nicht nur kommutativ sind assoziativ, um deterministische Ergebnisse zu liefern. Letzteres wird in Bezug auf Katomorphismen definiert und erfordert, dass die Sammlungen einen Sequenzbegriff haben (oder rekursiv wie verknüpfte Listen definiert sind) und daher keine kommutativen Operatoren erfordern.

In der Praxis aufgrund des unmathematischen Charakters der Programmierung reduceund foldneigen dazu, sich entweder korrekt (wie in Scala) oder falsch (wie in Spark) gleich zu verhalten.

Extra: Meine Meinung zur Spark-API

Meiner Meinung nach würde Verwirrung vermieden, wenn die Verwendung des Begriffs foldin Spark vollständig gestrichen würde. Zumindest hat spark einen Hinweis in der Dokumentation:

Dies verhält sich etwas anders als Fold-Operationen, die für nicht verteilte Sammlungen in funktionalen Sprachen wie Scala implementiert sind.

Wenn ich mich nicht irre, obwohl die Spark-API dies nicht erfordert, erfordert fold auch, dass das f kommutativ ist. Weil die Reihenfolge, in der die Partitionen aggregiert werden, nicht gewährleistet ist. Zum Beispiel wird im folgenden Code nur der erste Ausdruck sortiert:

import org.apache.spark.{SparkConf, SparkContext}

object FoldExample extends App{

  val conf = new SparkConf()
    .setMaster("local[*]")
    .setAppName("Simple Application")
  implicit val sc = new SparkContext(conf)

  val range = ('a' to 'z').map(_.toString)
  val rdd = sc.parallelize(range)

  println(range.reduce(_ + _))
  println(rdd.reduce(_ + _))
  println(rdd.fold("")(_ + _))
}  

Ausdrucken:

abcdefghijklmnopqrstuvwxyz

abcghituvjklmwxyzqrsdefnop

defghinopjklmqrstuvabcwxyz

foldin Apache Spark ist nicht dasselbe wie foldin nicht verteilten Sammlungen. Tatsächlich erfordert es eine kommutative Funktion , um deterministische Ergebnisse zu erzielen:

Dies verhält sich etwas anders als Fold-Operationen, die für nicht verteilte Sammlungen in funktionalen Sprachen wie Scala implementiert sind. Diese Faltoperation kann einzeln auf Partitionen angewendet werden und diese Ergebnisse dann in das Endergebnis falten, anstatt die Faltung nacheinander in einer definierten Reihenfolge auf jedes Element anzuwenden. Für Funktionen, die nicht kommutativ sind, kann das Ergebnis von dem einer Falte abweichen, die auf eine nicht verteilte Sammlung angewendet wird.

Dies wurde von Mishael Rosenthal gezeigt und von Make42 in seinem Kommentar vorgeschlagen .

Es wurde vermutet, dass das beobachtete Verhalten damit zusammenhängt, HashPartitionerdass tatsächlich parallelizenicht gemischt und nicht verwendet wirdHashPartitioner .

import org.apache.spark.sql.SparkSession

/* Note: standalone (non-local) mode */
val master = "spark://...:7077"  

val spark = SparkSession.builder.master(master).getOrCreate()

/* Note: deterministic order */
val rdd = sc.parallelize(Seq("a", "b", "c", "d"), 4).sortBy(identity[String])
require(rdd.collect.sliding(2).forall { case Array(x, y) => x < y })

/* Note: all posible permutations */
require(Seq.fill(1000)(rdd.fold("")(_ + _)).toSet.size == 24)

Erklärt:

Struktur vonfold für RDD

def fold(zeroValue: T)(op: (T, T) => T): T = withScope {
  var jobResult: T
  val cleanOp: (T, T) => T
  val foldPartition = Iterator[T] => T
  val mergeResult: (Int, T) => Unit
  sc.runJob(this, foldPartition, mergeResult)
  jobResult
}

ist die gleiche Struktur wiereduce für RDD:

def reduce(f: (T, T) => T): T = withScope {
  val cleanF: (T, T) => T
  val reducePartition: Iterator[T] => Option[T]
  var jobResult: Option[T]
  val mergeResult =  (Int, Option[T]) => Unit
  sc.runJob(this, reducePartition, mergeResult)
  jobResult.getOrElse(throw new UnsupportedOperationException("empty collection"))
}

wo runJob wird die Partitionsreihenfolge nicht beachtet und es wird eine kommutative Funktion benötigt.

foldPartitionund reducePartitionsind in Bezug auf die Reihenfolge der Verarbeitung gleichwertig und werden effektiv (durch Vererbung und Delegierung) von reduceLeftund foldLeftan implementiert TraversableOnce.

Schlussfolgerung: foldRDD kann nicht von der Reihenfolge der Chunks abhängen und benötigt Kommutativität und Assoziativität .

Ein weiterer Unterschied für Scalding ist die Verwendung von Kombinierern in Hadoop.

Stellen Sie sich vor, Ihre Operation ist ein kommutatives Monoid. Mit Reduzieren wird sie auch auf der Kartenseite angewendet, anstatt alle Daten zu Reduzierern zu mischen / zu sortieren. Bei foldLeft ist dies nicht der Fall.

pipe.groupBy('product) {
   _.reduce('price -> 'total){ (sum: Double, price: Double) => sum + price }
   // reduce is .mapReduceMap in disguise
}

pipe.groupBy('product) {
   _.foldLeft('price -> 'total)(0.0){ (sum: Double, price: Double) => sum + price }
}

Es ist immer empfehlenswert, Ihre Operationen in Scalding als Monoid zu definieren.