Spark结构化流中的Avro格式反序列化

Question

我正在使用 Spark Structured Streaming，如上所述 this page.

我从 Kafka 主题收到正确的消息，但值为 Avro 格式。有什么方法可以反序列化 Avro 记录（类似于 KafkaAvroDeserializer 方法）？

Answer 1

火花 >= 2.4

您可以使用spark-avro 库中的from_avro 函数。

import org.apache.spark.sql.avro._

val schema: String = ???
df.withColumn("value", from_avro($"value", schema))

火花

• 定义一个接受Array[Byte]（序列化对象）的函数：

  import scala.reflect.runtime.universe.TypeTag
  
  def decode[T : TypeTag](bytes: Array[Byte]): T = ???
  

  这将反序列化 Avro 数据并创建可以存储在 Dataset 中的对象。

• 根据函数创建udf。

  val decodeUdf  = udf(decode _)
  

• 拨打udfvalue

  val df = spark
    .readStream
    .format("kafka")
    ...
    .load()
  
  df.withColumn("value", decodeUdf($"value"))