NotSerializableException pour Jackson ObjectNode en cause de fermeture Spark

Question

Dire que j'ai l'objet Java suivante qui correspond à Jackson données complètes de liaison:

public class Student implements Serializable{ 
    private ObjectNode name; // two keys: "first_name", "last_name" 

    // getter and setter ... 
} 

Et j'ai un code suivant Spark qui tente de sérialiser la variable de fermeture student de type Student en raison de portées différentes.

class A(student : Student) extends Serializable { 
    def process(input: DataFrame): Unit = { 
     val test = input.map { a => 
      print(student) 
     } 
    } 
} 

qui donne l'erreur suivante: Caused by: java.io.NotSerializableException: com.fasterxml.jackson.databind.node.ObjectNode

Je comprends pourquoi je reçois une telle erreur. Fondamentalement, Spark tentera de sérialiser toutes les fermetures hors-champ et fermera le tout à chaque exécuteur. Mais puisque le ObjectNode lui-même n'est pas Serializable, l'exécuteur ne peut pas obtenir les instances Student. Ma question est, quelles sont les façons dont je peux résoudre ce problème?

J'ai essayé d'utiliser Map<String, String> au lieu de ObjectNode, mais depuis ObjectNode « s put et set ne peut avoir des « primitifs » et JsonNode en tant que valeur, il provoque une erreur lorsque je tente quelque chose comme ceci:

ObjectNode meta_info = JsonNodeFactory.instance.objectNode(); 
meta_info.set("field name", student.getName()); 

Answer 1

Il est plusieurs options.

Si vous avez besoin d'un nœud Object uniquement pour la sérialisation json, vous pouvez réécrire votre classe Student et supprimer complètement ObjectNode. Dans votre exemple, vous pouvez subsitute par objet avec firstName et lastName champs

class Name implements Serializable { 
    String firstName; 
    String lastName; 
} 

Toutefois, si cela est impossible, vous pouvez le faire sérialisation personnalisé pour comme celui-ci

public class Student implements Serializable { 
    private transient ObjectNode name; 

    private void writeObject(ObjectOutputStream out) throws IOException { 
     ObjectMapper mapper = new ObjectMapper(); 
     out.writeUTF(mapper.writeValueAsString(name)); 
     // other fields here 
    } 

    private void readObject(ObjectInputStream in) throws IOException, 
      ClassNotFoundException { 
     ObjectMapper mapper = new ObjectMapper(); 

     JsonNode node = mapper.readTree(in.readUTF()); 
     if (!node.isObject()) { 
      throw new IOException("malformed name field detected"); 
     } 

     name = (ObjectNode) node; 

     // read other fields 
    } 
} 

Dans mon exemple, je sérialisé noeud objet json string, mais vous pouvez bien entendu parcourir les champs de nœuds d'objet en stockant chaque champ séparément.

Vous pouvez en savoir plus sur la sérialisation personnalisée dans ObjectOutputStream javadoc.

Aussi vous pouvez expérimenter avec différents data serializers comme Kryo.

Answer 2

je fini par faire student pour objet Map<String, String>, et faire mapper.convertValue(student, ObjectNode.class chaque fois que j'avais besoin d'être dans le ObjectNode