运行apache flink程序会导致一个错误,这与kryo序列化有关。我还认为它不涉及flink本身,但很可能是我使用的数据结构。
我得到的例外是:
Caused by: com.esotericsoftware.kryo.KryoException: Unable to find class: ^@^@ ^A^A^@it.polimi.genomics.core.DataTypes$GDouble
at com.esotericsoftware.kryo.util.DefaultClassResolver.readName(DefaultClassResolver.java:138)
at com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:115)
at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:752)
at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:211)
at org.apache.flink.api.scala.typeutils.TraversableSerializer.deserialize(TraversableSerializer.scala:108)
at org.apache.flink.api.scala.typeutils.TraversableSerializer.deserialize(TraversableSerializer.scala:30)
at org.apache.flink.api.common.typeutils.base.GenericArraySerializer.deserialize(GenericArraySerializer.java:123)
at org.apache.flink.api.common.typeutils.base.GenericArraySerializer.deserialize(GenericArraySerializer.java:33)
at org.apache.flink.api.scala.typeutils.CaseClassSerializer.deserialize(CaseClassSerializer.scala:102)
at org.apache.flink.api.scala.typeutils.CaseClassSerializer.deserialize(CaseClassSerializer.scala:95)
at org.apache.flink.api.scala.typeutils.CaseClassSerializer.deserialize(CaseClassSerializer.scala:29)
at org.apache.flink.runtime.operators.sort.NormalizedKeySorter.getRecordFromBuffer(NormalizedKeySorter.java:338)
at org.apache.flink.runtime.operators.sort.NormalizedKeySorter.access$300(NormalizedKeySorter.java:40)
at org.apache.flink.runtime.operators.sort.NormalizedKeySorter$1.next(NormalizedKeySorter.java:443)
at org.apache.flink.runtime.operators.ReduceCombineDriver.sortAndCombine(ReduceCombineDriver.java:226)
at org.apache.flink.runtime.operators.ReduceCombineDriver.run(ReduceCombineDriver.java:157)
at org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:496)
at org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:362)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassNotFoundException: ^@^@ ^A^A^@it.polimi.genomics.core.DataTypes$GDouble
at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:278)
at com.esotericsoftware.kryo.util.DefaultClassResolver.readName(DefaultClassResolver.java:136)
... 20 more
这是所涉及的数据结构:
object DataTypes {
@SerialVersionUID(2212l)
sealed trait GValue extends Serializable with Comparable[GValue] with Ordered[GValue]{
def compare(o : GValue) : Int = {
....
}
def equal(o : GValue) : Boolean = {
...
}
override def compareTo(o: GValue): Int = {
....
}
}
@SerialVersionUID(2213l)
case class GInt(v: Int) extends GValue{
def this() = this(0)
override def toString() : String = {
...
}
override def equals(other : Any) : Boolean = {
....
}
}
@SerialVersionUID(2214l)
case class GDouble(v: Double) extends GValue {
def this() = this(0.0)
override def toString() : String = {
.....
}
override def equals(other : Any) : Boolean = {
....
}
}
@SerialVersionUID(2215l)
case class GString(v: String) extends GValue{
def this() = this(".")
override def toString() : String = {
.....
}
override def equals(other : Any) : Boolean = {
.....
}
}
@SerialVersionUID(2216l)
case class GNull() extends GValue{
override def toString() : String = {
.....
}
override def equals(other : Any) : Boolean = {
other match {
.....
}
}
}
你能提供一些帮助或提示吗?我被困住了。
ps:最近我还打开了另一个线程,有一个类似的(可能相关的)但不完全相同的问题。你在这里找到的。
暂无答案!
目前还没有任何答案,快来回答吧!