Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

SPARK-3926 [CORE] Result of JavaRDD.collectAsMap() is not Serializable #2805

Closed
wants to merge 4 commits into from
Closed

SPARK-3926 [CORE] Result of JavaRDD.collectAsMap() is not Serializable #2805

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
51c26c2
Make JavaPairRDD.collectAsMap result Serializable since Java Maps gen…
srowen Oct 14, 2014
ae1b36f
Expand to cover Maps returned from other Java API methods as well
srowen Oct 15, 2014
f4717f9
Oops, fix compile problem
srowen Oct 15, 2014
ecb78ee
Fix conflict between java.io.Serializable and use of Scala's Serializ…
srowen Oct 15, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 6 additions & 5 deletions core/src/main/scala/org/apache/spark/api/java/JavaPairRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -265,10 +265,10 @@ class JavaPairRDD[K, V](val rdd: RDD[(K, V)])
* before sending results to a reducer, similarly to a "combiner" in MapReduce.
*/
def reduceByKeyLocally(func: JFunction2[V, V, V]): java.util.Map[K, V] =
mapAsJavaMap(rdd.reduceByKeyLocally(func))
mapAsSerializableJavaMap(rdd.reduceByKeyLocally(func))

/** Count the number of elements for each key, and return the result to the master as a Map. */
def countByKey(): java.util.Map[K, Long] = mapAsJavaMap(rdd.countByKey())
def countByKey(): java.util.Map[K, Long] = mapAsSerializableJavaMap(rdd.countByKey())

/**
* :: Experimental ::
Expand All @@ -277,7 +277,7 @@ class JavaPairRDD[K, V](val rdd: RDD[(K, V)])
*/
@Experimental
def countByKeyApprox(timeout: Long): PartialResult[java.util.Map[K, BoundedDouble]] =
rdd.countByKeyApprox(timeout).map(mapAsJavaMap)
rdd.countByKeyApprox(timeout).map(mapAsSerializableJavaMap)

/**
* :: Experimental ::
Expand All @@ -287,7 +287,7 @@ class JavaPairRDD[K, V](val rdd: RDD[(K, V)])
@Experimental
def countByKeyApprox(timeout: Long, confidence: Double = 0.95)
: PartialResult[java.util.Map[K, BoundedDouble]] =
rdd.countByKeyApprox(timeout, confidence).map(mapAsJavaMap)
rdd.countByKeyApprox(timeout, confidence).map(mapAsSerializableJavaMap)

/**
* Aggregate the values of each key, using given combine functions and a neutral "zero value".
Expand Down Expand Up @@ -614,7 +614,8 @@ class JavaPairRDD[K, V](val rdd: RDD[(K, V)])
/**
* Return the key-value pairs in this RDD to the master as a Map.
*/
def collectAsMap(): java.util.Map[K, V] = mapAsJavaMap(rdd.collectAsMap())
def collectAsMap(): java.util.Map[K, V] = mapAsSerializableJavaMap(rdd.collectAsMap())


/**
* Pass each value in the key-value pair RDD through a map function without changing the keys;
Expand Down
13 changes: 10 additions & 3 deletions core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,7 +390,7 @@ trait JavaRDDLike[T, This <: JavaRDDLike[T, This]] extends Serializable {
* combine step happens locally on the master, equivalent to running a single reduce task.
*/
def countByValue(): java.util.Map[T, java.lang.Long] =
mapAsJavaMap(rdd.countByValue().map((x => (x._1, new java.lang.Long(x._2)))))
mapAsSerializableJavaMap(rdd.countByValue().map((x => (x._1, new java.lang.Long(x._2)))))

/**
* (Experimental) Approximate version of countByValue().
Expand All @@ -399,13 +399,13 @@ trait JavaRDDLike[T, This <: JavaRDDLike[T, This]] extends Serializable {
timeout: Long,
confidence: Double
): PartialResult[java.util.Map[T, BoundedDouble]] =
rdd.countByValueApprox(timeout, confidence).map(mapAsJavaMap)
rdd.countByValueApprox(timeout, confidence).map(mapAsSerializableJavaMap)

/**
* (Experimental) Approximate version of countByValue().
*/
def countByValueApprox(timeout: Long): PartialResult[java.util.Map[T, BoundedDouble]] =
rdd.countByValueApprox(timeout).map(mapAsJavaMap)
rdd.countByValueApprox(timeout).map(mapAsSerializableJavaMap)

/**
* Take the first num elements of the RDD. This currently scans the partitions *one by one*, so
Expand Down Expand Up @@ -587,4 +587,11 @@ trait JavaRDDLike[T, This <: JavaRDDLike[T, This]] extends Serializable {
rdd.foreachAsync(x => f.call(x))
}

private[java] def mapAsSerializableJavaMap[A, B](underlying: collection.Map[A, B]) =
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that these methods should live in JavaUtils rather than in this trait so that they can be used by the Streaming and SQL Java APIs, too.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, private[java] methods here will become public from Java's POV. We can't use Scala package-private access modifiers when implementing Java API classes.

new SerializableMapWrapper(underlying)

private[java] class SerializableMapWrapper[A, B](underlying: collection.Map[A, B])
extends MapWrapper(underlying) with java.io.Serializable


}