Différence entre réduire et foldLeft / fold dans la programmation fonctionnelle (en particulier les API Scala et Scala)?

Pourquoi Scala et des frameworks comme Spark et Scalding ont-ils à la fois reduceet foldLeft? Alors, quelle est la différence entre reduceet fold?

réduire vs replier

Une grande différence, qui n'est mentionnée dans aucune autre réponse de stackoverflow relative à ce sujet clairement, est qu'il reducefaut donner un monoïde commutatif , c'est-à-dire une opération à la fois commutative et associative. Cela signifie que l'opération peut être parallélisée.

Cette distinction est très importante pour le Big Data / MPP / l'informatique distribuée, et toute la raison pour laquelle reduceelle existe même. La collection peut être découpée et le reducepeut fonctionner sur chaque morceau, puis le reducepeut fonctionner sur les résultats de chaque morceau - en fait, le niveau de segmentation n'a pas besoin de s'arrêter d'un niveau de profondeur. Nous pourrions aussi couper chaque morceau. C'est pourquoi la somme des entiers dans une liste est O (log N) si on lui donne un nombre infini de processeurs.

Si vous regardez simplement les signatures, il n'y a aucune raison reduced'exister car vous pouvez réaliser tout ce que vous pouvez reduceavec un foldLeft. La fonctionnalité de foldLeftest supérieure à la fonctionnalité de reduce.

Mais vous ne pouvez pas paralléliser a foldLeft, donc son runtime est toujours O (N) (même si vous alimentez un monoïde commutatif). En effet, on suppose que l'opération n'est pas un monoïde commutatif et que la valeur cumulée sera donc calculée par une série d'agrégations séquentielles.

foldLeftne suppose ni commutativité ni associativité. C'est l'associativité qui donne la possibilité de découper la collection, et c'est la commutativité qui facilite le cumul car l'ordre n'est pas important (donc peu importe l'ordre d'agréger chacun des résultats de chacun des blocs). La commutativité à proprement parler n'est pas nécessaire pour la parallélisation, par exemple les algorithmes de tri distribué, elle facilite simplement la logique car vous n'avez pas besoin de donner un ordre à vos morceaux.

Si vous jetez un œil à la documentation Spark, reduceelle dit spécifiquement "... opérateur binaire commutatif et associatif"

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

Voici la preuve qui reducen'est PAS juste un cas particulier defoldLeft

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

réduire vs plier

Maintenant, c'est là que cela se rapproche un peu plus des racines FP / mathématiques, et un peu plus difficile à expliquer. Réduire est défini formellement dans le cadre du paradigme MapReduce, qui traite des collections sans ordre (multisets), Fold est formellement défini en termes de récursivité (voir catamorphisme) et suppose donc une structure / séquence pour les collections.

Il n'y a pas de foldméthode dans Scalding car sous le modèle de programmation (strict) Map Reduce, nous ne pouvons pas définir foldparce que les morceaux n'ont pas d'ordre et foldne nécessitent que l'associativité, pas la commutativité.

En termes simples, reducefonctionne sans ordre de cumul, foldnécessite un ordre de cumul et c'est cet ordre de cumul qui nécessite une valeur zéro PAS l'existence de la valeur zéro qui les distingue. Strictement parlant, reduce devrait fonctionner sur une collection vide, car sa valeur zéro peut être déduite en prenant une valeur arbitraire xpuis en la résolvant x op y = x, mais cela ne fonctionne pas avec une opération non commutative car il peut exister une valeur zéro gauche et droite distincte (ie x op y != y op x). Bien sûr, Scala ne prend pas la peine de déterminer quelle est cette valeur zéro car cela nécessiterait de faire des mathématiques (qui sont probablement non calculables), alors lève simplement une exception.

