spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Swapnil Shinde <swapnilushi...@gmail.com>
Subject Re: Inconsistent results with combineByKey API
Date Tue, 05 Sep 2017 14:30:45 GMT
Ping.. Can someone please correct me whether this is an issue or not.

-
Swapnil

On Thu, Aug 31, 2017 at 12:27 PM, Swapnil Shinde <swapnilushinde@gmail.com>
wrote:

> Hello All
>
> I am observing some strange results with aggregateByKey API which is
> implemented with combineByKey. Not sure if this is by design or bug -
>
> I created this toy example but same problem can be observed on large
> datasets as well -
>
> *case class ABC(key: Int, c1: Int, c2: Int)*
> *case class ABCoutput(key: Int, desc: String, c1Sum: Int, c2Sum: Int)*
>
> // Create RDD and making sure if has 1 or 2 partitions for this example.
> // With 2 partitions there are high chances that same key could be in same
> partition.
> *val a = sc.makeRDD[ABC](List(ABC(1, 10, 20), ABC(1, 10, 20), ABC(2, 20,
> 40), ABC(2, 20, 40))).coalece(2)*
>
> Now, I am running aggregateByKey where I am grouping by Key to sum c1 and
> c2 but return ABCoutput with new 'desc' property.
>
> *val b = a.keyBy(x => x.key).aggregateByKey(ABCoutput(0,"initial",0,0))
> ((x1: ABCoutput, x2: ABC) => ABCoutput(x1.key, "intermediate",
> x1.c1Sum+x2.c1, x1.c2Sum+x2.c2), (m1: ABCoutput, m2:ABCoutput) =>
> ABCoutput(m1.key, "final", m1.c1Sum+m2.c1Sum, m1.c2Sum+m2.c2Sum))*
>
> Above query may return results like this -
> [image: Inline image 1]
>
> It means for one of the keys which has all values in same partition didn't
> invoke mergeCombiner function which returns ABCoutput with desc=final.
>
> I am expecting mergeCombiner function to be invoked all the time which is
> not happening. Correct me if wrong, but is this expected behavior?
>
> Further debugging shows that it works fine if I create input RDD with more
> partitions( which increases chances of having rows with same key in
> different partitions)
>
> *val b = a.repartition(20).keyBy(x =>
> x.key).aggregateByKey(ABCoutput(0,"initial",0,0)) ((x1: ABCoutput, x2: ABC)
> => ABCoutput(x1.key, "intermediate", x1.c1Sum+x2.c1, x1.c2Sum+x2.c2), (m1:
> ABCoutput, m2:ABCoutput) => ABCoutput(m1.key, "final", m1.c1Sum+m2.c1Sum,
> m1.c2Sum+m2.c2Sum))*
> [image: Inline image 2]
>
> One more thing to mention - If I make sure my input RDD is partitioned
> then it simply runs aggregation with mapPartitions (here
> <https://github.com/apache/spark/blob/v2.0.1/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala#L93>).
> Now, this makes sense in terms of aggregations as all values for given key
> are in same partition. However, I have something in my mergeCombiner
> function that I would like to run which wont get invoked.
> Traditional map reduce allows to have different combiner and reduce
> function and it is guaranteed that reduce is always invoked. I can see that
> running aggregations with no shuffle has performance gains but API seems to
> be confusing/misleading. User might hope that mergeCombiner gets invoked
> but in reality it isn't. It will be great if this API designers can shed
> some light on this.
>
> *import org.apache.spark.HashPartitioner*
> *val b = a.keyBy(x => x.key).partitionBy(new
> HashPartitioner(20)).aggregateByKey(ABCoutput(0,"initial",0,0)) ((x1:
> ABCoutput, x2: ABC) => ABCoutput(x1.key, "intermediate", x1.c1Sum+x2.c1,
> x1.c2Sum+x2.c2), (m1: ABCoutput, m2:ABCoutput) => ABCoutput(m1.key,
> "final", m1.c1Sum+m2.c1Sum, m1.c2Sum+m2.c2Sum))*
>
> [image: Inline image 3]
>
> Above examples shows this behavior with AggregateByKey but same thing can
> be observed with CombineByKey as well.
> *val b = a.keyBy(x => x.key).combineByKey( (x: ABC) => ABCoutput(x.key,
> "initial", x.c1, x.c2), *
> *  (x1: ABCoutput, x2: ABC) => ABCoutput(x1.key, "intermediate", x1.c1Sum
> + x2.c1, x1.c2Sum+x2.c2),*
> *  (x1: ABCoutput, x2: ABCoutput) => ABCoutput(x1.key, "final", x1.c1Sum +
> x2.c1Sum, x1.c2Sum+x2.c2Sum))*
>
> *[image: Inline image 4]*
>
>
> Please let me know if you need any further information and correct me if
> my understanding of API is wrong.
>
> Thanks
> Swapnil
>

Mime
View raw message