apache-kafka - 我如何使用嵌入式模式从 Kafka 反序列化 Avro

Question

我从 Kafka 主题收到二进制 Avro 文件，我必须反序列化它们。在 Kafka 收到的消息中，我可以在每条消息的开头看到一个模式。我知道最好不要嵌入架构并将其与实际的 Avro 文件分开，但我无法控制生产者，也无法更改它。

我的代码在 Apache Storm 之上运行。首先我创建一个阅读器:

mDatumReader = new GenericDatumReader<GenericRecord>();

然后我尝试在不声明架构的情况下反序列化消息:

Decoder decoder = DecoderFactory.get().binaryDecoder(messageBytes, null);
GenericRecord payload = mDatumReader.read(null, decoder);

但是当消息到达时我得到一个错误:

Caused by: java.lang.NullPointerException: writer cannot be null!
at org.apache.avro.io.ResolvingDecoder.resolve(ResolvingDecoder.java:77) ~[stormjar.jar:?]
at org.apache.avro.io.ResolvingDecoder.<init>(ResolvingDecoder.java:46) ~[stormjar.jar:?]
at org.apache.avro.io.DecoderFactory.resolvingDecoder(DecoderFactory.java:307) ~[stormjar.jar:?]
at org.apache.avro.generic.GenericDatumReader.getResolver(GenericDatumReader.java:122) ~[stormjar.jar:?]
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:137) ~[stormjar.jar:?]

我看到的所有答案都是关于使用其他格式，更改传递给 Kafka 或其他东西的消息。我无法控制这些事情。

我的问题是，给定 bytes[] 中的消息，在二进制消息中嵌入架构，如何在不声明架构的情况下反序列化该 Avro 文件，以便我可以读取它。

Answer 1

有了 DatumReader/Writer，就没有嵌入式模式这样的东西了。也是我第一次看 Avro & Kafka 时的误解。但是 Avro Serializer 的源代码清楚地表明在使用 GenericDatumWriter 时没有嵌入模式。

DataFileWriter 会在文件的开头写入架构，然后使用 GenericDatumWriter 添加 GenericRecords。

既然你说开头有一个模式，我假设你可以读取它，将它变成一个模式对象，然后将它传递给 GenericDatumReader(schema) 构造函数。知道消息是如何序列化的会很有趣。也许 DataFileWriter 用于写入 byte[] 而不是实际文件，那么您可以使用 DataFileReader 反序列化数据？