如何使用嵌入式模式从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或其他东西的消息。我无法控制这些事情。

我的问题是，给定一条二进制消息中嵌入模式的字节消息，如何在不声明模式的情况下反序列化Avro文件，以便我可以读取它。

Answer 1

>

添加Maven依赖性

<dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.9.1</version>
    <type>maven-plugin</type>
</dependency>

创建如下文件

 {"namespace": "tachyonis.space",
   "type": "record",
   "name": "Avro",
   "fields": [
      {"name": "Id", "type": "string"},
    ]
  }

以上另存为Avro。src/main/resources中的avsc。

在Eclipse或任何IDE运行中

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, SCHEMA_REGISTRY_URL_CONFIG);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true); 
KafkaConsumer<String, Avro> consumer = new KafkaConsumer<>(props);

消费者/生产者必须在同一台机器上运行。否则，您需要在Windows/Linux中配置主机文件，并将所有组件配置属性从localhost更改为映射到实际IP地址，以便广播给生产者/消费者。否则会出现网络连接问题等错误

Connection to node -3 (/127.0.0.1:9092) could not be established. Broker may not be available

Answer 2

对于DatumReader/Writer，没有像嵌入式模式这样的东西。是我在看Avro时的误解

DataFileWriter在文件的开头编写模式，然后使用GenericDatumWriter添加GenericRecords。

既然您刚开始说有一个模式，我假设您可以读取它，将其转换为模式对象，然后将其传递给GenericDatumReader（模式）构造函数。想知道消息是如何序列化的。也许DataFileWriter用于写入字节[]，而不是实际的文件，那么您可以使用DataFileReader来反序列化数据？