NICTA/scoobi — Issue #26: Reducer loads too much in memory after a groupByKey
Author: espringe
Created: 2011-12-20 06:37:28+00:00
State: open

I'm not too sure why, but the reducer seems to be loading all the values into memory. The following code works with plain Java/Hadoop, but causes out-of-memory errors with Scoobi.

Expected O(1) memory use, observed O(n)

Program expects two arguments "input file" and "output directory". All the lines from the input file get sent to the same reducer, which will run out of memory if the file is large

import com.nicta.scoobi._
import com.nicta.scoobi.Scoobi._
import com.nicta.scoobi.io.text._

object Main {

def main(a: Array[String]) = withHadoopArgs(a) { args =>

val grouped: DList[(Int, Iterable[String])] = TextInput.fromTextFile(args(0)).map((1, _)).groupByKey

val r: DList[Long] = grouped.parallelDo(new DoFn[(Int, Iterable[String]), Long]() {
override def setup() {}
override def cleanup(emitter: Emitter[Long]) {}
override def process(input: (Int, Iterable[String]), emitter: Emitter[Long]) {
var count = 0L

for (c <- input._2) {
count = count + c.length
}

emitter.emit(count);
}
})

DList.persist(TextOutput.toTextFile(r, args(1)))
}
}