Scala:通过谓词将Iterable分组为Iterable of Iterable

Question

我有非常大的迭代器，我想把它们分成几个部分。我有一个谓词，它查看一个项目，如果它是一个新项目的开始，则返回true。我需要这些片段成为迭代器，因为即使是这些片段也无法放入内存。有如此多的部分，我会提防递归解决方案会毁了你的堆栈。这种情况类似于this question，但是我需要迭代器而不是列表，并且“标记”(谓词为真的项)出现(并且应该包括)在片段的开头。产生的迭代器将只按顺序使用，尽管有些可能根本不使用，并且它们应该只使用O(1)内存。我想这意味着它们应该共享相同的底层迭代器。性能很重要。

如果我要尝试函数签名，它将是这样的：

def groupby[T](iter: Iterator[T])(startsGroup: T => Boolean): Iterator[Iterator[T]] = ...

我很想使用takeWhile，但它丢失了最后一个元素。我研究了span，但它会缓冲结果。我目前最好的想法是使用BufferedIterator，但也许还有更好的方法。

你会知道你做的是对的，因为这样的事情不会让你的JVM崩溃：

groupby((1 to Int.MaxValue).iterator)(_ % (Int.MaxValue / 2) == 0).foreach(group => println(group.sum))
groupby((1 to Int.MaxValue).iterator)(_ % 10 == 0).foreach(group => println(group.sum))

Answer 1

这是我使用BufferedIterator的解决方案。它不能让你正确地跳过迭代器，但它是相当简单和实用的。即使是!startsGroup(first)，第一个元素也会进入一个组。

def groupby[T](iter: Iterator[T])(startsGroup: T => Boolean): Iterator[Iterator[T]] =
  new Iterator[Iterator[T]] {
    val base = iter.buffered
    override def hasNext = base.hasNext  
    override def next() = Iterator(base.next()) ++ new Iterator[T] {
      override def hasNext = base.hasNext && !startsGroup(base.head) 
      override def next() = if (hasNext) base.next() else Iterator.empty.next()
    }
  }

更新:保持一个小的状态可以让你跳过迭代器，并防止人们扰乱以前的迭代器：

def groupby[T](iter: Iterator[T])(startsGroup: T => Boolean): Iterator[Iterator[T]] =
new Iterator[Iterator[T]] {
  val base = iter.buffered
  var prev: Iterator[T] = Iterator.empty
  override def hasNext = base.hasNext  
  override def next() = {
    while (prev.hasNext) prev.next()        // Exhaust previous iterator; take* and drop* do NOT always work!!  (Jira SI-5002?)
    prev = Iterator(base.next()) ++ new Iterator[T] {
      var hasMore = true
      override def hasNext = { hasMore = hasMore && base.hasNext && !startsGroup(base.head) ; hasMore } 
      override def next() = if (hasNext) base.next() else Iterator.empty.next()
    }
    prev
  }
}

Answer 2

你有一个固有的问题。Iterable意味着你可以获得多个迭代器。Iterator表示您只能通过一次。这意味着您的Iterable[Iterable[T]]应该能够生成Iterator[Iterable[T]]s，但是当它返回一个元素--一个Iterable[T]--并且需要多个迭代器时，底层的单个迭代器要么缓存列表的结果(太大)，要么调用原始的迭代器并重新遍历所有内容(非常低效)。

所以，虽然你可以这样做，但我认为你应该用不同的方式来思考你的问题。

如果您可以从Seq开始，那么您可以获取范围形式的子集。

如果你已经知道如何使用你的iterable，你可以写一个方法

def process[T](source: Iterable[T])(starts: T => Boolean)(handlers: T => Unit *)

每当starts发出“真”值时，该值就会递增到处理程序集。如果有任何方法可以在一次扫描中完成处理，那么就可以使用下面这样的方法。(但是，您的处理程序必须通过可变数据结构或变量保存状态。)

如果您可以允许在外部列表上进行迭代以破坏内部列表，那么您可以拥有一个带有附加约束的Iterable[Iterator[T]]，一旦您迭代到后面的子迭代器，所有以前的子迭代器都是无效的。

下面是最后一种类型的解决方案(从Iterator[T]到Iterator[Iterator[T]]；可以将其包装为Iterable )。

class GroupedBy[T](source: Iterator[T])(starts: T => Boolean)
extends Iterator[Iterator[T]] {
  private val underlying = source
  private var saved: T = _
  private var cached = false
  private var starting = false
  private def cacheNext() {
    saved = underlying.next
    starting = starts(saved)
    cached = true
  }
  private def oops() { throw new java.util.NoSuchElementException("empty iterator") }
  // Comment the next line if you do NOT want the first element to always start a group
  if (underlying.hasNext) { cacheNext(); starting = true }
  def hasNext = {
    while (!(cached && starting) && underlying.hasNext) cacheNext()
    cached && starting
  }
  def next = {
    if (!(cached && starting) && !hasNext) oops()
    starting = false
    new Iterator[T] {
      var presumablyMore = true
      def hasNext = {
        if (!cached && !starting && underlying.hasNext && presumablyMore) cacheNext()
        presumablyMore = cached && !starting
        presumablyMore
      }
      def next = {
        if (presumablyMore && (cached || hasNext)) { 
          cached = false
          saved
        }
        else oops()
      }
    }
  }
}

Answer 3

如果你正在寻找内存约束，那么下面的方法将会起作用。只有当底层可迭代对象支持视图时，才能使用它。这个实现将在Iterable上迭代，然后生成IterableViews，然后可以迭代它。此实现不关心第一个元素是否作为开始组进行测试，因为它将是无关的。

def groupby[T](iter: Iterable[T])(startsGroup: T => Boolean): Iterable[Iterable[T]] = new Iterable[Iterable[T]] {
  def iterator = new Iterator[Iterable[T]] {
    val i = iter.iterator
    var index = 0
    var nextView: IterableView[T, Iterable[T]] = getNextView()
    private def getNextView() = {
      val start = index
      var hitStartGroup = false
      while ( i.hasNext && ! hitStartGroup ) {
        val next = i.next()
        index += 1
        hitStartGroup = ( index > 1 && startsGroup( next ) )
      }
      if ( hitStartGroup ) {
        if ( start == 0 ) iter.view( start, index - 1 )
        else iter.view( start - 1, index - 1 )
      } else { // hit end
        if ( start == index ) null
        else if ( start == 0 ) iter.view( start, index )
        else iter.view( start - 1, index )
      }
    }
    def hasNext = nextView != null
    def next() = {
      if ( nextView != null ) {
        val next = nextView
        nextView = getNextView()
        next
      } else null
    }
  }
}