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

Question

对于自定义 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 ，因为从案例类手动推断的新模式将 nullable 设置为 true (因为 ja java.Integer 实际上可能为空)

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

如何指定 nullable = false创建模式时？

Answer 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值将导致运行时异常。