运行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：最近我还打开了另一个线程，有一个类似的（可能相关的）但不完全相同的问题。你在这里找到的。