Как смоделировать 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 (), который, к сожалению, дает схему с nullable = true для «_id» в фальсифицированном фрейме данных, что делает тест неудачным (обратите внимание, что «фактический» вывод из функции имеет 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)
)

Мне пришлось использовать этот подход, чтобы сделать обнуляемое значение true для тех полей, где toDF делает обнуляемое значение false по умолчанию.

Как я могу создать фрейм данных с той же схемой, что и выходные данные макетируемой функции, и объявить значения, которые также могут быть классом case?

Answer 1

Из предоставленных вами журналов видно, что

Причина: java.lang.RuntimeException: models.AnimalCaseClass не является допустимым внешним типом для схемы структуры

, что означает, что вы пытаетесь вставить тип объекта AnimalCaseClass в тип данных structи это было вызвано тем, что вы использовали объект Row.

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()
}