Kryo : désérialisation d'une ancienne version d'une classe

Question

Je dois modifier une classe en ajoutant deux nouveaux paramètres. Cette classe est sérialisée avec Kryo. Je persiste actuellement les informations relatives à cette classe, entre autres, en tant que RDD, à chaque fois que j'arrête mon flux. Lorsque je redémarre le flux, je charge les informations que j'ai précédemment persistées et je les utilise pour avoir une cohérence entre le moment où j'ai arrêté et le moment où je redémarre.

Puisque la classe que je persiste a besoin de ces nouveaux paramètres, j'ai modifié la classe et le sérialiseur, en ajoutant le nouvel élément kryo.writeObject(output, object, ObjectSerializer) et le kryo.readObject(input, classOf[Object], ObjectSerializer) pour les nouveaux paramètres.

Maintenant, chaque fois que je redémarre mon flux, j'obtiens une exception : " Encountered unregistered class ...".

Cela semble évident, car j'essaie de désérialiser un objet qui n'est pas contenu dans les informations que j'ai conservées lorsque j'ai arrêté le flux. Si je supprime ces données et que je démarre le flux comme s'il n'avait pas été exécuté auparavant, l'exception ne se produit pas.

Existe-t-il un moyen d'éviter cette exception ? Peut-être en spécifiant des valeurs par défaut au cas où ces paramètres seraient manquants ?

Merci de votre attention

EDITAR:

J'ai trouvé quelque chose d'utile que je n'avais pas vu auparavant : Numéro Kryo 194 .

Ce type a implémenté le versioning en insérant simplement un long définissant la version du désérialiseur qu'il doit utiliser. C'est une solution simple mais, comme la société qui a écrit le code sur lequel je travaille n'a pas pensé à la compatibilité ascendante, je suppose que je vais devoir jeter par la fenêtre toutes les données qui ont été persistées avant le nouveau sérialiseur.

N'hésitez pas à me faire savoir si quelqu'un peut vous suggérer une meilleure solution.

EDIT 2 :

Cette situation pose toujours problème. J'ai essayé d'utiliser le CompatibleFieldSerializer comme décrit ici : Exemple de CompatibleFieldSerializer Ainsi, en enregistrant ce sérialiseur et non le sérialiseur personnalisé utilisé précédemment. Le résultat est que maintenant, lors du rechargement des données persistantes, il donne un java.lang.NullPointerException . Il n'y a toujours pas de problème si aucune donnée antérieure n'a été conservée. Je peux démarrer mon flux, sérialiser les nouvelles données, arrêter le flux, désérialiser et redémarrer mon flux. Toujours pas d'indice sur la résolution.

Answer 1

La solution à ce problème a été trouvée il y a quelques mois. J'ai donc pensé à poster une réponse à cette question dès que possible. Le problème résidait dans le fait que, à cause d'une erreur dans le code, la classe était sérialisée avec un FieldSerializer Kryo standard, qui n'est pas compatible avec l'avenir. Nous avons dû effectuer les actions suivantes pour désérialiser l'ancienne classe et la convertir en une nouvelle classe sérialisée.

La situation était la suivante :

case class ClassA(field1 : Long, field2 : String)

Il était présenté comme suit :

object ClassASerializer extends Serializer[ClassA] with Serializable{
  override def write(kryo: Kryo, output: Output, t: ClassA) = {
      output.writeLong    { t.field1 }
      output.writeString  { t.field2 }
 }
  override def read(kryo: Kryo, input: Input, aClass: Class[ClassA]) = 
       classA( 
           field1 = input.readLong(),
           field2 = input.readLong()
       )

Et un Seq contenant les classes à sérialiser avec un sérialiseur a été mis en boucle afin d'enregistrer tous les sérialiseurs pour toutes les classes.

    protected def registry: Seq[aClass: Class[A], serializer: Serializer[A]] = ...
    final def register(kryo: Kryo) = {
         registry.foreach { registrable => kryo.register(registrable.aClass, registrable.serializer) }
    }

La classe a dû être modifiée en ajoutant un nouveau champ, qui est l'instance d'une autre classe de cas.

Pour effectuer ce changement, nous avons dû utiliser une annotation relative à la bibliothèque Kryo "optional",

...
import com.esotericsoftware.kryo.serializers.FieldSerializer.Optional
import scala.annotation.meta.field
...

case class ClassA(field1 : Long, field2 : String,  @(Optional @field)("field3") field3 : ClassB)

Le sérialiseur a été modifié de manière à ce que, lors de la lecture de l'ancienne classe sérialisée, il puisse instancier le champ 3 avec une valeur par défaut et, lors de l'écriture, écrire cette valeur par défaut :

object ClassASerializer extends Serializer[ClassA] with Serializable{
  override def write(kryo: Kryo, output: Output, t: ClassA) = {
      output.writeLong    { t.field1 }
      output.writeString  { t.field2 }
      kryo.writeObject(output, Option { t.field3 } getOrElse ClassB.default, ClassBSerializer)

 }
  override def read(kryo: Kryo, input: Input, aClass: Class[ClassA]) = 
       ClassA( 
           field1 = input.readLong(),
           field2 = input.readLong(),
           field3 = ClassB.default
       )

L'enregistrement des sérialiseurs kryo a également été modifié pour enregistrer également le champ optionnel :

    protected def registry: Seq[aClass: Class[A], serializer: Serializer[A]] = ...
    def optionals = Seq("field3")

    final def register(kryo: Kryo) = {
        optionals.foreach { optional =>
        kryo.getContext.asInstanceOf[ObjectMap[Any, Any]].put(optional, true) }
        registry.foreach { registrable => kryo.register(registrable.aClass, registrable.serializer) }
    }

Nous avons ainsi pu écrire la nouvelle version de la classe sérialisée. Ensuite, nous avons dû supprimer l'annotation optionnelle, modifier le sérialiseur afin de lire le champ réel de la nouvelle classe sérialisée, et supprimer l'enregistrement du sérialiseur optionnel et l'ajouter au registre Seq.

Entre-temps, nous avons corrigé l'erreur dans le code qui imposait la sérialisation via FieldSerializer, mais cela n'entre pas dans le cadre de la question.