Requested array size exceeds VM limit when loading a large RDF file using Faunus

[Yuriy]

I am trying to load a large, slightly more than 100GB in size, RDF file in Faunus (based on Titan code base 0.4.4)

My configuration is:

# input graph parameters

faunus.graph.input.format=com.thinkaurelius.faunus.formats.edgelist.rdf.RDFInputFormat

faunus.graph.input.rdf.format=n-triples

faunus.graph.input.rdf.as-properties=http://www.w3.org/1999/02/22-rdf-syntax-ns#type,http://www.w3.org/2000/01/rdf-schema#label

faunus.graph.input.rdf.use-localname=true

faunus.graph.input.rdf.literal-as-property=true

faunus.input.location=LargeGraph.nt

# output data parameters

faunus.graph.output.format=com.thinkaurelius.faunus.formats.titan.hbase.TitanHBaseOutputFormat

faunus.graph.output.titan.storage.backend=hbase

faunus.graph.output.titan.storage.hostname=Zookeeper1, Zookeeper2

faunus.graph.output.titan.storage.port=2181

faunus.graph.output.titan.storage.tablename=titan

faunus.graph.output.titan.storage.batch-loading=true

faunus.graph.output.titan.ids.block-size=1000000

# long

faunus.graph.output.titan.storage.idauthority-wait-time=1000

# long

faunus.graph.output.titan.ids.renew-timeout=21474836400000

# int

faunus.graph.output.titan.ids.idauthority-retries=2147483640

faunus.graph.output.titan.infer-schema=true

faunus.graph.output.titan.ids.num-partitions=7

faunus.graph.output.titan.ids.partition=true

faunus.output.location=output

faunus.output.location.overwrite=true

mapreduce.linerecordreader.maxlength=5242880

mapreduce.input.fileinputformat.split.maxsize=5242880

mapreduce.map.memory.mb=4098

mapreduce.reduce.memory.mb=81920

mapreduce.map.java.opts=-Xmx4G

mapreduce.reduce.java.opts=-Xmx80G

mapreduce.job.reuse.jvm.num.tasks=-1

mapreduce.job.maxtaskfailures.per.tracker=256

mapreduce.map.maxattempts=128

mapreduce.reduce.maxattempts=128

mapreduce.job.reduces=70

mapreduce.job.maps=120

mapreduce.task.timeout=54000000

During the reduce phase of the first M/R job (EdgeListInputMapReduce.Map > EdgeListInputMapReduce.Reduce > IdentityMap.Map), I get this exception and increasing the reducer memory size from 25 GB to even 80GB does not help:

17:02:53 INFO mapreduce.Job: Task Id : attempt_1407466272407_0027_r_000006_0, Status : FAILED

Error: java.lang.RuntimeException: java.lang.OutOfMemoryError: Requested array size exceeds VM limit

        at org.apache.hadoop.mapreduce.lib.chain.Chain.joinAllThreads(Chain.java:526)

        at org.apache.hadoop.mapreduce.lib.chain.ChainReducer.run(ChainReducer.java:218)

        at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:627)

        at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:389)

        at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:167)

        at java.security.AccessController.doPrivileged(Native Method)

        at javax.security.auth.Subject.doAs(Subject.java:422)

        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1557)

        at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:162)

