具有正确可空性的案例类的 Spark 模式答案

【问题标题】：Spark schema from case class with correct nullability具有正确可空性的案例类的 Spark 模式
【发布时间】：2017-04-11 08:11:30
【问题描述】：

对于自定义 Estimator 的 transformSchema 方法，我需要能够将输入数据框的模式与案例类中定义的模式进行比较。通常这可以像Generate a Spark StructType / Schema from a case class 一样执行，如下所述。但是，使用了错误的可空性：

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

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

还有案例类：

case class MySchema(CUSTOMER_ID: Int)

比较我使用：

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

不幸的是，这总是产生false，因为从案例类手动推断的新架构将可空设置为true（因为 ja java.Integer 实际上可能为空）

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

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

【问题讨论】：

标签： apache-spark apache-spark-sql apache-spark-ml apache-spark-dataset spark-csv

【解决方案1】：

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

此外，类的架构定义为：

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])

如果您不想为空，请使用Int，如上所述。

您在这里遇到的另一个问题是，对于csv，每个字段根据定义都可以为空，并且此状态由编码的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 值都会导致运行时异常。

【讨论】：