You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Al M <al...@gmail.com> on 2015/06/17 08:45:49 UTC
Shuffle produces one huge partition
I have 2 RDDs I want to Join. We will call them RDD A and RDD B. RDD A has
1 billion rows; RDD B has 100k rows. I want to join them on a single key.
95% of the rows in RDD A have the same key to join with RDD B. Before I can
join the two RDDs, I must map them to tuples where the first element is the
key and the second is the value.
Since 95% of the rows in RDD A have the same key, they now go into the same
partition. When I perform the join, the system will try to execute this
partition in just one task. This one task will try to load too much data
into memory at once and die a horrible death.
I know that this is caused by the HashPartitioner that is used by default in
Spark; everything with the same hashed key goes into the same partition. I
also tried the RangePartitioner but still saw 95% of the data go into the
same partition. What I'd really like is a partitioner that puts everything
with the same key into the same partition *except* when the partition is
over a certain size, then it would just spill into the next partition.
Writing my own partitioner is a big job, and requires a lot of testing to
make sure I get it right. Is there a simpler way to solve this?
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Shuffle-produces-one-huge-partition-tp23358.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: Shuffle produces one huge partition
Posted by Akhil Das <ak...@sigmoidanalytics.com>.
Can you try repartitioning the rdd after creating the K,V. And also, while
calling the rdd1.join(rdd2, Pass the # partition argument too)
Thanks
Best Regards
On Wed, Jun 17, 2015 at 12:15 PM, Al M <al...@gmail.com> wrote:
> I have 2 RDDs I want to Join. We will call them RDD A and RDD B. RDD A
> has
> 1 billion rows; RDD B has 100k rows. I want to join them on a single key.
>
> 95% of the rows in RDD A have the same key to join with RDD B. Before I
> can
> join the two RDDs, I must map them to tuples where the first element is the
> key and the second is the value.
>
> Since 95% of the rows in RDD A have the same key, they now go into the same
> partition. When I perform the join, the system will try to execute this
> partition in just one task. This one task will try to load too much data
> into memory at once and die a horrible death.
>
> I know that this is caused by the HashPartitioner that is used by default
> in
> Spark; everything with the same hashed key goes into the same partition. I
> also tried the RangePartitioner but still saw 95% of the data go into the
> same partition. What I'd really like is a partitioner that puts everything
> with the same key into the same partition *except* when the partition is
> over a certain size, then it would just spill into the next partition.
>
> Writing my own partitioner is a big job, and requires a lot of testing to
> make sure I get it right. Is there a simpler way to solve this?
>
>
>
> --
> View this message in context:
> http://apache-spark-user-list.1001560.n3.nabble.com/Shuffle-produces-one-huge-partition-tp23358.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: Shuffle produces one huge partition and many tiny partitions
Posted by Corey Nolet <cj...@gmail.com>.
Sorry Du,
Repartition means coalesce(shuffle = true) as per [1]. They are the same
operation. Coalescing with shuffle = false means you are specifying the max
amount of partitions after the coalesce (if there are less partitions you
will end up with the lesser amount.
[1]
https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/rdd/RDD.scala#L341
On Thu, Jun 18, 2015 at 7:55 PM, Du Li <li...@yahoo-inc.com.invalid> wrote:
> repartition() means coalesce(shuffle=false)
>
>
>
> On Thursday, June 18, 2015 4:07 PM, Corey Nolet <cj...@gmail.com>
> wrote:
>
>
> Doesn't repartition call coalesce(shuffle=true)?
> On Jun 18, 2015 6:53 PM, "Du Li" <li...@yahoo-inc.com.invalid> wrote:
>
> I got the same problem with rdd,repartition() in my streaming app, which
> generated a few huge partitions and many tiny partitions. The resulting
> high data skew makes the processing time of a batch unpredictable and often
> exceeding the batch interval. I eventually solved the problem by using
> rdd.coalesce() instead, which however is expensive as it yields a lot of
> shuffle traffic and also takes a long time.
>
> Du
>
>
>
> On Thursday, June 18, 2015 1:00 AM, Al M <al...@gmail.com>
> wrote:
>
>
> Thanks for the suggestion. Repartition didn't help us unfortunately. It
> still puts everything into the same partition.
>
> We did manage to improve the situation by making a new partitioner that
> extends HashPartitioner. It treats certain "exception" keys differently.
> These keys that are known to appear very often are assigned random
> partitions instead of using the existing partitioning mechanism.
>
>
>
> --
> View this message in context:
> http://apache-spark-user-list.1001560.n3.nabble.com/Shuffle-produces-one-huge-partition-and-many-tiny-partitions-tp23358p23387.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: Shuffle produces one huge partition and many tiny partitions
Posted by Du Li <li...@yahoo-inc.com.INVALID>.
repartition() means coalesce(shuffle=false)
On Thursday, June 18, 2015 4:07 PM, Corey Nolet <cj...@gmail.com> wrote:
Doesn't repartition call coalesce(shuffle=true)?On Jun 18, 2015 6:53 PM, "Du Li" <li...@yahoo-inc.com.invalid> wrote:
I got the same problem with rdd,repartition() in my streaming app, which generated a few huge partitions and many tiny partitions. The resulting high data skew makes the processing time of a batch unpredictable and often exceeding the batch interval. I eventually solved the problem by using rdd.coalesce() instead, which however is expensive as it yields a lot of shuffle traffic and also takes a long time.
Du
On Thursday, June 18, 2015 1:00 AM, Al M <al...@gmail.com> wrote:
Thanks for the suggestion. Repartition didn't help us unfortunately. It
still puts everything into the same partition.
We did manage to improve the situation by making a new partitioner that
extends HashPartitioner. It treats certain "exception" keys differently.
These keys that are known to appear very often are assigned random
partitions instead of using the existing partitioning mechanism.
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Shuffle-produces-one-huge-partition-and-many-tiny-partitions-tp23358p23387.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: Shuffle produces one huge partition and many tiny partitions
Posted by Corey Nolet <cj...@gmail.com>.
Doesn't repartition call coalesce(shuffle=true)?
On Jun 18, 2015 6:53 PM, "Du Li" <li...@yahoo-inc.com.invalid> wrote:
> I got the same problem with rdd,repartition() in my streaming app, which
> generated a few huge partitions and many tiny partitions. The resulting
> high data skew makes the processing time of a batch unpredictable and often
> exceeding the batch interval. I eventually solved the problem by using
> rdd.coalesce() instead, which however is expensive as it yields a lot of
> shuffle traffic and also takes a long time.
>
> Du
>
>
>
> On Thursday, June 18, 2015 1:00 AM, Al M <al...@gmail.com>
> wrote:
>
>
> Thanks for the suggestion. Repartition didn't help us unfortunately. It
> still puts everything into the same partition.
>
> We did manage to improve the situation by making a new partitioner that
> extends HashPartitioner. It treats certain "exception" keys differently.
> These keys that are known to appear very often are assigned random
> partitions instead of using the existing partitioning mechanism.
>
>
>
> --
> View this message in context:
> http://apache-spark-user-list.1001560.n3.nabble.com/Shuffle-produces-one-huge-partition-and-many-tiny-partitions-tp23358p23387.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: Shuffle produces one huge partition and many tiny partitions
Posted by Du Li <li...@yahoo-inc.com.INVALID>.
I got the same problem with rdd,repartition() in my streaming app, which generated a few huge partitions and many tiny partitions. The resulting high data skew makes the processing time of a batch unpredictable and often exceeding the batch interval. I eventually solved the problem by using rdd.coalesce() instead, which however is expensive as it yields a lot of shuffle traffic and also takes a long time.
Du
On Thursday, June 18, 2015 1:00 AM, Al M <al...@gmail.com> wrote:
Thanks for the suggestion. Repartition didn't help us unfortunately. It
still puts everything into the same partition.
We did manage to improve the situation by making a new partitioner that
extends HashPartitioner. It treats certain "exception" keys differently.
These keys that are known to appear very often are assigned random
partitions instead of using the existing partitioning mechanism.
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Shuffle-produces-one-huge-partition-and-many-tiny-partitions-tp23358p23387.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: Shuffle produces one huge partition and many tiny partitions
Posted by Al M <al...@gmail.com>.
Thanks for the suggestion. Repartition didn't help us unfortunately. It
still puts everything into the same partition.
We did manage to improve the situation by making a new partitioner that
extends HashPartitioner. It treats certain "exception" keys differently.
These keys that are known to appear very often are assigned random
partitions instead of using the existing partitioning mechanism.
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Shuffle-produces-one-huge-partition-and-many-tiny-partitions-tp23358p23387.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