为什么列在 Apache Spark SQL 中更改为可为空？

Question

为什么 DataFrame 中没有 NaN 值，但执行某些函数后使用 nullable = true。

val myDf = Seq((2,"A"),(2,"B"),(1,"C"))
         .toDF("foo","bar")
         .withColumn("foo", 'foo.cast("Int"))

myDf.withColumn("foo_2", when($"foo" === 2 , 1).otherwise(0)).select("foo", "foo_2").show

现在调用 df.printSchema 时，两列的 nullable 都将是 false。

val foo: (Int => String) = (t: Int) => {
    fooMap.get(t) match {
      case Some(tt) => tt
      case None => "notFound"
    }
  }

val fooMap = Map(
    1 -> "small",
    2 -> "big"
 )
val fooUDF = udf(foo)

myDf
    .withColumn("foo", fooUDF(col("foo")))
    .withColumn("foo_2", when($"foo" === 2 , 1).otherwise(0)).select("foo", "foo_2")
    .select("foo", "foo_2")
    .printSchema

但是现在，对于之前 false 的至少一列，nullable 是 true。这怎么解释？

Answer 1

从静态类型结构（不依赖于 schema 参数）创建 Dataset 时，Spark 使用一组相对简单的规则来确定 nullable 属性.

如果给定类型的对象可以是 null 那么它的 DataFrame 表示是 nullable.
如果对象是 Option[_] 那么它的 DataFrame 表示是 nullable 而 None 被认为是 SQL NULL.
在任何其他情况下，它将被标记为不是 nullable。

由于 Scala String 是 java.lang.String，可以是 null，生成的列可以是 nullable。出于同样的原因，bar 列在初始数据集中是 nullable：

val data1 = Seq[(Int, String)]((2, "A"), (2, "B"), (1, "C"))
val df1 = data1.toDF("foo", "bar")
df1.schema("bar").nullable

Boolean = true

但 foo 不是（scala.Int 不能是 null）。

df1.schema("foo").nullable

Boolean = false

如果我们将数据定义更改为：

val data2 = Seq[(Integer, String)]((2, "A"), (2, "B"), (1, "C"))

foo 将是 nullable（Integer 是 java.lang.Integer 并且盒装整数可以是 null）：

data2.toDF("foo", "bar").schema("foo").nullable

Boolean = true

另请参阅：SPARK-20668 修改 ScalaUDF 以处理可空性。

Answer 2

您也可以非常快速地更改数据框的架构。像这样的东西就可以了 -

def setNullableStateForAllColumns( df: DataFrame, columnMap: Map[String, Boolean]) : DataFrame = {
    import org.apache.spark.sql.types.{StructField, StructType}
    // get schema
    val schema = df.schema
    val newSchema = StructType(schema.map {
    case StructField( c, d, n, m) =>
      StructField( c, d, columnMap.getOrElse(c, default = n), m)
    })
    // apply new schema
    df.sqlContext.createDataFrame( df.rdd, newSchema )
}

为什么列在 Apache Spark SQL 中更改为可为空？

Why do columns change to nullable in Apache Spark SQL?

apache-spark

apache-spark-sql

apache-spark-dataset