¿Diferencia entre reducir y doblar a la izquierda / doblar en la programación funcional (en particular, las API de Scala y Scala)?

¿Por qué Scala y frameworks como Spark y Scalding tienen ambos reducey foldLeft? Entonces, ¿cuál es la diferencia entre reducey fold?

reducir vs doblarIzquierda

Una gran diferencia, que no se menciona claramente en ninguna otra respuesta de desbordamiento de pila relacionada con este tema, es que reducese le debe dar un monoide conmutativo , es decir, una operación que sea tanto conmutativa como asociativa. Esto significa que la operación se puede paralelizar.

Esta distinción es muy importante para Big Data / MPP / computación distribuida, y la razón por la que reduceincluso existe. La colección se puede cortar y la reducepuede operar en cada fragmento, luego reducepuede operar en los resultados de cada fragmento; de hecho, el nivel de fragmentación no tiene por qué detenerse en un nivel de profundidad. También podríamos cortar cada trozo. Esta es la razón por la que sumar enteros en una lista es O (log N) si se le da un número infinito de CPU.

Si solo miras las firmas, no hay razón para reduceque exista porque puedes lograr todo lo que puedas reducecon un foldLeft. La funcionalidad de foldLeftes mayor que la funcionalidad de reduce.

Pero no puede paralelizar a foldLeft, por lo que su tiempo de ejecución es siempre O (N) (incluso si alimenta un monoide conmutativo). Esto se debe a que se supone que la operación no es un monoide conmutativo y, por lo tanto, el valor acumulado se calculará mediante una serie de agregaciones secuenciales.

foldLeftno asume conmutatividad ni asociatividad. Es la asociatividad la que da la capacidad de dividir la colección, y es la conmutatividad la que facilita la acumulación porque el orden no es importante (por lo que no importa en qué orden agregar cada uno de los resultados de cada uno de los fragmentos). Estrictamente hablando, la conmutatividad no es necesaria para la paralelización, por ejemplo, los algoritmos de clasificación distribuida, simplemente hace que la lógica sea más fácil porque no necesita ordenar sus fragmentos.

Si echas un vistazo a la documentación de Spark reduce, dice específicamente "... operador binario conmutativo y asociativo"

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

Aquí hay una prueba de que reduceNO es solo un caso especial 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

reducir vs doblar

Aquí es donde se acerca un poco más a las raíces matemáticas / FP, y es un poco más complicado de explicar. Reducir se define formalmente como parte del paradigma MapReduce, que se ocupa de las colecciones sin orden (multisets), Fold se define formalmente en términos de recursividad (ver catamorfismo) y, por lo tanto, asume una estructura / secuencia para las colecciones.

No existe un foldmétodo en Scalding porque bajo el modelo de programación (estricto) Map Reduce no podemos definir foldporque los fragmentos no tienen un orden y foldsolo requieren asociatividad, no conmutatividad.

En pocas palabras, reducefunciona sin un orden de acumulación, foldrequiere un orden de acumulación y es ese orden de acumulación el que necesita un valor cero, NO la existencia del valor cero que los distingue. Estrictamente hablando, reduce debería funcionar en una colección vacía, porque su valor cero puede deducirse tomando un valor arbitrario xy luego resolviendo x op y = x, pero eso no funciona con una operación no conmutativa, ya que puede existir un valor cero izquierdo y derecho que son distintos (es decir x op y != y op x). Por supuesto, Scala no se molesta en averiguar cuál es este valor cero, ya que eso requeriría hacer algunas matemáticas (que probablemente no se puedan calcular), por lo que solo lanza una excepción.

Parece (como suele ser el caso en etimología) que este significado matemático original se ha perdido, ya que la única diferencia obvia en la programación es la firma. El resultado es que se reduceha convertido en un sinónimo foldde MapReduce, en lugar de conservar su significado original. Ahora, estos términos a menudo se usan indistintamente y se comportan igual en la mayoría de las implementaciones (ignorando las colecciones vacías). La rareza se ve agravada por peculiaridades, como en Spark, que ahora abordaremos.

Así que la chispa no tiene unafold , pero el orden en el que se combinan sub resultados (uno para cada partición) (en el momento de la escritura) es el mismo orden en que se hayan completado las tareas - y por lo tanto no determinista. Gracias a @CafeFeed por señalar ese folduso runJob, que después de leer el código me di cuenta de que no es determinista. Spark crea más confusión porque tiene un treeReducepero no treeFold.

Conclusión

Hay una diferencia entre reduceyfold incluso cuando se aplica a secuencias no vacías. El primero se define como parte del paradigma de programación MapReduce en colecciones con orden arbitrario ( http://theory.stanford.edu/~sergei/papers/soda10-mrc.pdf ) y se debe asumir que los operadores son conmutativos además de ser asociativo para dar resultados deterministas. Este último se define en términos de catomorfismos y requiere que las colecciones tengan una noción de secuencia (o se definan de forma recursiva, como listas enlazadas), por lo que no requieren operadores conmutativos.

En la práctica, debido a la naturaleza no matemática de la programación, reduce y foldtienden a comportarse de la misma manera, ya sea correctamente (como en Scala) o incorrectamente (como en Spark).

Extra: Mi opinión sobre la API Spark

Mi opinión es que se evitaría la confusión si el uso del término foldse eliminara por completo en Spark. Al menos Spark tiene una nota en su documentación:

Esto se comporta de forma algo diferente a las operaciones de plegado implementadas para colecciones no distribuidas en lenguajes funcionales como Scala.

Si no me equivoco, aunque la API de Spark no lo requiere, fold también requiere que la f sea conmutativa. Porque el orden en el que se agregarán las particiones no está asegurado. Por ejemplo, en el siguiente código solo se ordena la primera impresión:

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

Imprimir:

ABCDEFGHIJKLMNOPQRSTU VWXYZ

abcghituvjklmwxyzqrsdefnop

defghinopjklmqrstuvabcwxyz

folden Apache Spark no es lo mismo que folden colecciones no distribuidas. De hecho , requiere una función conmutativa para producir resultados deterministas:

Esto se comporta de forma algo diferente a las operaciones de plegado implementadas para colecciones no distribuidas en lenguajes funcionales como Scala. Esta operación de plegado se puede aplicar a particiones individualmente y luego plegar esos resultados en el resultado final, en lugar de aplicar el plegado a cada elemento secuencialmente en algún orden definido. Para las funciones que no son conmutativas, el resultado puede diferir del de un pliegue aplicado a una colección no distribuida.

Esto ha sido demostrado por Mishael Rosenthal y sugerido por Make42 en su comentario .

Se ha sugerido que el comportamiento observado está relacionado con HashPartitionercuando en realidad parallelizeno se baraja y no se usa 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)

Explicado:

Estructura defold para 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
}

es la misma que la estructura dereduce 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"))
}

donde runJobse realiza sin tener en cuenta el orden de partición y resulta en la necesidad de una función conmutativa.

foldPartitiony reducePartitionson equivalentes en términos de orden de procesamiento y efectivamente (por herencia y delegación) implementados por reduceLefty foldLeftsucesivamente TraversableOnce.

Conclusión: folden RDD no puede depender del orden de los trozos y necesita conmutatividad y asociatividad .

Otra diferencia para Scalding es el uso de combinadores en Hadoop.

Imagine que su operación es un monoide conmutativo, con reducir se aplicará también en el lado del mapa en lugar de barajar / ordenar todos los datos a los reductores. Con foldLeft este no es el caso.

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

Siempre es una buena práctica definir sus operaciones como monoide en Scalding.