Scala聚合函数的示例

Question

我一直在寻找,我无法aggregate在Scala中找到我能理解的函数的示例或讨论.它看起来非常强大.

可以使用此函数来减少元组的值以生成多图类型集合吗？例如:

val list = Seq(("one", "i"), ("two", "2"), ("two", "ii"), ("one", "1"), ("four", "iv"))

应用聚合后:

Seq(("one" -> Seq("i","1")), ("two" -> Seq("2", "ii")), ("four" -> Seq("iv"))

此外,您还可以给实例参数z,segop和combop？我不清楚这些参数是做什么的.

Answer 1

让我们来看看一些ascii艺术是否有帮助.考虑以下类型的签名aggregate:

def aggregate [B] (z: B)(seqop: (B, A) ? B, combop: (B, B) ? B): B

另外,请注意,A指的是集合的类型.所以,假设我们在这个集合中有4个元素,那么aggregate可能会像这样工作:

z   A   z   A   z   A   z   A
 \ /     \ /seqop\ /     \ /    
  B       B       B       B
    \   /  combop   \   /
      B _           _ B
         \ combop  /
              B

让我们看一个实际的例子.说我有一个GenSeq("This", "is", "an", "example"),我想知道它中有多少个字符.我可以写下面的内容:

请注意par以下代码段的使用.传递给聚合的第二个函数是在计算各个序列之后调用的函数.Scala只能为可以并行化的集合执行此操作.

import scala.collection.GenSeq
val seq = GenSeq("This", "is", "an", "example")
val chars = seq.par.aggregate(0)(_ + _.length, _ + _)

所以,首先它会计算:

0 + "This".length     // 4
0 + "is".length       // 2
0 + "an".length       // 2
0 + "example".length  // 7

它无法预测它的作用(结合结果的方法不止一种),但它可能会这样做(如上面的ascii艺术):

4 + 2 // 6
2 + 7 // 9

在这一点上,它结束了

6 + 9 // 15

这给出了最终结果.现在,这在结构上有点类似foldLeft,但它有一个额外的功能(B, B) => B,折叠没有.但是,此功能使其能够并行工作!

例如,考虑四个计算中的每个计算初始计算彼此独立,并且可以并行完成.接下来的两个(产生6和9)可以在它们所依赖的计算完成后启动,但这两个也可以并行运行.

如上所述并行化的7次计算可以采用与3次串行计算相同的时间.

实际上,如此小的集合,同步计算的成本将足以消除任何收益.此外,如果你折叠它,它只需要4次计算.然而,一旦你的收藏品变大,你就会开始看到一些真正的收获.

另一方面,考虑一下foldLeft.因为它没有附加功能,所以它无法并行化任何计算:

(((0 + "This".length) + "is".length) + "an".length) + "example".length

必须先计算每个内括号,然后再进行外括号.

Answer 2

聚合函数不会这样做(除了它是一个非常通用的函数,它可以用来做到这一点).你想要的groupBy.至少接近.当你从a开始Seq[(String, String)],并通过取元组中的第一个项目进行分组(也就是说(String, String) => String),它将返回a Map[String, Seq[(String, String)]).然后,您必须丢弃Seq [String,String]]值中的第一个参数.

所以

list.groupBy(_._1).mapValues(_.map(_._2))

你有一个Map[String, Seq[(String, String)].如果你想要一个Seq而不是Map,请调用toSeq结果.我不认为你对Seq产生的订单有保证

聚合是一个更难的功能.

首先考虑reduceLeft和reduceRight.设为asA类型的非空元素序列,as = Seq(a1, ... an)并将两种类型的元素组合A成一种方式.我会将其视为二元运算符f: (A,A) => A,A而不是@.a1 @ a2 会计算f(a1, a2).as.reduceLeft(@)将括号放在另一个方向,(((a1 @ a2) @ a3)... @ an).如果reduceRight恰好是关联的,则不关心括号.人们可以把它计算为(a1 @ (a2 @... @ an))))(在2个大的parantheses中也会有parantheses,但是我们不关心那个).然后可以并行执行这两个部分,而reduceLeft或reduceRight中的嵌套包围强制执行完全顺序计算.但并行计算只有在@已知是关联的情况下才有可能,而reduceLeft方法无法知道.

仍然可以有方法(a1 @... @ ap) @ (ap+1 @...@an),其调用者将负责确保操作是关联的.然后@按照它认为合适的顺序排序,可能并行执行.的确,有这样一种方法.

然而,各种减少方法存在限制.Seq的元素只能组合成相同类型的结果:reduce必须reduce.但是人们可能会有将它们组合成一个更普遍的问题@.一个以(A,A) => A类型值开头B,并将其与序列的每个元素组合.操作员b是B,并且计算一个@.(B,A) => B那样做.(((b @ a1) @ a2) ... @ an)做同样的事情,但从开始foldLeft.在那里,该foldRight操作没有机会成为关联的.当一个人写道时an,它必须意味着@,就像b @ a1 @ a2输入错误一样.所以foldLeft和foldRight必须是顺序的.

然而,假设每个(b @ a1) @ a2都可以变成a (a1 @ a2),让我们用它来写A,B是类型的!.此外,假设有一个a!操作B,实际上+就是这样.不是将元素与@组合,而是可以先将它们全部转换为B ,然后将它们组合起来.那就是.如果是关联的,那么可以使用reduce完成,而不是顺序:as.map(!).reduce(+).可能存在一种假设的方法as.associativeFold(b,!,+).(B,B) => B@b @ ab + a!!+

