Scala：IterableをIterableのIterableに述語でグループ化する

Question

非常に大きなIteratorを分割して分割したい。私はアイテムを見て、新しいピースの始まりであれば真を返す述語を持っています。私は作品をイテレーターにする必要があります。なぜなら作品も記憶に収まらないからです。あなたのスタックを吹き飛ばす再帰的な解決策が気になるほど多くの部分があります。状況は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

あなたは固有の問題を抱えています。は、複数のイテレータを取得できることを意味します。 Iteratorは、一度しか通過できないことを意味します。つまり、Iterable[Iterable[T]]はIterator[Iterable[T]]を生成できるはずです。しかし、これが要素（Iterable[T]）を返し、複数のイテレータを要求すると、リストの結果をキャッシュする（あまりにも大きい）またはオリジナルのiterableを呼び出し、絶対に通過することなく、すべてをもう一度（非常に非効率的）。

だから、はになる可能性がありますが、私はあなたの問題を別の方法で考えなければならないと思います。

Seqで始めることができる場合は、範囲としてサブセットを取得できます。

あなたはすでにあなたのiterableを使用する方法を知っている場合は、「真の」オフstarts火災ハンドラのセットを介してそれぞれの時間をインクリメントする方法

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

を書くことができます。あなたが1回の掃引で処理を行うことができる方法があれば、このようなものが行く方法です。 （ただし、ハンドラは変更可能なデータ構造や変数を使用して状態を保存する必要があります）

外側リストの反復を許可して内側リストを破損する可能性がある場合は、追加の制約を持つ​​を持つことができます後のサブイテレータには、すべての以前のサブイテレータが無効です。

---

はここで最後のタイプのソリューションです（Iterator[T]Iterator[Iterator[T]]から; 1は、外層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オブジェクトがビューをサポートしている場合にのみ使用できます。この実装はIterableを反復処理し、IterableViewを生成します.IterableViewは反復処理が可能です。最初の要素が開始グループとしてテストされるかどうかは関係ありませんので、この実装は気にしません。

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

Answer 4

あなたはストリームを使用することにより、低メモリフットプリントを維持することができます。もう一度イテレータを使用する場合は、result.toIteratorを使用します。

ストリームでは、変更可能な状態はなく、単一の条件付きで、Jay Hackerのソリューションとほぼ同じくらい簡潔です。スタックオーバーフローペーストするにはあまりにも多くの

scala> batchBy((1 to Int.MaxValue).iterator)(_ % (Int.MaxValue/2) == 0) 
     .foreach{case(_,group) => println(group.sum)} 
-1610612735 
1073741823 
-536870909 
2147483646 
2147483647 

第二のテストプリント：

def batchBy[A,B](iter: Iterator[A])(f: A => B): Stream[(B, Iterator[A])] = { 
    val base = iter.buffered 
    val empty = Stream.empty[(B, Iterator[A])] 

    def getBatch(key: B) = { 
     Iterator(base.next()) ++ new Iterator[A] { 
     def hasNext: Boolean = base.hasNext && (f(base.head) == key) 
     def next(): A = base.next() 
     } 
    } 

    def next(skipList: Option[Iterator[A]] = None): Stream[(B, Iterator[A])] = { 
     skipList.foreach{_.foreach{_=>}} 

     if (base.isEmpty) empty 
     else { 
     val key = f(base.head) 
     val batch = getBatch(key) 

     Stream.cons((key, batch), next(Some(batch))) 
     } 
    } 

    next() 
    } 

は、私がテストを実施しました。

Answer 5

import scala.collection.mutable.ArrayBuffer 

object GroupingIterator { 

    /** 
    * Create a new GroupingIterator with a grouping predicate. 
    * 
    * @param it The original iterator 
    * @param p Predicate controlling the grouping 
    * @tparam A Type of elements iterated 
    * @return A new GroupingIterator 
    */ 
    def apply[A](it: Iterator[A])(p: (A, IndexedSeq[A]) => Boolean): GroupingIterator[A] = 
    new GroupingIterator(it)(p) 
} 

/** 
* Group elements in sequences of contiguous elements that satisfy a predicate. The predicate 
* tests each single potential next element of the group with the help of the elements grouped so far. 
* If it returns true, the potential next element is added to the group, otherwise 
* a new group is started with the potential next element as first element 
* 
* @param self The original iterator 
* @param p Predicate controlling the grouping 
* @tparam A Type of elements iterated 
*/ 
class GroupingIterator[+A](self: Iterator[A])(p: (A, IndexedSeq[A]) => Boolean) extends Iterator[IndexedSeq[A]] { 

    private[this] val source = self.buffered 
    private[this] val buffer: ArrayBuffer[A] = ArrayBuffer() 

    def hasNext: Boolean = source.hasNext 

    def next(): IndexedSeq[A] = { 
    if (hasNext) 
     nextGroup() 
    else 
     Iterator.empty.next() 
    } 

    private[this] def nextGroup(): IndexedSeq[A] = { 
    assert(source.hasNext) 

    buffer.clear() 
    buffer += source.next 

    while (source.hasNext && p(source.head, buffer)) { 
     buffer += source.next 
    } 

    buffer.toIndexedSeq 
    } 
}