如何使用 Apache Flink 解决 com.esotericsoftware.kryo.Kryo.readObject 上的 NPE？

Question

我正在为我的 pojo 类使用 Flink 和自定义 kryo 类。但是得到

Caused by: java.lang.NullPointerException
    at MyTreeSerializer.read(MyTreeSerializer.java:36)
    at MyTreeSerializer.read(MyTreeSerializer.java:11)
    at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:657)
    at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:172)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:414)
    ... 16 more

这里有详细信息-

1. Kryo 2.24.0

2. 我的 Pojo 课堂

```

public class MyTree extends TreeMap<String, Object> {
    private String id;
    public String getId() {
        return id;
    }
    public void setId(String id) {
        this.id = id;
    }
}

1. Pojo 的序列化器

```

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Serializer;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.serializers.MapSerializer;


public class MyTreeSerializer extends Serializer<MyTree> {


    public MyTreeSerializer() {
    }


    @Override
    public void write(Kryo kryo, Output output, MyTree object) {
        output.writeString(object.getId());
        kryo.writeObject(output, object, new MapSerializer());

    }

    @Override
    public MyTree read(Kryo kryo, Input input, Class<MyTree> type) {
        String id = input.readString();
        System.out.println("Serialized Id " + id);
        MyTree myTree = kryo.readObject(input, type, new MapSerializer());
        System.out.println("Serialized Object " + myTree);
        myTree.setId(id);
        return myTree;
    }
}

```

4. Flink Streaming 主程序

```

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;


public class MultiSinkTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // Setting Serializer
        env.getConfig().addDefaultKryoSerializer(MyTree.class, MyTreeSerializer.class);

        DataStreamSource<String> data = env.fromElements("1", "2");

        DataStream<MyTree> returns = data.map(new MapFunction<String, MyTree>() {
            @Override
            public MyTree map(String s) throws Exception {
                MyTree myTree = new MyTree();
                myTree.setId(s);
                myTree.put("name", "sohi");
                return myTree;
            }
        }).returns(MyTree.class);


        returns.addSink(new SinkFunction<MyTree>() {
            @Override
            public void invoke(MyTree myTree) throws Exception {
                System.out.println("==> " + myTree.toString());
            }
        });

        env.execute();
    }
}

通过使用所有提到的代码，只有 id 被序列化，而不是 MyTree 的 map 部分。

但如果我替换

env.getConfig().addDefaultKryoSerializer(MyTree.class, MyTreeSerializer.class);

与

env.getConfig().addDefaultKryoSerializer(MyTree.class, MapSerializer.class);

那么 id 未序列化，但 map 正在序列化。

只是需要帮助，为什么它在使用 MyTreeSerializer.class 时不起作用。

提前致谢。

Answer 1

MyTreeSerializer 中的以下行导致null:

MyTree myTree = kryo.readObject(input, type, new MapSerializer());

这也是为什么myTree.setId(id) 会导致

NullPointerException。

当您使用 MapSerializer 时，它可以正常工作（当然除了反序列化 id），因为 MyTree 是从实现 Map 的 TreeMap 扩展而来的。

在MyTreeSerializer 的实现中，您试图从MyTree 对象反序列化MyTree 类的成员。就像MyTreeSerializer 需要一个如下示例代码中的对象：

公共类 MyTree 扩展 TreeMap { 私有字符串 id； 私人的我的树我的树； 公共字符串 getId() { 返回标识； } 公共无效 setId（字符串 id）{ 这个.id = id; } 公共MyTree getMyTree（）{ 返回我的树； } 公共无效 setMyTree（MyTree myTree）{ this.myTree = myTree; } }

它认为您需要查看 MapSerializer 并从它扩展或将其用作您自己的实现的基础，以便序列化和反序列化 MyTree 对象。