我在java中找到了avrodeserializationschema类的示例
https://github.com/okkam-it/flink-examples/blob/master/src/main/java/org/okkam/flink/avro/avrodeserializationschema.java
代码段：
如果您想反序列化到特定的case类中，那么使用 new FlinkKafkaConsumer011[case_class_name] , new AvroDeserializationSchema[case_class_name](classOf[case_class_name] ```
val stream = env .addSource(new FlinkKafkaConsumer011[DeviceData]
("test", new AvroDeserializationSchemacase_class_name, properties))

如果您使用confluent的模式注册表，那么首选的解决方案是使用confluent提供的avro serde。我们只需调用deserialize（），最新版本的avro模式的解析将在后台自动完成，不需要进行字节操作。
类似于下面的scala。

import io.confluent.kafka.serializers.KafkaAvroDeserializer

...

val valueDeserializer = new KafkaAvroDeserializer()
valueDeserializer.configure(
Map(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG -> schemaRegistryUrl).asJava,
false)

...

override def deserialize(messageKey: Array[Byte], message: Array[Byte],
topic: String, partition: Int, offset: Long): KafkaKV = {

val key = keyDeserializer.deserialize(topic, messageKey).asInstanceOf[GenericRecord]
val value = valueDeserializer.deserialize(topic, message).asInstanceOf[GenericRecord]

KafkaKV(key, value)
}

...

详细说明如下：http://svend.kelesia.com/how-to-integrate-flink-with-confluents-schema-registry.html#how-将flink与合流模式注册表集成
希望有帮助！