Take the 2-minute tour ×

I'm using Scala to read columns out of our column store Cassandra. Each column contains a number of entries, n, where n can be between 10 and 20. We read a batch of entries, ie 1000 at a time, and have to create columns from the entries; each entry has an ID attached that we can use to group-by.

Currently we use an iterator to go through the entries in a batch and find out if we're onto a new column by comparing the current and previous ID, and we read many batches till we're done. We need to store a partial column at the end of each batch iteration because the rest of the column will be in the next batch. I've put some pseudo code below to demonstrate the basic algorithm we currently employ.

How could do this in a functional way? (If n was constant this would be a simple problem as we could set the batch size appropriately.)

Pseudo code:

val resultBuffer // collects all columns
val columnBuffer // collects entries for current column
var currentId    // id of current column

     val batch = batchIterator.getNext
     val entryIterator = batch.entries.iterator

           val entry = entryIterator.next
            if(entry.id != currentId) {
               currentId = entry.id  
               resultBuffer += columnBuilder(columnBuffer)
               columnBuffer += entry
            } else {
                columnBuffer += entry
share|improve this question

1 Answer 1

Here is a sketch of a more functional implementation that uses sliding to group the entries in the entry-iterator:

val resultBuffer // collects all columns

batchIterator.foreach(batch => {
  val buffer = 
    batch.entries.sliding(2).foldLeft(new ColumnBuffer){(buffer, (curr, next)) =>
      if (curr.id != next.id) {
        resultBuffer += columnBuilder(buffer :+ entry /* Append entry to buffer */)
        new ColumnBuffer
      } else
        buffer += entry /* Return buffer with entry added */

  if (buffer.nonEmpty) resultBuffer += columnBuilder(buffer)

Here, the only object that is "global" and thus has to be mutable is resultBuffer. We could even get rid of that by including it as another accumulator in the inner foldLeft and by replacing the outer foreach with another foldLeft.

If runtime efficiency is critical for your code you should definitely benchmark the various possible implementations in order to find a good trade-off between functional-ness and performance.

EDIT 1: Fixed an error in the sketch, namely, that the last sequence of entries stored in buffer were not added to the resultBuffer. The error already exists in the OP's code.

EDIT 2: (Addressing ChucK's comment)

curr will take the values entries(0) to entries(entries.size() - 2), that is, the last element will not be processed. One way of addressing this is to append a dummy-element to the iterator, e.g.

(batch.entries ++ List(dummy)).sliding(2).foldLeft ...

This isn't nice and, more importantly, it won't work when batch.entries is empty, because sliding(2) then produces the single window List(dummy). Another solution would be to include next in the accumulator of the inner foldLeft and to process it after the foldLeft has terminated. I haven't worked it out but it looks as if that would make the solution even less appealing.

share|improve this answer
In your code does 'entry' correspond to curr or next? The issue I had with sliding(2) is that we get 1 less iterations over batch.entries, so in the first or last iteration we have do so something special to ensure all entries are processed which would be nice to avoid. –  ChucK Nov 23 '12 at 11:53
Most probable case is that the imperative solution is the fastest. Now you say benchmark, whether you can replace the quick code with slower code (although it's nicer). Actually, if importance matters, I'd simply use the existing imperative solution. –  ziggystar Nov 23 '12 at 12:08
@ChucK See my edits. –  Malte Schwerhoff Nov 23 '12 at 13:23

Your Answer


