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

Question

对于自定义Estimator的transformSchema方法,我需要能够将输入数据帧的模式与案例类中定义的模式进行比较.通常,这可以像从案例类生成Spark StructType/Schema一样执行,如下所述.但是,使用了错误的可空性:

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

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,因为从case类手动推断的新模式设置为可为空true(因为ja java.Integer实际上可能为null)

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

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

Answer 1

可以说你混合的东西并不属于同一个空间.ML管道本质上是动态的,引入静态类型的对象并没有真正改变它.

此外,类的模式定义为:

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值都会导致运行时异常.