Wie ich Deserialize Avro von Kafka mit eingebettetem Schema

Question

Ich bekomme binäre Avro-Dateien von einem Kafka-Thema, und ich muss sie deserialisieren. In der von Kafka empfangenen Nachricht kann ich zu Beginn jeder Nachricht ein Schema sehen. Ich weiß, dass es besser ist, das Schema nicht einzubetten und es von der eigentlichen Avro-Datei zu trennen, aber ich habe keine Kontrolle über den Produzenten und kann das nicht ändern.

Mein Code läuft auf Apache Storm. Zunächst erstelle ich einen Leser:

mDatumReader = new GenericDatumReader<GenericRecord>(); 

Und später versuche ich die Nachricht deserialisiert ohne Schema erklärt:

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

Aber dann bekomme ich einen Fehler, wenn eine Nachricht eintrifft:

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:?] 

Bei all den Antworten, die ich gesehen habe, geht es darum, andere Formate zu verwenden, die an Kafka gelieferten Nachrichten zu ändern oder etwas anderes. Ich habe keine Kontrolle über diese Dinge.

Meine Frage ist, eine Nachricht in bytes[] mit eingebetteten Schema innerhalb binärer Nachricht, wie diese AVRO-Datei deserialisieren, ohne das Schema zu deklarieren, damit ich es lesen kann.

Answer 1

Mit dem DatumReader/Writer gibt es so etwas wie ein eingebettetes Schema nicht. War mein Missverständnis, wenn ich Avro & Kafka das erste Mal auch betrachtete. Aber der Quellcode des Avro Serializers zeigt deutlich, dass bei der Verwendung von GenericDatumWriter kein Schema eingebettet ist.

Es ist die Daten Datei Writer, der ein Schema am Anfang der Datei schreibt und dann GenericRecords mit dem GenericDatumWriter hinzufügt.

Da Sie am Anfang ein Schema angegeben haben, gehe ich davon aus, dass Sie es lesen, in ein Schema-Objekt umwandeln und dann in den Konstruktor GenericDatumReader (Schema) übergeben können. Wäre interessant zu wissen, wie die Nachricht serialisiert wird. Vielleicht wird der DataFileWriter verwendet, um in ein Byte [] anstelle einer tatsächlichen Datei zu schreiben, dann könnten Sie den DataFileReader verwenden, um die Daten zu deserialisieren?