如何用嵌套的案例类模式模拟Spark Scala DataFrame？

Question

我如何创建/模拟Spark Scala数据帧与一个案例类嵌套在顶层？

root
 |-- _id: long (nullable = true)
 |-- continent: string (nullable = true)
 |-- animalCaseClass: struct (nullable = true)
 |    |-- name: string (nullable = true)
 |    |-- gender: string (nullable = true)

我目前正在单元测试一个在上述模式中输出数据帧的函数。为了检查相等性，我使用了toDF（），不幸的是，它给出了一个在模拟数据帧中的“_id”为nullable=true的模式，从而使测试失败（注意，函数的“实际”输出对所有内容都为nullable=true）。

我还尝试以不同的方式创建模拟数据帧，这导致了错误：https://pastebin.com/WtxtgMJA

以下是我在这种方法中尝试的内容：

import org.apache.spark.sql.Encoders
val animalSchema = Encoders.product[AnimalCaseClass].schema

val schema = List(
  StructField("_id", LongType, true),
  StructField("continent", StringType, true),
  StructField("animalCaseClass", animalSchema, true)
)

val data = Seq(Row(12345L, "Asia", AnimalCaseClass("tiger", "male")), Row(12346L, "Asia", AnimalCaseClass("tigress", "female")))

val expected = spark.createDataFrame(
  spark.sparkContext.parallelize(data),
  StructType(schema)
)

对于那些toDF默认设置为false的字段，我必须使用这种方法使其为null。

如何使用与模拟函数的输出相同的模式创建数据帧，并声明也可以是case类的值？

Answer 1

从你提供的日志中，你可以看到

原因：java。RuntimeException:模型。AnimalCaseClass不是结构的架构的有效外部类型

这意味着您正试图将AnimalCaseClass的对象类型插入到struct的数据类型中

import org.apache.spark.SparkConf
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{LongType, StringType, StructField, StructType}
import org.apache.spark.sql.SparkSession

case class AnimalCaseClass(name: String, gender: String)

object Test extends App {

  val conf: SparkConf = new SparkConf()
  conf.setAppName("Test")
  conf.setMaster("local[2]")
  conf.set("spark.sql.test", "")
  conf.set(SQLConf.CODEGEN_FALLBACK.key, "false")

  val spark: SparkSession = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate()

  // ** The relevant part **
  import org.apache.spark.sql.Encoders
  val animalSchema = Encoders.product[AnimalCaseClass].schema

  val expectedSchema: StructType = StructType(Seq(
    StructField("_id", LongType, true),
    StructField("continent", StringType, true),
    StructField("animalCaseClass", animalSchema, true)
  ))

  import spark.implicits._
  val data = Seq((12345L, "Asia", AnimalCaseClass("tiger", "male")), (12346L, "Asia", AnimalCaseClass("tigress", "female"))).toDF()

  val expected = spark.createDataFrame(data.rdd, expectedSchema)

  expected.printSchema()

  expected.show()

  spark.stop()
}