java-Spark关闭问题中的Jackson ObjectNode的NotSerializableException

说我有以下映射到Jackson完整数据绑定的Java对象：

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

    // getter and setter ...
}

而且我有一个以下Spark代码,由于作用域不同,它试图序列化Student类型的闭包变量Student.

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

它给出以下错误：原因：java.io.NotSerializableException：com.fasterxml.jackson.databind.node.ObjectNode

我了解为什么我会收到这样的错误.基本上,Spark将尝试序列化所有范围外的变量,也称为闭包,并将其传递给每个执行者.但是由于ObjectNode本身不是可序列化的,因此执行者无法获取Student实例.

我的问题是,我有什么方法可以解决这个问题？

我尝试使用Map< String,String>而不是ObjectNode,但是由于ObjectNode的put和set只能将“ primitives”和JsonNode作为值,因此当我尝试执行类似以下操作时会导致错误：

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

解决方法:

有几种选择.

如果仅出于JSON序列化目的需要对象节点,则可以重写Student类并完全删除ObjectNode.在您的示例中,您可以使用带有firstName和lastName字段的对象替换它

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

但是,如果这不可能,您可以像这样进行自定义序列化

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
    }
}

在我的示例中,我将对象节点序列化为json字符串,但是您当然可以遍历对象节点字段,分别存储每个字段.

您可以在ObjectOutputStream javadoc中阅读有关自定义序列化的更多信息.

您也可以尝试使用不同的data serializers,例如Kryo.