Spark 闭包问题中 Jackson ObjectNode 的 NotSerializableException答案

【问题标题】：NotSerializableException for Jackson ObjectNode in Spark closure issueSpark 闭包问题中 Jackson ObjectNode 的 NotSerializableException
【发布时间】：2017-01-22 19:53:41
【问题描述】：

假设我有以下映射到 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)
        }
    }
}

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

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

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

我曾尝试使用Map<String, String> 而不是ObjectNode，但由于ObjectNode 的put 和set 只能将“原始”和JsonNode 作为值，因此当我尝试某些操作时会导致错误像这样：

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

【问题讨论】：

标签： java json scala jackson serializable

【解决方案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 javadoc 中阅读有关自定义序列化的更多信息。

您还可以尝试不同的data serializers，例如Kryo。

【讨论】：

【解决方案2】：

我最终将student 变成了Map<String, String> 对象，并在需要时将mapper.convertValue(student, ObjectNode.class 放在ObjectNode 中

【讨论】：

【解决方案3】：

将 jackson-databind 升级到 10.x 会有所帮助，请参阅 https://github.com/FasterXML/jackson-databind/issues/18

【讨论】：