如何使用apachebeam反序列化kafka avro消息

0wi1tuuw 于 2021-06-06 发布在 Kafka

关注(0)|答案(5)|浏览(421)

主要目标是聚合两个Kafka主题，一个压缩慢移动数据和另一个每秒接收的快速移动数据。
我已经能够在一些简单的场景中使用消息，比如kv（长，字符串），使用如下方式：

PCollection<KV<Long,String>> input = p.apply(KafkaIO.<Long, 
String>read()
.withKeyDeserializer(LongDeserializer.class)
.withValueDeserializer(StringDeserializer.class)

PCollection<String> output = input.apply(Values.<String>create());

但当您需要从avro反序列化时，这似乎不是一种方法。我有一个千伏（字符串，avro）我需要消费。
我尝试从avro模式生成java类，然后将它们包含在“apply”中，例如：

PCollection<MyClass> output = input.apply(Values.<MyClass>create());

但这似乎不是正确的方法。
是否有任何人可以向我指出的文档/示例，以便我了解您将如何与Kafka·阿夫罗和梁合作。任何帮助都将不胜感激。
我已更新代码：

import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.AvroCoder;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.kafka.common.serialization.LongDeserializer;

public class Main {

public static void main(String[] args) {

    PipelineOptions options = PipelineOptionsFactory.create();

    Pipeline p = Pipeline.create(options);

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

    p.run();

}
}

####################################################### 

import org.apache.beam.sdk.coders.AvroCoder;
import org.apache.beam.sdk.coders.DefaultCoder;

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

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

但我现在得到以下错误不兼容类型：java.lang.class<io.confluent.kafka.serializers.kafkaavrodeserializer>无法转换为java.lang.class<？扩展org.apache.kafka.common.serialization.deserializer<java.lang.string>>
我必须导入不正确的序列化程序？

Java avro apache-kafka apache-beam dataflow

来源：https://stackoverflow.com/questions/54755668/how-to-deserialising-kafka-avro-messages-using-apache-beam

5条答案

按热度按时间

sdnqo3pr1#

yohei的回答很好，但我也发现这是可行的

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

...

public static class CustomKafkaAvroDeserializer extends SpecificAvroDeserializer<MyCustomClass> {}

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

哪里 MyCustomClass 是用avro工具生成的代码。

赞(0）回复(0）举报 2021-06-07

sh7euo9m2#

改变 KafkaIO.<Long, String>read() 至 KafkaIO.<Long, Object>read() .
如果您研究kafkaavrodeserializer的实现，它将实现反序列化程序： public class KafkaAvroDeserializer extends AbstractKafkaAvroDeserializer implements Deserializer<Object>

赞(0）回复(0）举报 2021-06-07

4bbkushb3#

今天我遇到了一个类似的问题，下面的例子为我解决了这个问题。
https://github.com/andrewrjones/debezium-kafka-beam-example/blob/master/src/main/java/com/andrewjones/kafkaavroconsumerexample.java
我缺少的是Kafka夫罗德塞利泽

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

赞(0）回复(0）举报 2021-06-07

ozxc1zmp4#

您可以使用kafkaavrodeserializer，如下所示：

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

其中myclass是pojo类生成的avro模式。
确保pojo类具有注解avrocoder，如下例所示：

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

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

赞(0）回复(0）举报 2021-06-06

pdkcd3nj5#

我也面临同样的问题。在邮件存档中找到了解决方案。http://mail-archives.apache.org/mod_mbox/beam-user/201710.mbox/%3ccamsy_nivrt_9_xfxotk1inhxb=x_yadbcbn+4aquu_hn0gj0na@mail.gmail.com%3e
在您的例子中，您需要定义您自己的kafkaavrodeserializer，如下所示。

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() {} }

然后将kafkaavroderializer指定为valuedeserializer。

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

赞(0）回复(0）举报 2021-06-06

我来回答

如何使用apachebeam反序列化kafka avro消息

5条答案

相关问题

热门标签

最新问答