Flink接收Kafka中的Avro序列化消息
在Flink官网中可以看到flink接收kafka数据的简单示例程序Properties properties = new Properties();properties.setProperty("bootstrap.servers", "localhost:9092");// only required for Kafka 0.8properties.setProperty("zoo...
在Flink官网中可以看到flink接收kafka数据的简单示例程序
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
// only required for Kafka 0.8
properties.setProperty("zookeeper.connect", "localhost:2181");
properties.setProperty("group.id", "test");
DataStream<String> stream = env
.addSource(new FlinkKafkaConsumer08<>("topic", new SimpleStringSchema(), properties));
这里传给FlinkKafkaConsumer08构造方法的第二个参数new SimpleStringSchema()即指定了从kafka获取消息后反序列化的方式为字符串序列化方式,当我们的kafka中消息是采用Avro实现序列化后,我们反序列化也要指定为采用Avro的方式,这里flink提供了一个叫做AvroDeserializationSchema的类,它可以实现Avro反序列化,实现方式示例如下:
首先创建这个Avro反序列化对象,通过forSpecific(User.class)方法指定需要序列化的实例类class为User.class(我的数据对象是User类实例)
AvroDeserializationSchema<User> UserSchema = AvroDeserializationSchema.forSpecific(User.class);
其中User类的实例就是我在kafka中传输的数据,这样添加source时就可以使用这个序列化对象了
DataStream<String> stream = env
.addSource(new FlinkKafkaConsumer08<>("topic", UserSchema, properties));
另外,Flink官网上说这个Avro序列化类不期望使用嵌套的数据结构(嵌套的schema),原话:This deserialization schema expects that the serialized records DO NOT contain embedded schema.
但是我的User类里还嵌套了个类,使用起来也能正常解析,没发现问题
更多推荐
所有评论(0)