# GroupBy over a clustered iterator in Scala

Here's a solution for a lazy groupBy when the iterator contents are assumed to be clustered (repeated elements next to each other) over the key:

def clusteredGroupBy[B](h: Iterator[B])(f: B => _): Stream[Iterator[B]] = {
if (h.hasNext) {
val firstValue = h.next()
val projection = f(firstValue)
val (head, tail) = h.span(f(_) == projection)

} else Stream.empty
}


Any suggestions on how to improve this? Is there anything more native to Scala that does the same?

-

I'd like to add a completely different answer. The task can be very well solved with a concept called pipes or conduits (also iteratees).

I'll give a solution using my experimental library called scala-conduit (disclaimer: I'm the author).

A Pipe[I,O,R] is a thing that can read input elements of type I from one side (request) and emit output elements of type O on another side (respond). When it finishes, it produces a final result of type R. It decides on its own when it requests and when it responds. This is ideal for the task - we receive inputs and decide when to emit the cluster of equal elements. A pipe also knows when the input finishes. This was problematic with using Streams or other native Scala collections, because folding, mapping, etc. don't notify the function that it hit the end. We didn't know easily when to output the last accumulated buffer. With conduit it is easy, our pipe will get notified when it runs out of input. The solution would look like this:

import conduit._
import conduit.Pipe._
import conduit.Util._

def cluster[A](implicit eq: Equiv[A]): Pipe[A,Seq[A],Unit] = {
// no finalizers used here, import the empty implicit
implicit val fin = Finalizer.empty;

def loop(xs: Seq[A]): Pipe[A,Seq[A],Unit] =
request(
y => xs match {
case Seq(x, _*) if eq.equiv(x, y) => loop(xs :+ y);
case Seq()                        => loop(Seq(y));
case _                            => respond(xs, loop(Seq(y)))
},
// When upstream finishes producing output,
// we just respond the buffer and finish.
_ => if (xs.isEmpty) done else respond(xs)
)

loop(Seq.empty);
}


The elements accumulated so far are kept in loop's argument xs. When we hit an non-equal element, we respond the buffer and start with a new one. When we hit the end, we respond the buffer (if it's non-empty).

Then we can construct a source pipe from an interator and combine the result into a collection:

def runCluster[A](xs: Iterator[A])(implicit eq: Equiv[A]): Seq[Seq[A]] =
runPipe(fromIterator(xs) >-> cluster >-> toCol);

println(runCluster(Stream(1,1,2,2,2,3,4,4,4,5).iterator));


Processing of pipes is single-threaded and fully lazy. A pipe is processed only if its downstream pipe requests an input and only until it produces its output, then it's suspended again. And if a pipe terminates (means stops receiving input), its upstream pipes are notified and any registered finalizers are executed.

For example we could read the source values from a file, process it with cluster and do something with the clustered output. And if the right-most (sink) pipe decided to terminate early, the file would be automatically closed, without reading unneeded input.

(We can't convert a pipe to an Iterator or to a lazy Stream. We must run it and extract its final, definite output. The reason is that if extracted values lazily, we wouldn't be able to detect when no more values are requested and finalizers should be run. However this is not a big problem, we simply build the whole computation from Pipes.)

-

As far as I know, Scala doesn't have anything like you suggest. Your solution is short and fast.

I tried to think of a solution that doesn't use mutable state and uses existing functions, without examining the structure directly. It is based on Streams instead of Iterators, which are imperative in their nature.

The main idea can be expressed using a scanning function that sums together elements that are equal according to a given comparison function:

def scanKey[A](s: Seq[A], x: A)(implicit eq: Equiv[A]): Seq[A] =
s match {
case Seq(y,  _*) if (eq.equiv(x,y)) => s :+ x;
case _                              => Seq(x);
}


Instead of having an explicit key function I used an implicit equality witness.

After applying this function, we have a stream of the same length, whose elements are those sums. What we need to do is get exactly those which are followed by a shorter (or equally sized) sequence. We can do this by zipping the stream and its tail augmented with an empty sequence and filter out just those where the size of a sequence doesn't increase:

def cluster[A](s: Stream[A])(implicit eq: Equiv[A]): Stream[Seq[A]] = {
val seqs = s.scanLeft(Seq.empty[A])(scanKey[A] _)
seqs.zip(seqs.tail append Stream(Seq.empty[A]))
.collect({ case (xs, ys) if (!xs.isEmpty && (ys.size <= xs.size))  => xs })
}

-