spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Gustavo Enrique Salazar Torres <gsala...@ime.usp.br>
Subject Re: LBGFS optimizer performace
Date Wed, 04 Mar 2015 13:03:18 GMT
Yeah, without caching makes it gets really slow. I will try to minimize the
number of columns on my tables, that may save lots of memory and will
eventually work.
I will let you know.

Thanks!
Gustavo

On Tue, Mar 3, 2015 at 8:58 PM, Joseph Bradley <joseph@databricks.com>
wrote:

> I would recommend caching; if you can't persist, iterative algorithms will
> not work well.
>
> I don't think calling count on the dataset is problematic; every iteration
> in LBFGS iterates over the whole dataset and does a lot more computation
> than count().
>
> It would be helpful to see some error occurring within LBFGS.  With the
> given stack trace, I'm not sure what part of LBFGS it's happening in.
>
> On Tue, Mar 3, 2015 at 2:27 PM, Gustavo Enrique Salazar Torres <
> gsalazar@ime.usp.br> wrote:
>
>> Yeah, I can call count before that and it works. Also I was over caching
>> tables but I removed those. Now there is no caching but it gets really slow
>> since it calculates my table RDD many times.
>> Also hacked the LBFGS code to pass the number of examples which I
>> calculated outside in a Spark SQL query but just moved the location of the
>> problem.
>>
>> The query I'm running looks like this:
>>
>> s"SELECT $mappedFields, tableB.id as b_id FROM tableA LEFT JOIN tableB
>>  ON tableA.id=tableB.table_a_id WHERE tableA.status='' OR tableB.status='' "
>>
>> mappedFields contains a list of fields which I'm interested in. The
>> result of that query goes through (including sampling) some transformations
>> before being input to LBFGS.
>>
>> My dataset has 180GB just for feature selection, I'm planning to use
>> 450GB to train the final model and I'm using 16 c3.2xlarge EC2 instances,
>> that means I have 240GB of RAM available.
>>
>> Any suggestion? I'm starting to check the algorithm because I don't
>> understand why it needs to count the dataset.
>>
>> Thanks
>>
>> Gustavo
>>
>> On Tue, Mar 3, 2015 at 6:08 PM, Joseph Bradley <joseph@databricks.com>
>> wrote:
>>
>>> Is that error actually occurring in LBFGS?  It looks like it might be
>>> happening before the data even gets to LBFGS.  (Perhaps the outer join
>>> you're trying to do is making the dataset size explode a bit.)  Are you
>>> able to call count() (or any RDD action) on the data before you pass it to
>>> LBFGS?
>>>
>>> On Tue, Mar 3, 2015 at 8:55 AM, Gustavo Enrique Salazar Torres <
>>> gsalazar@ime.usp.br> wrote:
>>>
>>>> Just did with the same error.
>>>> I think the problem is the "data.count()" call in LBFGS because for
>>>> huge datasets that's naive to do.
>>>> I was thinking to write my version of LBFGS but instead of doing
>>>> data.count() I will pass that parameter which I will calculate from a Spark
>>>> SQL query.
>>>>
>>>> I will let you know.
>>>>
>>>> Thanks
>>>>
>>>>
>>>> On Tue, Mar 3, 2015 at 3:25 AM, Akhil Das <akhil@sigmoidanalytics.com>
>>>> wrote:
>>>>
>>>>> Can you try increasing your driver memory, reducing the executors and
>>>>> increasing the executor memory?
>>>>>
>>>>> Thanks
>>>>> Best Regards
>>>>>
>>>>> On Tue, Mar 3, 2015 at 10:09 AM, Gustavo Enrique Salazar Torres <
>>>>> gsalazar@ime.usp.br> wrote:
>>>>>
>>>>>> Hi there:
>>>>>>
>>>>>> I'm using LBFGS optimizer to train a logistic regression model. The
>>>>>> code I implemented follows the pattern showed in
>>>>>> https://spark.apache.org/docs/1.2.0/mllib-linear-methods.html but
>>>>>> training data is obtained from a Spark SQL RDD.
>>>>>> The problem I'm having is that LBFGS tries to count the elements
in
>>>>>> my RDD and that results in a OOM exception since my dataset is huge.
>>>>>> I'm running on a AWS EMR cluster with 16 c3.2xlarge instances on
>>>>>> Hadoop YARN. My dataset is about 150 GB but I sample (I take only
1% of the
>>>>>> data) it in order to scale logistic regression.
>>>>>> The exception I'm getting is this:
>>>>>>
>>>>>> 15/03/03 04:21:44 WARN scheduler.TaskSetManager: Lost task 108.0
in
>>>>>> stage 2.0 (TID 7600, ip-10-155-20-71.ec2.internal):
>>>>>> java.lang.OutOfMemoryError: Java heap space
>>>>>>         at java.util.Arrays.copyOfRange(Arrays.java:2694)
>>>>>>         at java.lang.String.<init>(String.java:203)
>>>>>>         at
>>>>>> com.esotericsoftware.kryo.io.Input.readString(Input.java:448)
>>>>>>         at
>>>>>> com.esotericsoftware.kryo.serializers.DefaultSerializers$StringSerializer.read(DefaultSerializers.java:157)
>>>>>>         at
>>>>>> com.esotericsoftware.kryo.serializers.DefaultSerializers$StringSerializer.read(DefaultSerializers.java:146)
>>>>>>         at
>>>>>> com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:732)
>>>>>>         at
>>>>>> com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ObjectArraySerializer.read(DefaultArraySerializers.java:338)
>>>>>>         at
>>>>>> com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ObjectArraySerializer.read(DefaultArraySerializers.java:293)
>>>>>>         at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:651)
>>>>>>         at
>>>>>> com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(FieldSerializer.java:605)
>>>>>>         at
>>>>>> com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:221)
>>>>>>         at
>>>>>> com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:732)
>>>>>>         at
>>>>>> com.twitter.chill.Tuple2Serializer.read(TupleSerializers.scala:42)
>>>>>>         at
>>>>>> com.twitter.chill.Tuple2Serializer.read(TupleSerializers.scala:33)
>>>>>>         at
>>>>>> com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:732)
>>>>>>         at
>>>>>> org.apache.spark.serializer.KryoDeserializationStream.readObject(KryoSerializer.scala:144)
>>>>>>         at
>>>>>> org.apache.spark.serializer.DeserializationStream$$anon$1.getNext(Serializer.scala:133)
>>>>>>         at
>>>>>> org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:71)
>>>>>>         at
>>>>>> org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:32)
>>>>>>         at
>>>>>> scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
>>>>>>         at
>>>>>> org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:32)
>>>>>>         at
>>>>>> org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:39)
>>>>>>         at
>>>>>> scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
>>>>>>         at
>>>>>> scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
>>>>>>         at org.apache.spark.sql.execution.joins.HashOuterJoin.org
>>>>>> $apache$spark$sql$execution$joins$HashOuterJoin$$buildHashTable(HashOuterJoin.scala:179)
>>>>>>         at
>>>>>> org.apache.spark.sql.execution.joins.HashOuterJoin$$anonfun$execute$1.apply(HashOuterJoin.scala:199)
>>>>>>         at
>>>>>> org.apache.spark.sql.execution.joins.HashOuterJoin$$anonfun$execute$1.apply(HashOuterJoin.scala:196)
>>>>>>         at
>>>>>> org.apache.spark.rdd.ZippedPartitionsRDD2.compute(ZippedPartitionsRDD.scala:88)
>>>>>>         at
>>>>>> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)
>>>>>>         at org.apache.spark.rdd.RDD.iterator(RDD.scala:247)
>>>>>>         at
>>>>>> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)
>>>>>>         at
>>>>>> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)
>>>>>>
>>>>>> I'm using this parameters at runtime:
>>>>>> --num-executors 128 --executor-memory 1G --driver-memory 4G
>>>>>> --conf spark.serializer=org.apache.spark.serializer.KryoSerializer
>>>>>> --conf spark.storage.memoryFraction=0.2
>>>>>>
>>>>>> I also persist my dataset using MEMORY_AND_DISK_SER but get the same
>>>>>> error.
>>>>>> I will appreciate any help on this problem. I have been trying to
>>>>>> solve it for days and I'm running out of time and hair.
>>>>>>
>>>>>> Thanks
>>>>>> Gustavo
>>>>>>
>>>>>
>>>>>
>>>>
>>>
>>
>

Mime
View raw message