Scala - 懒惰地对有序迭代器进行分组答案

【问题标题】：Scala - grouping on an ordered iterator lazilyScala - 懒惰地对有序迭代器进行分组
【发布时间】：2016-03-01 11:56:03
【问题描述】：

我有一个Iterator[Record]，它是通过这种方式在record.id 上订购的：

record.id=1
record.id=1
...
record.id=1
record.id=2
record.id=2
..
record.id=2

特定 ID 的记录可能会出现很多次，所以我想编写一个函数，将此迭代器作为输入，并以惰性方式返回 Iterator[Iterator[Record]] 输出。

我能够提出以下建议，但在 500K 左右的记录后，它在 StackOverflowError 上失败了：

def groupByIter[T, B](iterO: Iterator[T])(func: T => B): Iterator[Iterator[T]] = new Iterator[Iterator[T]] {
    var iter = iterO
    def hasNext = iter.hasNext

    def next() = {
      val first = iter.next()
      val firstValue = func(first)
      val (i1, i2) = iter.span(el => func(el) == firstValue)
      iter = i2
      Iterator(first) ++ i1
    }
  }

我做错了什么？

【问题讨论】：

分组假设您遍历整个集合并对所有值进行分组。我认为懒惰是不可能的

标签： scala stream iterator lazy-evaluation

【解决方案1】：

这里的问题是每个Iterator.span 调用都会为trailing 迭代器创建另一个堆叠闭包，并且没有任何蹦床很容易溢出。

实际上我不认为有一个实现，它不记忆前缀迭代器的元素，因为可以在前缀耗尽之前访问跟随的迭代器。

即使在.span implementation 中，也有一个Queue 来记忆Leading 定义中的元素。

我可以想象的最简单的实现是通过Stream 实现的。

implicit class StreamChopOps[T](xs: Stream[T]) {
  def chopBy[U](f: T => U): Stream[Stream[T]] = xs match {
    case x #:: _ =>
      def eq(e: T) = f(e) == f(x)
      xs.takeWhile(eq) #:: xs.dropWhile(eq).chopBy(f)
    case _ => Stream.empty
  }
}

虽然它可能不是最高效的，因为它记住了很多。但是通过适当的迭代，GC 应该处理过多的中间流的问题。

您可以将其用作myIterator.toStream.chopBy(f)

简单检查验证以下代码可以在没有 SO 的情况下运行

Iterator.fill(10000000)(Iterator(1,1,2)).flatten //1,1,2,1,1,2,...
  .toStream.chopBy(identity)                     //(1,1),(2),(1,1),(2),...
  .map(xs => xs.sum * xs.size).sum               //60000000

【讨论】：

有一个小错字。您必须在 chopBy 中应用 f，例如 xs.takeWhile(e => f(e) == f(x)) #:: xs.dropWhile(e => f(e) == f(x)).chopBy(f)。
@jeffreyveon 绝对！谢谢
我尝试在调用链的末尾将建议的方法与 foreach 一起使用，但出现内存溢出。看起来它首先迭代整个流，同时根据需要对其进行切碎，并且只有在所有切碎都准备好之后才对它们中的每一个调用 foreach 。所以实际上流的行为是丢失的，因为所有的流都是一次加载到内存中的。
@alexanderostrikov 它绝对不能处理很长的相似元素序列，因为它在发送到下游之前等待每个元素结束。
@Odomontois 试图用可以发布在这里的简单代码重现问题，我发现了一些我无法根据我目前的 scala 知识解释的东西。所以实际上当我这样称呼它时，您的代码看起来像预期的那样工作（对不起第一条评论）：Iterator.range(0,1000000000,1).toStream .chopBy(x=> x % 20 == 0) .map(s => s.toList) .foreach(println) 但是当我这样称呼它时：val chopped= Iterator.range(0,1000000000,1).toStream .chopBy(x=> x % 20 == 0) .map(s => s.toList) chopped.foreach(println) 我得到 OutOfMemoryError

【解决方案2】：

受@Odomontois 实现的chopBy 的启发，这里是我为Iterator 实现的chopBy。当然，每个块都应该适合分配的内存。它看起来不是很优雅，但似乎可以工作:)

implicit class IteratorChopOps[A](toChopIter: Iterator[A]) {

 def chopBy[U](f: A => U) = new Iterator[Traversable[A]] {
  var next_el: Option[A] = None
  @tailrec
  private def accum(acc: List[A]): List[A] = {
    next_el = None
    val new_acc = hasNext match {
      case true =>
        val next = toChopIter.next()
        acc match {
          case Nil =>
            acc :+ next
          case _ MatchTail t if (f(t) == f(next)) =>
            acc :+ next
          case _ =>
            next_el = Some(next)
            acc
        }
      case false =>
        next_el = None
        return acc
    }

    next_el match{
      case Some(_) =>
        new_acc
      case None => accum(new_acc)
    }
  }

  def hasNext = {
    toChopIter.hasNext || next_el.isDefined
  }
  def next: Traversable[A] = accum(next_el.toList)
}
}

这是一个匹配尾部的提取器：

object MatchTail {
  def unapply[A] (l: Traversable[A]) = Some( (l.init, l.last) )
}

【讨论】：