Aggregate非常接近.然而,可能有一种更有效的方法来实现as.map(!).reduceLeft(+),+例如,如果type b@a是b+a!,而b @ a是a :: b,B则将是List[A],并且a!将是a::Nil.a :: b比(a :: Nil)::: b更好.从相关性中获益,但仍使用b1 + b2,首先分裂b2 ::: b1,进入@,然后回去用b + a1! + ... + an!用(b + a1! + ap!) + (ap+1! + ..+ an!).一个还需要!在ap + 1上,因为必须以某些b开头.+仍然是必要的,出现在parantheses之间.要做到这一点,@可以改为(b @ a1 @ an) + (ap+1! @ @ an).

回到as.associativeFold(!, +).as.optimizedAssociativeFold(b, !, @, +)是一个关联的,或者等价的,+是一个半群.在实践中,编程中使用的大多数半群也恰好是幺半群,即它们在B中包含中性元素+(对于零),因此对于每个(B, +),z= b= z + b.在这种情况下,b + z可能会有意义的操作b.此外,由于z是中性元件!,其是a! = z @ a.所以总是可以用z开始聚合.如果b @ a1 ..@ an = (b + z) @ a1 @ an需要,你最后会这样做b + (z + a1 @ an).有了所有这些假设,我们就可以做到b.这是做什么的b + result.s.aggregate(z, @, +)是aggregate参数(在施加序列 @),并且seqop被z @ a1 @ a2 @ ap(施加到已经部分组合的结果,如在+).

总结一下,combop计算出与(z + a1@...@ap) + (z + ap+1@...@an)提供的相同的东西

• as.aggregate(z)(seqop, combop) 是一个幺半群
• as.foldLeft(z)( seqop)

聚合实现可以使用combop的关联性来按照自己喜欢的方式对计算进行分组(但不是交换元素,+不能交换,:::不是).它可以并行运行它们.

最后,解决使用的初始问题(B, combop, z)留给读者练习.提示:实现使用seqop(b,a) = combop(b, seqop(z,a)),然后查找aggregate并foldLeft满足上述条件.

Answer 3

具有类型A的元素的集合的签名是:

def aggregate [B] (z: B)(seqop: (B, A) ? B, combop: (B, B) ? B): B 

• z是B型作为中性元素的对象.如果你想计算一些东西,你可以使用0,如果你想建立一个列表,从一个空列表开始,等等.
• segop与传递给fold方法的函数类似.它需要两个参数,第一个是与您传递的中性元素相同的类型,表示在上一次迭代中已经聚合的东西,第二个是集合的下一个元素.结果也必须是类型B.
• combop:是一个将两个结果合二为一的函数.

在大多数集合中,聚合实现TraversableOnce如下:

  def aggregate[B](z: B)(seqop: (B, A) => B, combop: (B, B) => B): B 
    = foldLeft(z)(seqop)

因此combop被忽略了.但是,它对并行集合很有意义,因为seqop它将首先在本地并行应用,然后combop调用以完成聚合.

因此,对于您的示例,您可以先尝试折叠:

val seqOp = 
  (map:Map[String,Set[String]],tuple: (String,String)) => 
    map + ( tuple._1 -> ( map.getOrElse( tuple._1, Set[String]() ) + tuple._2 ) )


list.foldLeft( Map[String,Set[String]]() )( seqOp )
// returns: Map(one -> Set(i, 1), two -> Set(2, ii), four -> Set(iv))

然后你必须找到一种折叠两个multimaps的方法:

val combOp = (map1: Map[String,Set[String]], map2: Map[String,Set[String]]) =>
       (map1.keySet ++ map2.keySet).foldLeft( Map[String,Set[String]]() ) { 
         (result,k) => 
           result + ( k -> ( map1.getOrElse(k,Set[String]() ) ++ map2.getOrElse(k,Set[String]() ) ) ) 
       } 

现在,您可以并行使用聚合:

list.par.aggregate( Map[String,Set[String]]() )( seqOp, combOp )
//Returns: Map(one -> Set(i, 1), two -> Set(2, ii), four -> Set(iv))

将方法"par"应用于列表,从而使用列表的并行集合(scala.collection.parallel.immutable.ParSeq)来真正利用多核处理器.如果没有"par",则不会有任何性能提升,因为并行集合上没有聚合.

Answer 4

aggregate是喜欢foldLeft但可以并行执行.

正如missfactor所说,线性版本aggregate(z)(seqop, combop)相当于foldleft(z)(seqop).然而,在并行的情况下,这是不切实际的,我们需要不仅将下一个元素与前一个结果组合在一起(如在正常折叠中),但我们希望将iterable拆分为子迭代,我们称之为聚合,需要再次结合这些.(从左到右的顺序但不是关联的,因为我们可能在迭代的第一部分之前组合了最后的部分.)这种重新组合通常是非平凡的,因此,需要一种方法(S, S) => S来实现它.

定义ParIterableLike是:

def aggregate[S](z: S)(seqop: (S, T) => S, combop: (S, S) => S): S = {
  executeAndWaitResult(new Aggregate(z, seqop, combop, splitter))
}

确实使用combop.

供参考,Aggregate定义为:

protected[this] class Aggregate[S](z: S, seqop: (S, T) => S, combop: (S, S) => S, protected[this] val pit: IterableSplitter[T])
  extends Accessor[S, Aggregate[S]] {
    @volatile var result: S = null.asInstanceOf[S]
    def leaf(prevr: Option[S]) = result = pit.foldLeft(z)(seqop)
    protected[this] def newSubtask(p: IterableSplitter[T]) = new Aggregate(z, seqop, combop, p)
    override def merge(that: Aggregate[S]) = result = combop(result, that.result)
}

最重要的部分是merge其中combop与两个子结果应用.