如何使用从Flink序列化到AVRO的Kafka事件？

Question

我是Scala和Apache Flink的初学者，但到目前为止，一切都很顺利。我正在尝试使用Flink应用程序中序列化到AVRO的Kafka事件。我阅读了文档(https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html#the-反序列化模式）和google搜索了很多小时，但我仍然在同一页上。我有一个case类urresponse（status:int，domain:String，url:String，queue:String，html:String）和一个schemaval schema:schema=new schema。解析器（）。解析（“{”type\“：\”record\“，\”name\“：\”urresponse\“，\”fields\“：[{”name\“：\”status\“，\”type\“：\”long\“，{”name\“：”domain\“，\”type\“：”string\“，{”name\“：”url\“，”type\”string\“，\”name\“：”queue\，”type\“：”string\“，{”name\”html“，”type“：\”string“}]}”）。我尝试了3种方法：

1. val stream=env.addSource（new Flink Kafka消费者（kafkaTopic， AvroDeserializationSchema.for通用（模式），属性））运行时出错：

Serialization trace:
reserved (org.apache.avro.Schema$Field)
fieldMap (org.apache.avro.Schema$RecordSchema)
schema (org.apache.avro.generic.GenericData$Record)

inferred type arguments [schemas.URLResponse] do not conform to method forSpecific's type parameter bounds [T <: org.apache.avro.specific.SpecificRecord]
      kafkaTopic, AvroDeserializationSchema.forSpecific(classOf[URLResponse]), properties))

constructor AvroDeserializationSchema in class AvroDeserializationSchema cannot be accessed in object MyApp
    val des: AvroDeserializationSchema[URLResponse] = new AvroDeserializationSchema[URLResponse](classOf[URLResponse])

求求你，救命！首选的方法是什么？为什么不起作用？谢谢

Answer 1

似乎推荐第一种方法。提到的异常与avro反序列化scala实现的问题有关。如果我使用java实现（https://flink.apache.org/news/2020/04/15/flink-serialization-tuning-vol-1.html#avro），它会很好。我的解决方案：

    val javaStream = env.getJavaEnv.addSource(new FlinkKafkaConsumer[GenericRecord](
      kafkaTopic, ConfluentRegistryAvroDeserializationSchema.forGeneric(schema, schemaRegistryURL), properties),
      new GenericRecordAvroTypeInfo(schema))
    val stream = new DataStream[GenericRecord](javaStream)