为什么Spark DataFrames不改变他们的模式，该怎么办？

Question

我正在使用Spark 2.1的结构化流来读取Kafka主题，该主题的内容是二进制avro编码的。

因此，在设置DataFrame之后

val messages = spark
  .readStream
  .format("kafka")
  .options(kafkaConf)
  .option("subscribe", config.getString("kafka.topic"))
  .load()

如果我打印这个DataFrame (messages.printSchema())的模式，我会得到以下结果：

root
 |-- key: binary (nullable = true)
 |-- value: binary (nullable = true)
 |-- topic: string (nullable = true)
 |-- partition: integer (nullable = true)
 |-- offset: long (nullable = true)
 |-- timestamp: long (nullable = true)
 |-- timestampType: integer (nullable = true)

这个问题应该与avro解码的问题正交，但让我们假设我想通过函数Array[Byte] => BusinessObject以某种方式将value内容从消息DataFrame转换为Dataset[BusinessObject]。例如完整性，函数可能只是(使用avro4s)：

case class BusinessObject(userId: String, eventId: String)

def fromAvro(bytes: Array[Byte]): BusinessObject =
    AvroInputStream.binary[BusinessObject](
        new ByteArrayInputStream(bytes)
    ).iterator.next

当然，as miguno says in this related question我不能仅仅使用DataFrame.map()来应用转换，因为我需要为这样的BusinessObject提供一个隐式编码器。

它可以定义为：

implicit val myEncoder : Encoder[BusinessObject] = org.apache.spark.sql.Encoders.kryo[BusinessObject]

现在，执行map：

val transformedMessages : Dataset[BusinessObjecŧ] = messages.map(row => fromAvro(row.getAs[Array[Byte]]("value")))

但是，如果我查询新的模式，我会得到以下结果：

root
 |-- value: binary (nullable = true)

我认为这没有任何意义，因为数据集应该使用BusinessObject案例类的产品属性并获得正确的值。

我在读者中看过一些关于Spark SQL使用.schema(StructType)的示例，但我不能这样做，这不仅仅是因为我使用的是readStream，还因为我实际上必须先转换列，然后才能对这些字段进行操作。

我希望告诉Spark SQL engine，transformedMessages Dataset模式是一个带有case类字段的StructField。