Il semble (comme c'est souvent le cas en étymologie) que cette signification mathématique originelle a été perdue, puisque la seule différence évidente dans la programmation est la signature. Le résultat est que c'est reducedevenu un synonyme de fold, plutôt que de préserver sa signification originale de MapReduce. Maintenant, ces termes sont souvent utilisés de manière interchangeable et se comportent de la même manière dans la plupart des implémentations (ignorant les collections vides). La bizarrerie est exacerbée par des particularités, comme dans Spark, que nous allons maintenant aborder.

Spark a donc un fold, mais l'ordre dans lequel les sous-résultats (un pour chaque partition) sont combinés (au moment de l'écriture) est le même ordre dans lequel les tâches sont terminées - et donc non déterministe. Merci à @CafeFeed d'avoir souligné cette foldutilisation runJob, qui après avoir lu le code, j'ai réalisé que ce n'était pas déterministe. Une confusion supplémentaire est créée par Spark ayant un treeReducemais non treeFold.

Conclusion

Il y a une différence entre reduceetfold même lorsqu'il est appliqué à des séquences non vides. Le premier est défini comme faisant partie du paradigme de programmation MapReduce sur les collections avec un ordre arbitraire ( http://theory.stanford.edu/~sergei/papers/soda10-mrc.pdf ) et on devrait supposer que les opérateurs sont commutatifs en plus d'être associatif pour donner des résultats déterministes. Ce dernier est défini en termes de catomorphismes et nécessite que les collections aient une notion de séquence (ou soient définies de manière récursive, comme les listes chaînées), donc ne nécessitent pas d'opérateurs commutatifs.

En pratique, en raison de la nature non mathématique de la programmation, reduce et foldont tendance à se comporter de la même manière, soit correctement (comme dans Scala), soit incorrectement (comme dans Spark).

Extra: Mon avis sur l'API Spark

Mon opinion est que la confusion serait évitée si l'utilisation du terme foldétait complètement abandonnée dans Spark. Au moins, Spark a une note dans sa documentation:

Cela se comporte quelque peu différemment des opérations de repli implémentées pour les collections non distribuées dans des langages fonctionnels comme Scala.

Si je ne me trompe pas, même si l'API Spark ne l'exige pas, fold exige également que le f soit commutatif. Parce que l'ordre dans lequel les partitions seront agrégées n'est pas garanti. Par exemple, dans le code suivant, seule la première impression est triée:

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("")(_ + _))
}  

Imprimer:

abcdefghijklmnopqrstuvwxyz

abcghituvjklmwxyzqrsdefnop

defghinopjklmqrstuvabcwxyz

folddans Apache Spark n'est pas la même chose que foldsur les collections non distribuées. En fait, il faut une fonction commutative pour produire des résultats déterministes:

Cela se comporte quelque peu différemment des opérations de repli implémentées pour les collections non distribuées dans des langages fonctionnels comme Scala. Cette opération de pliage peut être appliquée aux partitions individuellement, puis plier ces résultats dans le résultat final, plutôt que d'appliquer le pli à chaque élément séquentiellement dans un certain ordre défini. Pour les fonctions non commutatives, le résultat peut différer de celui d'un repli appliqué à une collection non distribuée.

Cela a été démontré par Mishael Rosenthal et suggéré par Make42 dans son commentaire .

Il a été suggéré que le comportement observé est lié au HashPartitionermoment où, en fait, parallelizeil ne mélange pas et n'utilise pas HashPartitioner.

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)

Expliqué:

Structure de fold pour 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
}

est le même que la structure dereduce pour 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"))
}

où runJobest exécuté sans tenir compte de l'ordre de partition et entraîne le besoin d'une fonction commutative.

foldPartitionet reducePartitionsont équivalents en termes d'ordre de traitement et effectivement (par héritage et délégation) mis en œuvre par reduceLeftetfoldLeft sur TraversableOnce.

Conclusion: foldsur RDD ne peut pas dépendre de l'ordre des morceaux et nécessite commutativité et associativité .

Une autre différence pour Scalding est l'utilisation de combineurs dans Hadoop.

Imaginez que votre opération soit monoïde commutative, avec la réduction, elle sera également appliquée du côté de la carte au lieu de mélanger / trier toutes les données vers les réducteurs. Ce n'est pas le cas avec foldLeft .

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

Il est toujours bon de définir vos opérations comme monoïdes dans Scalding.