You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by "igor.berman" <ig...@gmail.com> on 2015/05/31 17:33:13 UTC
union and reduceByKey wrong shuffle?
I've encountered very strange problem, after doing union of 2 rdds the
reduceByKey works wrong(unless I'm missing something very basic) and brings
to the function that reduces 2 objects with different key! I've rewrited
java class to scala to test it in spark-shell and I see same problem
I have SingleIdDailyAggData object with following fileds(it's avro object
really)
long dyid;
com.dy.agg.gen.AggData data;
long lastSeen;
I have AvroReader.readKeys that returns me this objects as
JavaRDD<SingleIdDailyAggData>(reading from s3 files):
val ctx = new JavaSparkContext(sc)
val absent = Optional.absent[Integer]();
val dayInputs = new ArrayList[String]();
dayInputs.add("s3n://xxx/blocks/2015-05-26/8765260/day/")
//java rdd of SingleIdDailyAggData
val day = AvroReader.readKeys[SingleIdDailyAggData](ctx,
dayInputs.asInstanceOf[List[String]], absent);
val prevDayInputs = new ArrayList[String]();
prevDayInputs.add("s3n://xxx/blocks/2015-05-25/8765260/day/")
//java rdd of SingleIdDailyAggData
val prevDay = AvroReader.readKeys[SingleIdDailyAggData](ctx,
prevDayInputs.asInstanceOf[List[String]], absent);
//long->SingleIdDailyAggData
val dayByKey = day.rdd.keyBy((x:SingleIdDailyAggData)=> x.getDyid());
//long->SingleIdDailyAggData
val prevDayByKey = prevDay.rdd.keyBy((x:SingleIdDailyAggData)=>
x.getDyid());
//union
val un = dayByKey.union(prevDayByKey)
val r = un.reduceByKey((v1,v2) => if(v1.getDyid().longValue ==
v2.getDyid().longValue()) new SingleIdDailyAggData(v1.getDyid(), null, null)
else throw new IllegalArgumentException(v1.getDyid().longValue + " != " +
v2.getDyid().longValue()) )
r.count()
Task 7 in stage 3.0 failed 1 times, most recent failure: Lost task 7.0 in
stage 3.0 (TID 133, localhost): java.lang.IllegalArgumentException:
522417601156300917 != 2481483430461576405
so it's not custom object as a key, it's java's Long(i.e. not hashCode vs
equals problem)
Any ideas would be highly appreciated!
Igor
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/union-and-reduceByKey-wrong-shuffle-tp23092.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: union and reduceByKey wrong shuffle?
Posted by Igor Berman <ig...@gmail.com>.
this jira seems to be connected to our issue
https://issues.apache.org/jira/browse/SPARK-1018
On 2 June 2015 at 19:54, Josh Rosen <ro...@gmail.com> wrote:
> Ah, interesting. While working on my new Tungsten shuffle manager, I came
> up with some nice testing interfaces for allowing me to manually trigger
> spills in order to deterministically test those code paths without
> requiring large amounts of data to be shuffled. Maybe I could make similar
> test interface changes to the existing shuffle code, which might make it
> easier to reproduce this in an isolated environment.
>
> On Mon, Jun 1, 2015 at 11:41 PM, Igor Berman <ig...@gmail.com>
> wrote:
>
>> Hi,
>> small mock data doesn't reproduce the problem. IMHO problem is reproduced
>> when we make shuffle big enough to split data into disk.
>> We will work on it to understand and reproduce the problem(not first
>> priority though...)
>>
>>
>> On 1 June 2015 at 23:02, Josh Rosen <ro...@gmail.com> wrote:
>>
>>> How much work is to produce a small standalone reproduction? Can you
>>> create an Avro file with some mock data, maybe 10 or so records, then
>>> reproduce this locally?
>>>
>>> On Mon, Jun 1, 2015 at 12:31 PM, Igor Berman <ig...@gmail.com>
>>> wrote:
>>>
>>>> switching to use simple pojos instead of using avro for spark
>>>> serialization solved the problem(I mean reading avro from s3 and than
>>>> mapping each avro object to it's pojo serializable counterpart with same
>>>> fields, pojo is registered withing kryo)
>>>> Any thought where to look for a problem/misconfiguration?
>>>>
>>>> On 31 May 2015 at 22:48, Igor Berman <ig...@gmail.com> wrote:
>>>>
>>>>> Hi
>>>>> We are using spark 1.3.1
>>>>> Avro-chill (tomorrow will check if its important) we register avro
>>>>> classes from java
>>>>> Avro 1.7.6
>>>>> On May 31, 2015 22:37, "Josh Rosen" <ro...@gmail.com> wrote:
>>>>>
>>>>>> Which Spark version are you using? I'd like to understand whether
>>>>>> this change could be caused by recent Kryo serializer re-use changes in
>>>>>> master / Spark 1.4.
>>>>>>
>>>>>> On Sun, May 31, 2015 at 11:31 AM, igor.berman <ig...@gmail.com>
>>>>>> wrote:
>>>>>>
>>>>>>> after investigation the problem is somehow connected to avro
>>>>>>> serialization
>>>>>>> with kryo + chill-avro(mapping avro object to simple scala case
>>>>>>> class and
>>>>>>> running reduce on these case class objects solves the problem)
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> --
>>>>>>> View this message in context:
>>>>>>> http://apache-spark-user-list.1001560.n3.nabble.com/union-and-reduceByKey-wrong-shuffle-tp23092p23093.html
>>>>>>> Sent from the Apache Spark User List mailing list archive at
>>>>>>> Nabble.com.
>>>>>>>
>>>>>>> ---------------------------------------------------------------------
>>>>>>> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
>>>>>>> For additional commands, e-mail: user-help@spark.apache.org
>>>>>>>
>>>>>>>
>>>>>>
>>>>
>>>
>>
>
Re: union and reduceByKey wrong shuffle?
Posted by Josh Rosen <ro...@gmail.com>.
Ah, interesting. While working on my new Tungsten shuffle manager, I came
up with some nice testing interfaces for allowing me to manually trigger
spills in order to deterministically test those code paths without
requiring large amounts of data to be shuffled. Maybe I could make similar
test interface changes to the existing shuffle code, which might make it
easier to reproduce this in an isolated environment.
On Mon, Jun 1, 2015 at 11:41 PM, Igor Berman <ig...@gmail.com> wrote:
> Hi,
> small mock data doesn't reproduce the problem. IMHO problem is reproduced
> when we make shuffle big enough to split data into disk.
> We will work on it to understand and reproduce the problem(not first
> priority though...)
>
>
> On 1 June 2015 at 23:02, Josh Rosen <ro...@gmail.com> wrote:
>
>> How much work is to produce a small standalone reproduction? Can you
>> create an Avro file with some mock data, maybe 10 or so records, then
>> reproduce this locally?
>>
>> On Mon, Jun 1, 2015 at 12:31 PM, Igor Berman <ig...@gmail.com>
>> wrote:
>>
>>> switching to use simple pojos instead of using avro for spark
>>> serialization solved the problem(I mean reading avro from s3 and than
>>> mapping each avro object to it's pojo serializable counterpart with same
>>> fields, pojo is registered withing kryo)
>>> Any thought where to look for a problem/misconfiguration?
>>>
>>> On 31 May 2015 at 22:48, Igor Berman <ig...@gmail.com> wrote:
>>>
>>>> Hi
>>>> We are using spark 1.3.1
>>>> Avro-chill (tomorrow will check if its important) we register avro
>>>> classes from java
>>>> Avro 1.7.6
>>>> On May 31, 2015 22:37, "Josh Rosen" <ro...@gmail.com> wrote:
>>>>
>>>>> Which Spark version are you using? I'd like to understand whether
>>>>> this change could be caused by recent Kryo serializer re-use changes in
>>>>> master / Spark 1.4.
>>>>>
>>>>> On Sun, May 31, 2015 at 11:31 AM, igor.berman <ig...@gmail.com>
>>>>> wrote:
>>>>>
>>>>>> after investigation the problem is somehow connected to avro
>>>>>> serialization
>>>>>> with kryo + chill-avro(mapping avro object to simple scala case class
>>>>>> and
>>>>>> running reduce on these case class objects solves the problem)
>>>>>>
>>>>>>
>>>>>>
>>>>>>
>>>>>> --
>>>>>> View this message in context:
>>>>>> http://apache-spark-user-list.1001560.n3.nabble.com/union-and-reduceByKey-wrong-shuffle-tp23092p23093.html
>>>>>> Sent from the Apache Spark User List mailing list archive at
>>>>>> Nabble.com.
>>>>>>
>>>>>> ---------------------------------------------------------------------
>>>>>> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
>>>>>> For additional commands, e-mail: user-help@spark.apache.org
>>>>>>
>>>>>>
>>>>>
>>>
>>
>
Re: union and reduceByKey wrong shuffle?
Posted by Igor Berman <ig...@gmail.com>.
Hi,
small mock data doesn't reproduce the problem. IMHO problem is reproduced
when we make shuffle big enough to split data into disk.
We will work on it to understand and reproduce the problem(not first
priority though...)
On 1 June 2015 at 23:02, Josh Rosen <ro...@gmail.com> wrote:
> How much work is to produce a small standalone reproduction? Can you
> create an Avro file with some mock data, maybe 10 or so records, then
> reproduce this locally?
>
> On Mon, Jun 1, 2015 at 12:31 PM, Igor Berman <ig...@gmail.com>
> wrote:
>
>> switching to use simple pojos instead of using avro for spark
>> serialization solved the problem(I mean reading avro from s3 and than
>> mapping each avro object to it's pojo serializable counterpart with same
>> fields, pojo is registered withing kryo)
>> Any thought where to look for a problem/misconfiguration?
>>
>> On 31 May 2015 at 22:48, Igor Berman <ig...@gmail.com> wrote:
>>
>>> Hi
>>> We are using spark 1.3.1
>>> Avro-chill (tomorrow will check if its important) we register avro
>>> classes from java
>>> Avro 1.7.6
>>> On May 31, 2015 22:37, "Josh Rosen" <ro...@gmail.com> wrote:
>>>
>>>> Which Spark version are you using? I'd like to understand whether this
>>>> change could be caused by recent Kryo serializer re-use changes in master /
>>>> Spark 1.4.
>>>>
>>>> On Sun, May 31, 2015 at 11:31 AM, igor.berman <ig...@gmail.com>
>>>> wrote:
>>>>
>>>>> after investigation the problem is somehow connected to avro
>>>>> serialization
>>>>> with kryo + chill-avro(mapping avro object to simple scala case class
>>>>> and
>>>>> running reduce on these case class objects solves the problem)
>>>>>
>>>>>
>>>>>
>>>>>
>>>>> --
>>>>> View this message in context:
>>>>> http://apache-spark-user-list.1001560.n3.nabble.com/union-and-reduceByKey-wrong-shuffle-tp23092p23093.html
>>>>> Sent from the Apache Spark User List mailing list archive at
>>>>> Nabble.com.
>>>>>
>>>>> ---------------------------------------------------------------------
>>>>> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
>>>>> For additional commands, e-mail: user-help@spark.apache.org
>>>>>
>>>>>
>>>>
>>
>
Re: union and reduceByKey wrong shuffle?
Posted by Josh Rosen <ro...@gmail.com>.
How much work is to produce a small standalone reproduction? Can you
create an Avro file with some mock data, maybe 10 or so records, then
reproduce this locally?
On Mon, Jun 1, 2015 at 12:31 PM, Igor Berman <ig...@gmail.com> wrote:
> switching to use simple pojos instead of using avro for spark
> serialization solved the problem(I mean reading avro from s3 and than
> mapping each avro object to it's pojo serializable counterpart with same
> fields, pojo is registered withing kryo)
> Any thought where to look for a problem/misconfiguration?
>
> On 31 May 2015 at 22:48, Igor Berman <ig...@gmail.com> wrote:
>
>> Hi
>> We are using spark 1.3.1
>> Avro-chill (tomorrow will check if its important) we register avro
>> classes from java
>> Avro 1.7.6
>> On May 31, 2015 22:37, "Josh Rosen" <ro...@gmail.com> wrote:
>>
>>> Which Spark version are you using? I'd like to understand whether this
>>> change could be caused by recent Kryo serializer re-use changes in master /
>>> Spark 1.4.
>>>
>>> On Sun, May 31, 2015 at 11:31 AM, igor.berman <ig...@gmail.com>
>>> wrote:
>>>
>>>> after investigation the problem is somehow connected to avro
>>>> serialization
>>>> with kryo + chill-avro(mapping avro object to simple scala case class
>>>> and
>>>> running reduce on these case class objects solves the problem)
>>>>
>>>>
>>>>
>>>>
>>>> --
>>>> View this message in context:
>>>> http://apache-spark-user-list.1001560.n3.nabble.com/union-and-reduceByKey-wrong-shuffle-tp23092p23093.html
>>>> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>>>>
>>>> ---------------------------------------------------------------------
>>>> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
>>>> For additional commands, e-mail: user-help@spark.apache.org
>>>>
>>>>
>>>
>
Re: union and reduceByKey wrong shuffle?
Posted by Igor Berman <ig...@gmail.com>.
switching to use simple pojos instead of using avro for spark serialization
solved the problem(I mean reading avro from s3 and than mapping each avro
object to it's pojo serializable counterpart with same fields, pojo is
registered withing kryo)
Any thought where to look for a problem/misconfiguration?
On 31 May 2015 at 22:48, Igor Berman <ig...@gmail.com> wrote:
> Hi
> We are using spark 1.3.1
> Avro-chill (tomorrow will check if its important) we register avro classes
> from java
> Avro 1.7.6
> On May 31, 2015 22:37, "Josh Rosen" <ro...@gmail.com> wrote:
>
>> Which Spark version are you using? I'd like to understand whether this
>> change could be caused by recent Kryo serializer re-use changes in master /
>> Spark 1.4.
>>
>> On Sun, May 31, 2015 at 11:31 AM, igor.berman <ig...@gmail.com>
>> wrote:
>>
>>> after investigation the problem is somehow connected to avro
>>> serialization
>>> with kryo + chill-avro(mapping avro object to simple scala case class and
>>> running reduce on these case class objects solves the problem)
>>>
>>>
>>>
>>>
>>> --
>>> View this message in context:
>>> http://apache-spark-user-list.1001560.n3.nabble.com/union-and-reduceByKey-wrong-shuffle-tp23092p23093.html
>>> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>>>
>>> ---------------------------------------------------------------------
>>> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
>>> For additional commands, e-mail: user-help@spark.apache.org
>>>
>>>
>>
Re: union and reduceByKey wrong shuffle?
Posted by Igor Berman <ig...@gmail.com>.
Hi
We are using spark 1.3.1
Avro-chill (tomorrow will check if its important) we register avro classes
from java
Avro 1.7.6
On May 31, 2015 22:37, "Josh Rosen" <ro...@gmail.com> wrote:
> Which Spark version are you using? I'd like to understand whether this
> change could be caused by recent Kryo serializer re-use changes in master /
> Spark 1.4.
>
> On Sun, May 31, 2015 at 11:31 AM, igor.berman <ig...@gmail.com>
> wrote:
>
>> after investigation the problem is somehow connected to avro serialization
>> with kryo + chill-avro(mapping avro object to simple scala case class and
>> running reduce on these case class objects solves the problem)
>>
>>
>>
>>
>> --
>> View this message in context:
>> http://apache-spark-user-list.1001560.n3.nabble.com/union-and-reduceByKey-wrong-shuffle-tp23092p23093.html
>> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>>
>> ---------------------------------------------------------------------
>> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
>> For additional commands, e-mail: user-help@spark.apache.org
>>
>>
>
Re: union and reduceByKey wrong shuffle?
Posted by Josh Rosen <ro...@gmail.com>.
Which Spark version are you using? I'd like to understand whether this
change could be caused by recent Kryo serializer re-use changes in master /
Spark 1.4.
On Sun, May 31, 2015 at 11:31 AM, igor.berman <ig...@gmail.com> wrote:
> after investigation the problem is somehow connected to avro serialization
> with kryo + chill-avro(mapping avro object to simple scala case class and
> running reduce on these case class objects solves the problem)
>
>
>
>
> --
> View this message in context:
> http://apache-spark-user-list.1001560.n3.nabble.com/union-and-reduceByKey-wrong-shuffle-tp23092p23093.html
> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
>
Re: union and reduceByKey wrong shuffle?
Posted by "igor.berman" <ig...@gmail.com>.
after investigation the problem is somehow connected to avro serialization
with kryo + chill-avro(mapping avro object to simple scala case class and
running reduce on these case class objects solves the problem)
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/union-and-reduceByKey-wrong-shuffle-tp23092p23093.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org