是否可以在 Scala 中将多个 map 和 reduce 函数组合成一次传递？

Question

我有多个地图函数在相同的数据上运行，我希望它们一次运行。我正在寻找一种通用的方法来做到这一点。

val fruits: Seq[String] = Seq("apple", "banana", "cherry")

def mapF(s: String): Char = s.head
def reduceF(c1: Char, c2: Char): Char = if(c1 > c2) c1 else c2

def mapG(s: String): Int = s.length
def reduceG(i1: Int, i2: Int): Int = i1 + i2

val largestStartingChar = fruits.map(mapF).reduce(reduceF)
val totalStringLength = fruits.map(mapG).reduce(reduceG)

我想减少通过fruits 的次数。我可以使这两个地图通用并像这样减少：

def productMapFunction[A, B, C](f: A=>B, g: A=>C): A => (B, C) = {
  x => (f(x), g(x))
}

def productReduceFunction[T, U](f: (T, T)=>T, g: (U, U) => U):
    ((T,U), (T,U)) => (T, U) = {
  (tu1, tu2) => (f(tu1._1, tu2._1), g(tu1._2, tu2._2))
}

val xMapFG = productMapFunction(mapF, mapG)
val xReduceFG = productReduceFunction(reduceF, reduceG)

val (largestStartingChar2, totalStringLength2) = 
  fruits.map(xMapFG).reduce(xReduceFG))

我想更通用地执行此操作，使用任意数量的 map 和 reduce 函数，但我不确定如何进行，或者是否可行。

Answer 1

有趣的问题！

我不知道标准库甚至 scalaz/cats 中有任何这样的实现。 这并不奇怪，因为如果您的列表不是很大，您可以按顺序执行 map-reduce，我什至不确定构造大量中间对象的开销是否会小于多次遍历列表的开销。

如果列表可能不适合内存，您应该使用流式库之一 (fs2/zio-streams/akka-streams)

虽然如果您的输入是 Iterator 而不是 List，那么这样的功能会很有用。

关于这个问题有一篇有趣的文章： https://softwaremill.com/beautiful-folds-in-scala/

tldr： Map-reduce 工作流程可以形式化如下：

trait Fold[I, O] {
  type M
  def m: Monoid[M]

  def tally: I => M
  def summarize: M => O
}

在你的情况下I = List[A]、tally = list => list.map(mapF)、summarize = list => list.reduce(reduceF)。

要使用 fold 的实例在 list 上运行 map-reduce，您需要运行

fold.summarize(fold.tally(list))

您可以对它们定义combine 操作： def combine[I, O1, O2](f1: Fold[I, O1], f2: Fold[I, O2]): Fold[I, (O1, O2)]

使用combine 几次会得到你想要的：

combine(combine(f1, f2), f3): Fold[I, ((O1, O2), O3)]

Answer 2

我认为你只是想重塑transducers。我已经有一段时间没有使用 Scala 了，但至少有 one implementation。

Answer 3

以下解决方案使用 Cats 2 和自定义类型 MapReduce。

可以通过函数reduce: (O, O) => O指定归约操作 或猫reducer: Semigroup[O]。 implicit def mapReduceApply[I]提供的Apply实例可以将多个MapReduce对象合二为一

import cats._
import cats.implicits._

trait MapReduce[I, O] {
  type R

  def reducer: Semigroup[R]

  def map: I => R

  def mapResult: R => O

  def apply(input: Seq[I]): O = mapResult(input.map(map).reduce(reducer.combine))
}

object MapReduce {
  def apply[I, O, _R](_reducer: Semigroup[_R], _map: I => _R, _mapResult: _R => O): MapReduce[I, O] =
    new MapReduce[I, O] {
      override type R = _R

      override def reducer = _reducer

      override def map = _map

      override def mapResult = _mapResult
    }

  def apply[I, O](map: I => O)(implicit r: Semigroup[O]): MapReduce[I, O] =
    MapReduce[I, O, O](r, map, identity)

  def apply[I, O](map: I => O, reduce: (O, O) => O): MapReduce[I, O] = {
    val reducer = new Semigroup[O] {
      override def combine(x: O, y: O): O = reduce(x, y)
    }
    MapReduce(map)(reducer)
  }

  implicit def mapReduceApply[I] =
    new Apply[({type F[X] = MapReduce[I, X]})#F] {
      override def map[A, B](f: MapReduce[I, A])(fn: A => B): MapReduce[I, B] =
        MapReduce(f.reducer, f.map, f.mapResult.andThen(fn))

      override def ap[A, B](ff: MapReduce[I, (A) => B])(fa: MapReduce[I, A]): MapReduce[I, B] =
        MapReduce(ff.reducer product fa.reducer,
          i => (ff.map(i), fa.map(i)),
          (t: (ff.R, fa.R)) => ff.mapResult(t._1)(fa.mapResult(t._2))
        )
    }

}

object MultiMapReduce extends App {

  val fruits: Seq[String] = Seq("apple", "banana", "cherry")

  def mapF(s: String): Char = s.head

  def reduceF(c1: Char, c2: Char): Char = if (c1 > c2) c1 else c2

  val biggestFirsChar = MapReduce(mapF, reduceF)
  val totalChars = MapReduce[String, Int](_.length) // (Semigroup[Int]) reduce by _ + _
  def count[A] = MapReduce[A, Int](_ => 1)

  val multiMapReduce = (biggestFirsChar, totalChars, count[String]).mapN((_, _, _))
  println(multiMapReduce(fruits))

  val sum = MapReduce[Double, Double](identity)
  val average = (sum, count[Double]).mapN(_ / _)
  println(sum(List(1, 2, 3, 4)))
  println(average(List(1, 2, 3, 4)))

}

GitHub 上也提供可运行版本。