Scala: Raggruppa un Iterable in Iterable of Iterables secondo un predicato

Question

Ho degli iteratori molto grandi che voglio dividere in pezzi. Ho un predicato che guarda un oggetto e restituisce true se è l'inizio di un nuovo pezzo. Ho bisogno che i pezzi siano degli iteratori, perché anche i pezzi non si adatteranno alla memoria. Ci sono così tanti pezzi che vorrei diffidare di una soluzione ricorsiva che spazza via il tuo stack. La situazione è simile a this question, ma ho bisogno di Iterators anziché di Lists e le "sentinelle" (voci per le quali il predicato è vero) si verificano (e dovrebbero essere incluse) all'inizio di un pezzo. Gli iteratori risultanti verranno utilizzati solo nell'ordine, anche se alcuni potrebbero non essere utilizzati affatto e dovrebbero utilizzare solo la memoria O (1). Immagino che questo significhi che dovrebbero condividere tutti lo stesso iteratore sottostante. Le prestazioni sono importanti.

Se dovessi prendere una pugnalata a una firma di funzione, sarebbe questo:

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

Mi sarebbe piaciuto usare takeWhile, ma perde l'ultimo elemento. Ho studiato span, ma buffer risultati. La mia migliore idea attuale riguarda lo BufferedIterator, ma forse c'è un modo migliore.

Saprete che avete capito bene, perché una cosa del genere non va in crash 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

Ecco la mia soluzione utilizzando BufferedIterator. Non ti permette di saltare gli iteratori correttamente, ma è abbastanza semplice e funzionale. Il primo elemento (i) entra in un gruppo anche se !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() 
    } 
    } 

Aggiornamento: Mantenere un piccolo stato permette di saltare iteratori e impedire alla gente di fare scherzi con i precedenti:

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

avete un problema inerente. Iterable implica che è possibile ottenere più iteratori. Iterator implica che è possibile passare solo una volta. Ciò significa che il tuo Iterable[Iterable[T]] dovrebbe essere in grado di produrre Iterator[Iterable[T]] s. Ma quando questo restituisce un elemento - un Iterable[T] - e richiede più iteratori, il singolo iteratore sottostante non può essere rispettato senza memorizzare nella cache i risultati dell'elenco (troppo grande) o chiamando l'originale iterabile e passando attraverso assolutamente di nuovo tutto (molto inefficiente).

Quindi, mentre tu potresti fare questo, penso che dovresti concepire il tuo problema in un modo diverso.

Se si potesse iniziare con uno Seq, si potrebbero prendere sottoinsiemi come intervalli.

Se sai già come si desidera utilizzare l'iterabile, si potrebbe scrivere un metodo

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

che incrementa attraverso il set di gestori di volta in volta starts incendi fuori un "vero". Se c'è un modo in cui puoi eseguire l'elaborazione in un'unica operazione, qualcosa del genere è la strada da percorrere. (I gestori devono salvare lo stato tramite strutture di dati o variabili mutevoli.)

Se è possibile consentire l'iterazione nell'elenco esterno per rompere l'elenco interno, è possibile avere un Iterable[Iterator[T]] con il vincolo aggiuntivo che una volta iterato a un sotto-iteratore successivo, tutti i precedenti iteratori non sono validi.

---

Ecco una soluzione di quest'ultimo tipo (dalla Iterator[T] a Iterator[Iterator[T]]; si può avvolgere questo per fare gli strati esterni Iterable invece).

Answer 3

Se si stanno osservando i vincoli di memoria, funzionerà quanto segue. Puoi usarlo solo se il tuo oggetto iterabile sottostante supporta le viste.Questa implementazione eseguirà iterazioni su Iterable e quindi genererà IterableViews che potrà quindi essere iterato. Questa implementazione non interessa se il primo elemento test come gruppo di partenza poiché sarà a prescindere.

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

È possibile mantenere a bassa foot-print memoria utilizzando flussi. Utilizzare result.toIterator, se si è di nuovo un iteratore.

Con gli stream, non esiste uno stato mutabile, solo un singolo condizionale ed è quasi conciso come la soluzione di Jay Hacker.

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

Ho eseguito il test:

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

Il secondo stampe di prova troppo per incollare a Stack Overflow.

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