Apache Flink从Kafka读取Avro byte []

本文介绍了Apache Flink从Kafka读取Avro byte []的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

在查看示例时，我看到了很多:

In reviewing examples I see alot of this:

FlinkKafkaConsumer08<Event> kafkaConsumer = new FlinkKafkaConsumer08<>("myavrotopic", avroSchema, properties);

我看到他们在这里已经知道架构了.

I see that they here already know the schema.

在将byte []读入通用记录之前，我不知道模式 然后获取架构. (因为记录可能会有所不同)

I do not know the schema until I read the byte[] into a Generic Record then get the schema. (As it may change from record to record)

有人可以将我指向一个FlinkKafkaConsumer08，该FlinkKafkaConsumer08从byte[]读入一个映射过滤器，以便我可以删除一些前导位，然后将该byte[]加载到通用记录中吗?

Can someone point me into a FlinkKafkaConsumer08 that reads from byte[] into a map filter so that I can remove some leading bits, then load that byte[] into a Generic Record ?

推荐答案

我正在做类似的事情(我正在使用09消费者)

I'm doing something similar (I'm using the 09 consumer)

在主代码中传递自定义反序列化器:

In your main code pass in your custom deserializer:

FlinkKafkaConsumer09<Object> kafkaConsumer = new FlinkKafkaConsumer09<>( parameterTool.getRequired("topic"), new MyDeserializationSchema<>(), parameterTool.getProperties());

自定义反序列化模式读取字节，找出模式和/或从模式注册表中检索它，反序列化为GenericRecord并返回GenericRecord对象.

The custom Deserialization Schema reads the bytes, figures out the schema and/or retrieves it from a schema registry, deserializes into a GenericRecord and returns the GenericRecord object.

public class MyDeserializationSchema<T> implements DeserializationSchema<T> { private final Class<T> avrotype = (Class<T>) org.apache.avro.generic.GenericRecord.class; @Override public T deserialize(byte[] arg0) throws IOException { //do your stuff here, strip off your bytes //deserialize and create your GenericRecord return (T) (myavroevent); } @Override public boolean isEndOfStream(T nextElement) { return false; } @Override public TypeInformation<T> getProducedType() { return TypeExtractor.getForClass(avrotype); } }