OutOfMemory during serialization

[Sebastian Schelter]

Hi,

I get a strange out of memory error from the serialization code when I
try to run the following program:

def compute(trackingGraphFile: String, domainIndexFile: String,
   outputPath: String) = {

implicit val env = ExecutionEnvironment.getExecutionEnvironment

val edges = GraphUtils.readEdges(trackingGraphFile)
val domains = GraphUtils.readVertices(domainIndexFile)

val domainsByCompany = DomainsByCompany.mapping
val companyEdges = edges.filter { edge =>
     domainsByCompany.contains(edge.src.toInt) }
   .map { edge => domainsByCompany(edge.src.toInt) -> edge.target.toInt }
   .distinct

val companyBitMaps = companyEdges.groupBy(0).reduceGroup {
     domainsByCompany: Iterator[(String,Int)] =>

     var company = ""
     val seenAt = new util.BitSet(42889800)

     for ((name, domain) <- domainsByCompany) {
       company = name
       seenAt.set(domain)
     }

     company -> seenAt
   }

   companyBitMaps.print()

   env.execute()

}


The error looks as follows:


2015-02-20 11:22:54 INFO  JobClient:345 - java.lang.OutOfMemoryError:
Java heap space
        at
org.apache.flink.runtime.io.network.serialization.DataOutputSerializer.resize(DataOutputSerializer.java:249)
        at
org.apache.flink.runtime.io.network.serialization.DataOutputSerializer.write(DataOutputSerializer.java:93)
        at
org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream.write(DataOutputViewStream.java:39)
        at com.esotericsoftware.kryo.io.Output.flush(Output.java:163)
        at com.esotericsoftware.kryo.io.Output.require(Output.java:142)
        at com.esotericsoftware.kryo.io.Output.writeBoolean(Output.java:613)
        at com.twitter.chill.java.BitSetSerializer.write(BitSetSerializer.java:42)
        at com.twitter.chill.java.BitSetSerializer.write(BitSetSerializer.java:29)
        at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:599)
        at
org.apache.flink.api.java.typeutils.runtime.KryoSerializer.serialize(KryoSerializer.java:155)
        at
org.apache.flink.api.scala.typeutils.CaseClassSerializer.serialize(CaseClassSerializer.scala:91)
        at
org.apache.flink.api.scala.typeutils.CaseClassSerializer.serialize(CaseClassSerializer.scala:30)
        at
org.apache.flink.runtime.plugable.SerializationDelegate.write(SerializationDelegate.java:51)
        at
org.apache.flink.runtime.io.network.serialization.SpanningRecordSerializer.addRecord(SpanningRecordSerializer.java:76)
        at
org.apache.flink.runtime.io.network.api.RecordWriter.emit(RecordWriter.java:82)
        at
org.apache.flink.runtime.operators.shipping.OutputCollector.collect(OutputCollector.java:88)
        at
org.apache.flink.api.scala.GroupedDataSet$$anon$2.reduce(GroupedDataSet.scala:262)
        at
org.apache.flink.runtime.operators.GroupReduceDriver.run(GroupReduceDriver.java:124)
        at
org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:493)
        at
org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:360)
        at
org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
        at java.lang.Thread.run(Thread.java:745)

I run the job locally, giving 2GB of Ram to the VM. The code will
produce less than 10 groups and the bitsets used internally should not
be larger than a few megabytes.

Any tips on how to fix this?

Best,