Десериализовать сообщения Kafka AVRO с помощью Apache Beam

Вы можете использовать KafkaAvroDeserializer следующим образом:

PCollection<KV<Long,MyClass>> input = p.apply(KafkaIO.<Long, String>read()
.withKeyDeserializer(LongDeserializer.class)
  .withValueDeserializerAndCoder(KafkaAvroDeserializer.class, AvroCoder.of(MyClass.class))

Где Мои занятия — это схема Avro, сгенерированная классом POJO.

Убедитесь, что ваш класс POJO имеет аннотацию AvroCoder, как показано в примере ниже:

@DefaultCoder(AvroCoder.class)
   public class MyClass{
      String name;
      String age;

      MyClass(){}
      MyClass(String n, String a) {
         this.name= n;
         this.age= a;
      }
  }

Замените KafkaIO.<Long, String>read() на KafkaIO.<Long, Object>read().

Если вы посмотрите на реализацию KafkaAvroDeserializer, она реализует Deserializer<Object>:

public class KafkaAvroDeserializer extends AbstractKafkaAvroDeserializer implements Deserializer<Object>

Я столкнулся с той же проблемой. Нашел решение в этом mail-архиве. http://mail-archives.apache.org/mod_mbox/beam-user/201710.mbox/%3CCAMsy_NiVrT_9_xfxOtK1inHxb=x_yAdBcBN+4aquu_hn0GJ0nA@mail.gmail.com%3E

В вашем случае вам нужно определить свой собственный Deserializer<MyClass>, который может простираться от AbstractKafkaAvroDeserializer следующим образом.

public class MyClassKafkaAvroDeserializer extends
  AbstractKafkaAvroDeserializer implements Deserializer<MyClass> {
  
  @Override
  public void configure(Map<String, ?> configs, boolean isKey) {
      configure(new KafkaAvroDeserializerConfig(configs));
  }

  @Override
  public MyClass deserialize(String s, byte[] bytes) {
      return (MyClass) this.deserialize(bytes);
  }

  @Override
  public void close() {} }

Затем укажите свой КафкаАвроДесериализатор как ValueDeserializer.

p.apply(KafkaIO.<Long, MyClass>read()
 .withKeyDeserializer(LongDeserializer.class)
 .withValueDeserializer(MyClassKafkaAvroDeserializer.class) );

Ответ Йохея хорош, но я также обнаружил, что это работает

import io.confluent.kafka.streams.serdes.avro.SpecificAvroDeserializer;

...

public static class CustomKafkaAvroDeserializer extends SpecificAvroDeserializer<MyCustomClass> {}

...
.withValueDeserializerAndCoder(CustomKafkaAvroDeserializer.class, AvroCoder.of(MyCustomClass.class))
...

где MyCustomClass — код, созданный с помощью инструментов Avro.

Сегодня у меня была аналогичная проблема, и я наткнулся на следующий пример, который решил ее для меня.

https://github.com/andrewrjones/debezium-kafka-beam-example/blob/master/src/main/java/com/andrewjones/KafkaAvroConsumerExample.java

недостающей частью для меня был (Class)KafkaAvroDeserializer

KafkaIO.<String, MyClass>read()
        .withBootstrapServers("kafka:9092")
        .withTopic("dbserver1.inventory.customers")
        .withKeyDeserializer(StringDeserializer.class)
        .withValueDeserializerAndCoder((Class)KafkaAvroDeserializer.class, AvroCoder.of(MyClass.class))