-
Notifications
You must be signed in to change notification settings - Fork 28.3k
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.
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
SPARK-3926 [CORE] Reopened: result of JavaRDD collectAsMap() is not serializable #3587
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -19,7 +19,8 @@ package org.apache.spark.api.java | |
|
||
import com.google.common.base.Optional | ||
|
||
import scala.collection.convert.Wrappers.MapWrapper | ||
import java.{util => ju} | ||
import scala.collection.mutable | ||
|
||
private[spark] object JavaUtils { | ||
def optionToOptional[T](option: Option[T]): Optional[T] = | ||
|
@@ -32,7 +33,65 @@ private[spark] object JavaUtils { | |
def mapAsSerializableJavaMap[A, B](underlying: collection.Map[A, B]) = | ||
new SerializableMapWrapper(underlying) | ||
|
||
// Implementation is copied from scala.collection.convert.Wrappers.MapWrapper, | ||
// but implements java.io.Serializable and adds a no-arg constructor | ||
class SerializableMapWrapper[A, B](underlying: collection.Map[A, B]) | ||
extends MapWrapper(underlying) with java.io.Serializable | ||
extends ju.AbstractMap[A, B] with java.io.Serializable { self => | ||
|
||
// Add no-arg constructor just for serialization | ||
def this() = this(null) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. You can remove There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It won't have a no-arg constructor in addition to its main one arg constructor otherwise right? This is the point of this change. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I ran your unit test after removing the no-arg constructor, and it worked fine. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Hm, so it does. Maybe I misunderstood the original error. It's complaining about the superclass ( |
||
|
||
override def size = underlying.size | ||
|
||
override def get(key: AnyRef): B = try { | ||
underlying get key.asInstanceOf[A] match { | ||
case None => null.asInstanceOf[B] | ||
case Some(v) => v | ||
} | ||
} catch { | ||
case ex: ClassCastException => null.asInstanceOf[B] | ||
} | ||
|
||
override def entrySet: ju.Set[ju.Map.Entry[A, B]] = new ju.AbstractSet[ju.Map.Entry[A, B]] { | ||
def size = self.size | ||
|
||
def iterator = new ju.Iterator[ju.Map.Entry[A, B]] { | ||
val ui = underlying.iterator | ||
var prev : Option[A] = None | ||
|
||
def hasNext = ui.hasNext | ||
|
||
def next() = { | ||
val (k, v) = ui.next | ||
prev = Some(k) | ||
new ju.Map.Entry[A, B] { | ||
import scala.util.hashing.byteswap32 | ||
def getKey = k | ||
def getValue = v | ||
def setValue(v1 : B) = self.put(k, v1) | ||
override def hashCode = byteswap32(k.hashCode) + (byteswap32(v.hashCode) << 16) | ||
override def equals(other: Any) = other match { | ||
case e: ju.Map.Entry[_, _] => k == e.getKey && v == e.getValue | ||
case _ => false | ||
} | ||
} | ||
} | ||
|
||
def remove() { | ||
prev match { | ||
case Some(k) => | ||
underlying match { | ||
case mm: mutable.Map[a, _] => | ||
mm remove k | ||
prev = None | ||
case _ => | ||
throw new UnsupportedOperationException("remove") | ||
} | ||
case _ => | ||
throw new IllegalStateException("next must be called at least once before remove") | ||
} | ||
} | ||
} | ||
} | ||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Does this create any licensing issues? Do we need to add a separate header somewhere?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good question. It appears to be licensed just like the rest of the Scala code (http://www.scala-lang.org/license.html) Spark already integrates some Scala code and has the proper entries in
LICENSE
as a result. I can modify the text to clearly call out that part ofMapWrapper
was copied, for good measure.