Lets create an issue in Flink to somehow fix the issue. Lets a) see if the new serializer registration in 0.9 allows users to replace the serializers if they had been already set by chill. and b) fix the issue in twitter/chill. I think we can ask them to release a new version with the fix (they seem to release quite often). Also, I made good experiences with contributing to twitter/chill.
On Fri, Feb 20, 2015 at 2:02 PM, Ufuk Celebi <u...@apache.org> wrote: > I've just looked into the BitSetSerializer of Chill. And it seems to be > true that each bit is encoded as a boolean (for all bit positions <= > "logical" length). > > Regarding the DataOutputSerializer: would help to catch OoM exceptions > during resize operations and rethrow it with a more detailed message (how > large the buffer is currently, new size after resize). > > On 20 Feb 2015, at 13:22, Stephan Ewen <se...@apache.org> wrote: > > > What happens (in the original stack trace) is the following: The > serializer starts producing the byte stream data and we buffer it, to > determine the length, before sending it over the network. While buffering > that data, the memory runs out. > > > > It may be that you are simply short of memory, it may also be that the > serializer (here the Kryo Chill BitsetSerializer) is simply extremely > inefficient in terms of space. It seems that it tries to write a boolean > (coded as one byte) per bit. That is blowing up your bitset quite a bit. > > > > A solution may also be to register a better bitset serializer. Chill's > default one seems to be sort of inefficient... > > > > > > > > > > > > On Fri, Feb 20, 2015 at 1:03 PM, Sebastian <ssc.o...@googlemail.com> > wrote: > > I don't have a build unfortunately, I'm using the maven dependency. I'll > try to find a workaround. Thx for your help. > > > > -s > > > > On 20.02.2015 12:44, Robert Metzger wrote: > > Hey Sebastian, > > > > I've fixed the issue in this branch: > > https://github.com/rmetzger/flink/tree/flink1589: > > > > Configuration c =newConfiguration(); > > c.setFloat(ConfigConstants.TASK_MANAGER_MEMORY_FRACTION_KEY,0.5f); > > finalExecutionEnvironment env = > ExecutionEnvironment.createLocalEnvironment(c); > > > > > > I'll also backport the fix to the release-0.8 branch to make it > > available in the 0.8.2 release. > > > > Maybe you can easily cherry-pick the commit to your 0.8.1 Flink build. > > > > > > Best, > > Robert > > > > On Fri, Feb 20, 2015 at 12:11 PM, Robert Metzger <rmetz...@apache.org > > <mailto:rmetz...@apache.org>> wrote: > > > > Hi Sebastian, > > > > Looks like you've found a limitation of Flink. > > I've already filed two JIRAs to resolve the issue > > (https://issues.apache.org/jira/browse/FLINK-1588, > > https://issues.apache.org/jira/browse/FLINK-1589). > > > > I don't know your setup, when you use Flink just as a dependency > > without a version being checked out, there is probably no way right > > now to use change the configuration settings. > > Then, you have to start yourself a local cluster > > (./bin/start-local.sh (+ your settings in conf/flink-conf.yaml)). > > You can then either submit your job with ./bin/flink or using the > > RemoteExecutionEnvironment > (ExecutionEnvironment.createRemoteEnvironment()). > > > > If you have the Flink source checked out, you can also hard-code the > > configuration values into org.apache.flink.client.LocalExecutor. > > > > > > By the way, Flink 0.8.1 is now available on maven central (I suspect > > you had to build it yourself yesterday evening). > > But given these issues here, it doesn't matter for you anymore ;) > > > > > > Best, > > Robert > > > > > > > > On Fri, Feb 20, 2015 at 11:48 AM, Sebastian <ssc.o...@googlemail.com > > <mailto:ssc.o...@googlemail.com>> wrote: > > > > I'm running flink from my IDE, how do change this setting in > > that context? > > > > > > On 20.02.2015 11:41, Fabian Hueske wrote: > > > > Have you tried to increase the heap size by shrinking the > > TM-managed memory? > > > > Reduce the fraction (taskmanager.memory.fraction) or fix the > > amount of TM memory (taskmanager.memory.size) in the > > flink-config.yaml [1]. > > > > Cheers, Fabian > > > > [1] http://flink.apache.org/docs/__0.8/config.html > > <http://flink.apache.org/docs/0.8/config.html> > > > > > > On 20 Feb 2015, at 11:30, Sebastian > > <ssc.o...@googlemail.com > > <mailto:ssc.o...@googlemail.com>> wrote: > > > > 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 > > <http://org.apache.flink.runtime.io > >.__network.serialization.__DataOutputSerializer.resize(__DataOutputSerializer.java:249) > > at org.apache.flink.runtime.io > > <http://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 > > <http://com.esotericsoftware.kryo.io > >.__Output.flush(Output.java:163) > > at com.esotericsoftware.kryo.io > > <http://com.esotericsoftware.kryo.io > >.__Output.require(Output.java:__142) > > at com.esotericsoftware.kryo.io > > <http://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 > > <http://org.apache.flink.runtime.io > >.__network.serialization.__SpanningRecordSerializer.__addRecord(__SpanningRecordSerializer.java:__76) > > at org.apache.flink.runtime.io > > <http://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, > > Sebastian > > > > PS: Still waiting for a reduceGroup that gives me the > key ;) > > > > > > > > > > > > > > > >
