带有Play-json验证的Apache Spark Scala

Question

java.lang.UnsupportedOperationException:不支持类型特征对象的架构

trait Container {
  def aa: String
  def bb: Int
}

case class First(aa: String, bb: Int) extends Container
case class Second(aa: String, bb: Int) extends Container

implicit val aaContainerFormat: Format[First] = Json.format[First]

implicit val bbContainerFormat: Format[Second] = Json.format[Second]

implicit def nodeContainerReads: Reads[Container] =
  try {
    Json.format[First].map(x => x: Container) or
    Json.format[Second].map(x => x: Container)
  } catch {
    case e: Exception => Reads {
      case _ => JsError(JsonValidationError("Cannot De-serialize value."))
    }
  }

implicit def nodeContainerWrites = new Writes[Container] {
  override def writes(node: Container): JsValue = node match {
    case a: First => Json.toJson(a)
    case b: Second => Json.toJson(b)
    case _ => Json.obj("error" -> "wrong Json")
  }
}

// Example Usage....
val spark: SparkSession = SparkSession.builder.appName("Unit Test").getOrCreate()
val js: Container = First("unit", "test")

spark.createDataFrame(Seq(js))

我期望容器对象的数据集的输出，但实际输出是java.lang.UnsupportedOperationException:不支持容器类型的架构。

Answer 1

Spark不使用Play JSON中的类型类将Scala类型转换为Spark SQL类型。相反，您需要查看Spark Encoders，它构成了将Scala类型转换为Spark类型的基础。如果你有Spark Session在作用域中，你可以使用import sparkSession.implicits._，这样它就会自动为你的case类创建编码器。我相信Spark不支持开箱即用的sum类型，所以你需要实现你自己的编码器，以某种方式在Spark中以一种特别的方式建模。如果您想在Spark中编码sum类型，请阅读here了解更多信息