Caused by: java.lang.OutOfMemoryError: Requested array size exceeds VM limit

        at java.util.Arrays.copyOf(Arrays.java:3230)

        at java.io.ByteArrayOutputStream.grow(ByteArrayOutputStream.java:113)

        at java.io.ByteArrayOutputStream.ensureCapacity(ByteArrayOutputStream.java:93)

        at java.io.ByteArrayOutputStream.write(ByteArrayOutputStream.java:140)

        at java.io.DataOutputStream.write(DataOutputStream.java:107)

        at java.io.FilterOutputStream.write(FilterOutputStream.java:97)

        at com.thinkaurelius.faunus.FaunusElement$ElementProperties.write(FaunusElement.java:248)

        at com.thinkaurelius.faunus.FaunusElement.write(FaunusElement.java:217)

        at com.thinkaurelius.faunus.FaunusEdge.writeCompressed(FaunusEdge.java:108)

        at com.thinkaurelius.faunus.FaunusVertex$EdgeMap.write(FaunusVertex.java:396)

        at com.thinkaurelius.faunus.FaunusVertex.write(FaunusVertex.java:281)

        at org.apache.hadoop.io.serializer.WritableSerialization$WritableSerializer.serialize(WritableSerialization.java:98)

        at org.apache.hadoop.io.serializer.WritableSerialization$WritableSerializer.serialize(WritableSerialization.java:82)

        at org.apache.hadoop.util.ReflectionUtils.copy(ReflectionUtils.java:292)

        at org.apache.hadoop.mapreduce.lib.chain.Chain$ChainRecordWriter.writeToQueue(Chain.java:264)

        at org.apache.hadoop.mapreduce.lib.chain.Chain$ChainRecordWriter.write(Chain.java:252)

        at org.apache.hadoop.mapreduce.lib.chain.ChainReduceContextImpl.write(ChainReduceContextImpl.java:103)

        at org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer$Context.write(WrappedReducer.java:105)

        at com.thinkaurelius.faunus.formats.edgelist.EdgeListInputMapReduce$Reduce.reduce(EdgeListInputMapReduce.java:127)

        at com.thinkaurelius.faunus.formats.edgelist.EdgeListInputMapReduce$Reduce.reduce(EdgeListInputMapReduce.java:112)

        at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:171)

        at org.apache.hadoop.mapreduce.lib.chain.Chain$ReduceRunner.run(Chain.java:349)

Any ideas on what should I try to get this data load to work?

Would splitting the 100GB file into the smaller files and specifying in the configuration the folder where these files are make any difference? 

Thanks,

Yuriy

[Dan LaRocque]

Hi Yuriy,

The config seems to include both Faunus 0.4.x and Hadoop 2.x configuration keys.  Faunus 0.4.x only shipped with official support for Hadoop 1.  What Hadoop version are you using?  Did you compile faunus's hadoop2 feature branch or are you using the stock Maven artifacts/zipfile?

thanks,
Dan

[Yuriy]

Hi Dan,

Yes, I compiled Faunus Hadoop2 (based on Hadoop2 branch and Lab41 clone). The other changes I've made are just to get the code to run in Windows/HDP2.1 environment (I did not touch any M/R code).

I believe that I've hit a super node problem - from my RDF file I am trying to load a vertex with 40 millions of edges and the array-backed DataOutput simply cannot accoumode the data size :(

I consider trying incremental data loading script with Faunus but given that it relies on the calls to groovy it might considerably slow down the loading process. 

Also, the problem with that approach is that I'll have to find some approach to divide the data in such a way that the super node's would not end up in the single batch, but it might just be easier to improve the M/R data loading code. Sigh.

Any ideas for a work around from your side? 

Thanks,

Yuriy

[Dan LaRocque]

Hi Yuriy,

On 08/25/2014 06:25 PM, Yuriy wrote:
> Hi Dan,
>
> Yes, I compiled Faunus Hadoop2 (based on Hadoop2 branch and Lab41
> clone). The other changes I've made are just to get the code to run in
> Windows/HDP2.1 environment (I did not touch any M/R code).
>
> I believe that I've hit a super node problem - from my RDF file I am
> trying to load a vertex with 40 millions of edges and the
> array-backed DataOutput simply cannot accoumode the data size :(

Hi Yuriy,

I think the array-backed DataOutput here is part of ChainReducer. The
job seems to get through mapping and through the reducer's shuffle and
sort phases, since it's failing inside the reduce method
implementation. 40 million edges consuming 80 GB of RAM implies a worst
case of 2 GB / edge, which is insane. It might be a little less due to
non-supernode keys in the same Iterable passed to the dying reducer, but
the order of magnitude is enough. Does the supernode have a large
number of properties besides its edges? If not, then I think there's
either an unresolved config issue or a bug in here.

>
> I consider trying incremental data loading script with Faunus but
> given that it relies on the calls to groovy it might considerably slow
> down the loading process.
> Also, the problem with that approach is that I'll have to find some
> approach to divide the data in such a way that the super node's would
> not end up in the single batch, but it might just be easier to improve
> the M/R data loading code. Sigh.
>
> Any ideas for a work around from your side?

Sill question: did you configure YARN's resource settings to limit MR to
only one reducer task at a time?

thanks,
Dan

[Yuriy]

Hi Dan, 

Thanks for your help.

>40 million edges consuming 80 GB of RAM implies a worst 
>case of 2 GB / edge, which is insane.

Correct me if I am wrong, but I don't think that that is the case. I believe that we are just limited by the size of Java array (~2GB). 

40 million edges consume more than 2 GB (in total) and that's the maximum size of Java array. Thus, array-backed DataOutput is unable to allocate an array that is larger than those 2 GB. 

IMHO, when FaunusVertex tries to write its data, it writes to DataOutput which is limited to 2 GB because of its array back-end and throws as it tries saving more than 2GB of data.

If I look at the process, it's private working set never gets anywhere close to 80GB (I'll keep looking, but I did not see any java process consuming so much memory in this phase of M/R; in the last phase).

