在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类里还嵌套了个类,使用起来也能正常解析,没发现问题

 

Logo

Kafka开源项目指南提供详尽教程,助开发者掌握其架构、配置和使用,实现高效数据流管理和实时处理。它高性能、可扩展,适合日志收集和实时数据处理,通过持久化保障数据安全,是企业大数据生态系统的核心。

更多推荐