From 4f1dcd3dce270268b42fbe59409790364fa5c5df Mon Sep 17 00:00:00 2001
From: WeichenXu <WeichenXu123@outlook.com>
Date: Sat, 22 Oct 2016 11:59:28 -0700
Subject: [PATCH] [SPARK-18051][SPARK CORE] fix bug of custom
PartitionCoalescer causing serialization exception
## What changes were proposed in this pull request?
add a require check in `CoalescedRDD` to make sure the passed in `partitionCoalescer` to be `serializable`.
and update the document for api `RDD.coalesce`
## How was this patch tested?
Manual.(test code in jira [SPARK-18051])
Author: WeichenXu <WeichenXu123@outlook.com>
Closes #15587 from WeichenXu123/fix_coalescer_bug.
---
core/src/main/scala/org/apache/spark/rdd/CoalescedRDD.scala | 4 ++++
core/src/main/scala/org/apache/spark/rdd/RDD.scala | 3 ++-
2 files changed, 6 insertions(+), 1 deletion(-)
diff --git a/core/src/main/scala/org/apache/spark/rdd/CoalescedRDD.scala b/core/src/main/scala/org/apache/spark/rdd/CoalescedRDD.scala
index 9c198a61f3..2cba1febe8 100644
--- a/core/src/main/scala/org/apache/spark/rdd/CoalescedRDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/CoalescedRDD.scala
@@ -80,6 +80,10 @@ private[spark] class CoalescedRDD[T: ClassTag](
require(maxPartitions > 0 || maxPartitions == prev.partitions.length,
s"Number of partitions ($maxPartitions) must be positive.")
+ if (partitionCoalescer.isDefined) {
+ require(partitionCoalescer.get.isInstanceOf[Serializable],
+ "The partition coalescer passed in must be serializable.")
+ }
override def getPartitions: Array[Partition] = {
val pc = partitionCoalescer.getOrElse(new DefaultPartitionCoalescer())
diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
index be119578d2..db535de9e9 100644
--- a/core/src/main/scala/org/apache/spark/rdd/RDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
@@ -432,7 +432,8 @@ abstract class RDD[T: ClassTag](
* of partitions. This is useful if you have a small number of partitions,
* say 100, potentially with a few partitions being abnormally large. Calling
* coalesce(1000, shuffle = true) will result in 1000 partitions with the
- * data distributed using a hash partitioner.
+ * data distributed using a hash partitioner. The optional partition coalescer
+ * passed in must be serializable.
*/
def coalesce(numPartitions: Int, shuffle: Boolean = false,
partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
--
GitLab