(Btw, the number of edges 40 mil was obtained by modifying Faunus code and adding additional logging statements.)

There is only two properties on the super-node with like 30 char each. Edges should not have many properties either - just a name which is no more than 100 chars (as it is a uri)

>Sill question: did you configure YARN's resource settings to limit MR to 

>only one reducer task at a time? 

No, but since I've allowed Hadoop to allocate insane amount of RAM to  the reducer by specifying

mapreduce.reduce.memory.mb=81920

mapreduce.reduce.java.opts=-Xmx80G

per my configuration above, Yarn does not start more than a single reducer at a time on a node.

If I remove these config values, I see multiple reducers.

I don't have any other specific configurations settings for Yarn. 

Thanks,

Yuriy

[Dan LaRocque]

On 08/25/2014 09:19 PM, Yuriy wrote:

Hi Dan, 

Thanks for your help.

>40 million edges consuming 80 GB of RAM implies a worst 
>case of 2 GB / edge, which is insane.

Correct me if I am wrong, but I don't think that that is the case. I believe that we are just limited by the size of Java array (~2GB). 

40 million edges consume more than 2 GB (in total) and that's the maximum size of Java array. Thus, array-backed DataOutput is unable to allocate an array that is larger than those 2 GB. 

IMHO, when FaunusVertex tries to write its data, it writes to DataOutput which is limited to 2 GB because of its array back-end and throws as it tries saving more than 2GB of data.

Hi Yuriy,

Yeah, this makes sense, and I was off by 10^6 besides :).  Should have written 2 kB per edge.  But fitting a 40M edge supernode in a 2GB bytearrayoutputstream is more like 50 bytes / edge, right?  And that's best case.  The array size limit is usually lower than Integer.MAX_VALUE (and appears to be in your trace), and there are likely to be other reducer keys besides the supernode's in the dying task.

To read supernodes in RDF, I think we would have to either introduce an option to disable ChainReducer so that reduce writes go to disk before being fed into the post-reduce mapper, or complicate EdgeList/RDF input so that it can partition supernodes into fixed-size reducer keys.  I haven't thought the latter through carefully, but I suspect that it would be much more disruptive to Faunus internals than making chain reducer configurable.  Titan vertex partitioning wouldn't help in this case since the job is just reading from an RDF file in HDFS at this point and does not appear to have even started talking to Titan yet.

If I look at the process, it's private working set never gets anywhere close to 80GB (I'll keep looking, but I did not see any java process consuming so much memory in this phase of M/R; in the last phase).

(Btw, the number of edges 40 mil was obtained by modifying Faunus code and adding additional logging statements.)

There is only two properties on the super-node with like 30 char each. Edges should not have many properties either - just a name which is no more than 100 chars (as it is a uri)

>Sill question: did you configure YARN's resource settings to limit MR to 

>only one reducer task at a time? 

No, but since I've allowed Hadoop to allocate insane amount of RAM to  the reducer by specifying

mapreduce.reduce.memory.mb=81920

mapreduce.reduce.java.opts=-Xmx80G

per my configuration above, Yarn does not start more than a single reducer at a time on a node.

If I remove these config values, I see multiple reducers.

I don't have any other specific configurations settings for Yarn. 

Sounds good as long as YARN only starts one reducer at a time.

thanks,
Dan

[Edi Bice]

I've been running into this repeatedly trying to load LinkedBrainz/artist*.nt using Faunus+Titan 0.4.4. Has this been addressed in Titan+Hadoop 0.5?