Spark：在不更改列的可为空属性的情况下转换小数答案

【问题标题】：Spark: cast decimal without changing nullable property of columnSpark：在不更改列的可为空属性的情况下转换小数
【发布时间】：2018-11-24 01:46:04
【问题描述】：

在DataFrame 中将列转换为DecimalType 似乎会更改可为空的属性。具体来说，我有一个类型为DecimalType(12, 4) 的不可为空的列，我使用df.withColumn(columnName, df.col(columnName).cast(dataType)) 将其转换为DecimalType(38, 9)。这会产生一个具有预期数据类型的字段，但该字段现在可以为空。有没有办法在不改变列的可为空属性的情况下进行转换？

我在 Spark 2.2.1 和 Spark 2.3.0 中都观察到了这种行为。

【问题讨论】：

标签： apache-spark apache-spark-sql

【解决方案1】：

感谢您提出有趣的观点。我对源代码进行了一些研究以了解这种行为，IMO 的答案是在 Cast.scala 中表示强制转换表达式。暴露可空性的属性是这样计算的：

override def nullable: Boolean = Cast.forceNullable(child.dataType, dataType) || child.nullable

  def forceNullable(from: DataType, to: DataType): Boolean = (from, to) match {
  case (NullType, _) => true
  case (_, _) if from == to => false

  case (StringType, BinaryType) => false
  case (StringType, _) => true
  case (_, StringType) => false

  case (FloatType | DoubleType, TimestampType) => true
  case (TimestampType, DateType) => false
  case (_, DateType) => true
  case (DateType, TimestampType) => false
  case (DateType, _) => true
  case (_, CalendarIntervalType) => true

  case (_, _: DecimalType) => true  // overflow
  case (_: FractionalType, _: IntegralType) => true  // NaN, infinity
  case _ => false
}

如您所见，从任何类型到DecimalType 的转换总是返回一个可为空的类型。我想知道为什么，这可能是因为这里表达的溢出风险：

/**
 * Change the precision / scale in a given decimal to those set in `decimalType` (i  f any),
 * returning null if it overflows or modifying `value` in-place and returning it if successful.
 *
 * NOTE: this modifies `value` in-place, so don't call it on external data.
 */
private[this] def changePrecision(value: Decimal, decimalType: DecimalType): Decimal = {
  if (value.changePrecision(decimalType.precision,   decimalType.scale)) value else null
}

changePrecision 方法依次检查是否可以修改精度，如果是则返回 true，否则返回 false。它解释了为什么上述方法可以返回 null 以及为什么 DecimalType 在源类型上独立强制转换时默认设置为可为空。

由于 IMO 没有简单的方法来保持原始列的可空性。也许您可以尝试查看 UserDefinedTypes 并构建自己的源属性保持 DecimalType ？但 IMO 的可空性并非没有原因，我们尊重这一点，以避免迟早在管道中出现一些不好的意外。

【讨论】：

【解决方案2】：

df.withColumn(columnName, df.col(columnName).cast(dataType))可以改写为：

import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull

df.withColumn(columnName, new Column(AssertNotNull(df.col(columnName).cast(dataType).expr)))

注意：如果结果是强制转换导致 null 值，这将引发 NullPointerException。

来源：https://dev.to/kevinwallimann/how-to-make-a-column-non-nullable-in-spark-structured-streaming-4b62

【讨论】：