Flink AvroDeserializationSchema for specificRecord -- 使用泛型类类型答案

【问题标题】：Flink AvroDeserializationSchema for SpecificRecord -- Usage with Generic Class TypeFlink AvroDeserializationSchema for specificRecord -- 使用泛型类类型
【发布时间】：2020-09-08 13:57:38
【问题描述】：

目标：我想实现一个可以重用的函数，以便根据作为类型参数传递的生成的类为特定记录创建 AvroDeserializationSchema。生成的类使用 AvroHugger 生成并继承 SpecificRecordBase。

问题：无法找到设置泛型类型参数以使编译工作的方法。 AvroDeserializationSchema.forSpecific 需要获取一个类作为生成的参数。

函数如下：

  ConfluentRegistryAvroDeserializationSchema[T] = {
    ConfluentRegistryAvroDeserializationSchema.forSpecific(
      classOf[T],
      schemaRegistryUrl
    )
  }

用法如下：

val registry = AvroSerde.createEventDeserializerForRegistry[SomeGeneratedClass]("localhost")

这是我在运行代码时遇到的错误：

class type required but T found classOf[T]

我知道 classOf 不适用于任意类型（并且 T 是任意类型），但我不知道如何在此处传递 Class 以便可以重用此函数来传递任意生成的类。

所以如果我尝试传入 T 的运行时类，它也不起作用：

ConfluentRegistryAvroDeserializationSchema.forSpecific(
      classTag[T].runtimeClass,
      schemaRegistryUrl
    )

错误：

cannot be applied to (Class[_$1], String) ConfluentRegistryAvroDeserializationSchema.forSpecific(

如果有人知道如何实现这个抽象，我将不胜感激。

谢谢。

【问题讨论】：

标签： apache-flink avro

【解决方案1】：

我不是世界上最大的 scala 人，但我假设您需要为您的类型参数指定下限（这实际上是一个 scala 问题）。

你可以在Java签名中看到T必须扩展SpecificRecord

    public static <T extends SpecificRecord> ConfluentRegistryAvroDeserializationSchema<T> forSpecific(Class<T> tClass, String url)

所以你的函数也应该指定相同的较低类型

  ConfluentRegistryAvroDeserializationSchema[T >: SpecificRecord] = {
    ConfluentRegistryAvroDeserializationSchema.forSpecific(
      classOf[T],
      schemaRegistryUrl
    )
  }

此外，您可能还想检查 scala 如何具体化泛型类型，例如，here。

【讨论】：