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

Question

假设我有以下映射到 Jackson 完整数据绑定(bind)的 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)
        }
    }
} 

给出以下错误:Caused by: java.io.NotSerializableException: com.fasterxml.jackson.databind.node.ObjectNode

我明白为什么我会收到这样的错误。基本上，Spark 将尝试序列化所有超出范围的变量 a.k.a. 闭包并将其传递给每个执行程序。但是由于 ObjectNode 本身不是 Serializable ，执行者无法获得 Student实例。

我的问题是，我可以通过哪些方式解决这个问题？

我试过使用 Map<String, String>而不是 ObjectNode , 但自 ObjectNode的 put和 set只能有“原语”和JsonNode作为值(value)，当我尝试这样的事情时它会导致错误:

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

Answer 1

有几个选项。

如果您只需要 Object 节点用于 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 中阅读有关自定义序列化的更多信息java文档。

您也可以尝试不同的 data serializers喜欢Kryo .