在 Scala 中反序列化 Avro 数据时出现问题

Question

>我正在用Scala构建一个Apache Flink应用程序，它从Kafka总线读取流数据，然后对其执行汇总操作。来自 Kafka 的数据采用 Avro 格式，需要一个特殊的反序列化类。我找到了这个scala类AvroDeserializationScehema (http://codegists.com/snippet/scala/avrodeserializationschemascala_saveveltri_scala(：

package org.myorg.quickstart
import org.apache.avro.io.BinaryDecoder
import org.apache.avro.io.DatumReader
import org.apache.avro.io.DecoderFactory
import org.apache.avro.reflect.ReflectDatumReader
import org.apache.avro.specific.{SpecificDatumReader, SpecificRecordBase}
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils.TypeExtractor
import org.apache.flink.api.common.serialization._
import java.io.IOException
class AvroDeserializationSchema[T](val avroType: Class[T]) extends DeserializationSchema[T] {
private var reader: DatumReader[T] = null
private var decoder : BinaryDecoder = null
def deserialize(message: Array[Byte]): T = {
ensureInitialized()
try {
decoder = DecoderFactory.get.binaryDecoder(message, decoder)
reader.read(null.asInstanceOf[T], decoder)
}
catch {
case e: IOException => {
throw new RuntimeException(e)
}
}
}
def isEndOfStream(nextElement: T): Boolean = false

def getProducedType: TypeInformation[T] = TypeExtractor.getForClass(avroType)
private def ensureInitialized() {
if (reader == null) {
if (classOf[SpecificRecordBase].isAssignableFrom(avroType)) {
reader = new SpecificDatumReader[T](avroType)
}
else {
reader = new ReflectDatumReader[T](avroType)
}
}
}
}

在我的流课程中，我按如下方式使用它：

val stream = env
.addSource(new FlinkKafkaConsumer010[String]("test", new 
AvroDeserializationSchema[DeviceData](Class[DeviceData]), properties))

其中 DeviceData 是在同一项目中定义的 Scala 案例类

/** Case class to hold the Device data. */
case class DeviceData(deviceId: String,
sw_version: String,
timestamp: String,
reading: Double
)

我在编译 StreamingKafkaClient.scala 类时收到以下错误

Error:(24, 102) object java.lang.Class is not a value
.addSource(new FlinkKafkaConsumer010[String]("test", new 
AvroDeserializationSchema[DeviceData](Class[DeviceData]), properties))

也尝试过

val stream = env
.addSource(new FlinkKafkaConsumer010[String]("test", new 
AvroDeserializationSchema[DeviceData](classOf[DeviceData]), properties))

有了这个，我得到了一个不同的错误：

Error:(21, 20) overloaded method constructor FlinkKafkaConsumer010 with alternatives:
(x$1: java.util.regex.Pattern,x$2: org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema[String],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010[String] <and>
(x$1: java.util.regex.Pattern,x$2: org.apache.flink.api.common.serialization.DeserializationSchema[String],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010[String] <and>
(x$1: java.util.List[String],x$2: org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema[String],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010[String] <and>
(x$1: java.util.List[String],x$2: org.apache.flink.api.common.serialization.DeserializationSchema[String],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010[String] <and>
(x$1: String,x$2: org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema[String],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010[String] <and>
(x$1: String,x$2: org.apache.flink.api.common.serialization.DeserializationSchema[String],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010[String]
cannot be applied to (String, org.myorg.quickstart.AvroDeserializationSchema[org.myorg.quickstart.DeviceData], java.util.Properties)
.addSource(new FlinkKafkaConsumer010[String]("test", new AvroDeserializationSchema[DeviceData](classOf[DeviceData]), properties))

我是 Scala 的新手(这是我的第一个 scala 程序(，所以我知道我在这里缺少一些基本的东西。当我尝试学习Scala时，有人可以指出我做错了什么。我的目的是基本上将 avro 编码的数据从 Kafka 读取到 Flink 中，并对流数据进行一些操作。我找不到任何使用 AvroDeserializationSchema 类的示例，在我看来，这是应该原生构建到 Flink 包中的东西。

Answer 1

为了在 Scala 中获取类对象，你需要classOf[DeviceData]，而不是Class[DeviceData]

new AvroDeserializationSchema[DeviceData](classOf[DeviceData])

我找不到任何使用 AvroDeserializationSchema 类的示例

我找到了一个(在Java中(

此外，看起来在 Flink 1.6 版本中，他们将添加这个类，而不是你从其他地方复制。FLINK-9337 和FLINK-9338

如注释中所述，如果您想使用 Confluent Avro Schema Registry 而不是给出类类型，请参阅此答案，或参考上述 Github 链接中的代码

此外，如果您正在运行Kafka 0.11+(或Confluent 3.3+(，那么理想情况下，您应该将FlinkKafkaConsumer011与要反序列化为的类一起使用

。

new FlinkKafkaConsumer011[DeviceData]