1

我已经实现了一个运算符来从 avro byte[] 反序列化为 Object。之后，我将对象发送到 ConsoleOutputOperator。

public final transient DefaultInputPort<byte[]> input = new DefaultInputPort<byte[]>() {
    @Override
    public void process(byte[] tuple) {
        SpecificDatumReader reader = new SpecificDatumReader<T>(User.getClassSchema());
        BinaryDecoder binaryDecoder = DecoderFactory.get().binaryDecoder(tuple, null);
        try {
            binaryDecoder.readInt();
            obj = (T) reader.read(null, binaryDecoder);
        } catch (IOException e) {
            errorCount++;
            LOG.error("Avro decoder error.");
        }

        if (obj != null) {
            output.emit(""+obj);
            recordCount++;
        }else if (errorPort.isConnected()) {
            errorPort.emit(obj);
            errorCount++;
        }
    }
};

但是，我得到了这个错误。

2016-05-09 10:53:21,481 ERROR com.datatorrent.stram.engine.StreamingContainer: deploy request failed
com.esotericsoftware.kryo.KryoException: Class cannot be created (missing no-arg constructor): org.apache.avro.generic.GenericData$Array
Serialization trace:
user (com.log.avro.User)
obj (com.log.avro.AvroOperator)
at  com.esotericsoftware.kryo.Kryo$DefaultInstantiatorStrategy.newInstantiatorOf(Kryo.java:1228)
at com.esotericsoftware.kryo.Kryo.newInstantiator(Kryo.java:1049)
at com.esotericsoftware.kryo.Kryo.newInstance(Kryo.java:1058)
at com.esotericsoftware.kryo.serializers.CollectionSerializer.create(CollectionSerializer.java:89)
at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:93)
at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:22)
at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
at com.datatorrent.common.util.FSStorageAgent.retrieve(FSStorageAgent.java:193)
at com.datatorrent.common.util.FSStorageAgent.load(FSStorageAgent.java:137)
at com.datatorrent.stram.engine.StreamingContainer.deployNodes(StreamingContainer.java:894)
at com.datatorrent.stram.engine.StreamingContainer.deploy(StreamingContainer.java:843)
at com.datatorrent.stram.engine.StreamingContainer.processHeartbeatResponse(StreamingContainer.java:801)
at com.datatorrent.stram.engine.StreamingContainer.heartbeatLoop(StreamingContainer.java:694)
at com.datatorrent.stram.engine.StreamingContainer.main(StreamingContainer.java:312)

我认为它不支持“ArrayList”。

kryo.readObject(input, ArrayList.class);

如何修复此错误？

4

1 回答 1

3

看看：http ://avro.apache.org/docs/1.8.0/api/java/org/apache/avro/generic/GenericData.html

您会注意到 GenericData.Array 没有 Kryo 序列化/反序列化所需的默认构造函数（即没有参数的构造函数）。

您可以使用注释：@DefaultSerializer(JavaSerializer.class)，如果该类是 Java 可序列化的，如 Apex Malhar 中的 library/src/main/java/com/datatorrent/lib/codec/KryoJdkContainer.java 中所做的那样。

请查看：http ://docs.datatorrent.com/troubleshooting/#application-throwing-following-kryo-exception 了解其他选项。

于 2016-05-10T05:19:45.130 回答