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

[srowen]

Make JavaPairRDD.collectAsMap result Serializable since Java Maps generally are

[SparkQA]

QA tests have started for PR 2805 at commit 51c26c2.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2805 at commit 51c26c2.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class SerializableMapWrapper(underlying: collection.Map[K, V])

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21735/
Test PASSed.

[JoshRosen]

mapAsJavaMap is used in a few other places, too, so we should make sure to update those:

• reduceByKeyLocally
• countByKey
• countByKeyApprox
• countByValue
• countByValueApprox

I think it's also used in SparkSQL's Java API Row class and a few other places.

[SparkQA]

QA tests have started for PR 2805 at commit ae1b36f.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2805 at commit ae1b36f.

• This patch fails to build.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21763/
Test FAILed.

[SparkQA]

QA tests have started for PR 2805 at commit f4717f9.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2805 at commit f4717f9.

• This patch fails MiMa tests.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21766/
Test FAILed.

[SparkQA]

QA tests have started for PR 2805 at commit ecb78ee.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2805 at commit ecb78ee.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21769/
Test PASSed.

[JoshRosen]

I can confirm that this seems to have fixed the serialization issue; here's my test-case:

import org.apache.spark.api.java._
val pairs = sc.parallelize(1 to 10).map(x => (x, x))
val map = new JavaPairRDD(pairs).collectAsMap()
def ser(a: AnyRef) =
    (new java.io.ObjectOutputStream(new java.io.ByteArrayOutputStream())).writeObject(a)
ser(map)

It looks like there's one more case in sql/core/src/main/scala/org/apache/spark/sql/api/java/Row.scala that needs to be addressed: https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/api/java/Row.scala#L117. This is a private method, but its return value flows to user-code. I'll fix this up myself on merge.

There still might be some other corner-cases with serializability of results that we haven't tested yet. The result of collect() is serializable, so perhaps this issue only affected our use of MapWrapper. Long term, it would be great to add a fuzz-test that runs random Java API workloads and attempts to serialize their results.

I mentioned this over on JIRA, but for GitHub readers: I've opened an issue to fix this upstream in Scala: https://issues.scala-lang.org/browse/SI-8911

I'll merge this now with my fixup. Thanks!

[JoshRosen]

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.

[JoshRosen]

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.

[JoshRosen]

I made those minor fixups and committed this as f406a83. I also cherry-picked it into branch-1.1. Thanks!

[srowen]

Yes, all SGTM.