来自 case class 的 Spark 模式具有正确的可空性

Question

对于自定义 Estimator 的 transformSchema 方法，我需要能够将输入数据框的架构与案例 class 中定义的架构进行比较。通常这可以像下面概述的那样执行。但是，使用了错误的可空性：

spark.read.csv().as[MyClass] 推断的 df 的真实模式可能如下所示：

root
 |-- CUSTOMER_ID: integer (nullable = false)

案例class：

case class MySchema(CUSTOMER_ID: Int)

为了比较我使用：

val rawSchema = ScalaReflection.schemaFor[MySchema].dataType.asInstanceOf[StructType]
  if (!rawSchema.equals(rawDf.schema))

不幸的是，这总是会产生 false，因为从案例 class 手动推断的新模式将 nullable 设置为 true（因为 ja java.Integer 实际上可能为 null )

root
 |-- CUSTOMER_ID: integer (nullable = true)

如何在创建架构时指定 nullable = false？

Answer 1

可以说您将不属于同一个 space 的东西混在一起了。 ML 管道本质上是动态的，引入静态类型的对象并没有真正改变这一点。

此外，class 的架构定义为：

case class MySchema(CUSTOMER_ID: Int)

将不可为空 CUSTOMER_ID。 scala.Int 与 java.lang.Integer 不同：

scala> import org.apache.spark.sql.catalyst.ScalaReflection.schemaFor
import org.apache.spark.sql.catalyst.ScalaReflection.schemaFor

scala> case class MySchema(CUSTOMER_ID: Int)
defined class MySchema

scala> schemaFor[MySchema].dataType
res0: org.apache.spark.sql.types.DataType = StructType(StructField(CUSTOMER_ID,IntegerType,false))

话虽如此，如果你想要 nullable 个字段 Option[Int]:

case class MySchema(CUSTOMER_ID: Option[Int])

如果你不想为 nullable 使用上面的 Int。

你在这里遇到的另一个问题是，对于 csv，每个字段根据定义都是可以为空的，并且这个状态是 "inherited" 编码的 Dataset。所以在实践中：

spark.read.csv(...)

将始终导致：

root
 |-- CUSTOMER_ID: integer (nullable = true)

这就是架构不匹配的原因。不幸的是，无法为不强制执行可空性约束的来源覆盖 nullable 字段，例如 csv 或 json。

如果没有可空模式是一个硬性要求，您可以尝试：

spark.createDataFrame(
  spark.read.csv(...).rdd,
  schemaFor[MySchema].dataType.asInstanceOf[StructType]
).as[MySchema]

只有当您知道数据实际上是 null 免费时，此方法才有效。 任何 null 值都会导致运行时异常。

来自 case class 的 Spark 模式具有正确的可空性

Spark schema from case class with correct nullability

apache-spark

apache-spark-sql

spark-csv

apache-spark-ml

apache-spark-dataset