Spark 2 数据集空值异常

Question

在 spark Dataset.filter 中得到这个 null 错误

输入 CSV：

name,age,stat
abc,22,m
xyz,,s

工作代码：

case class Person(name: String, age: Long, stat: String)

val peopleDS = spark.read.option("inferSchema","true")
  .option("header", "true").option("delimiter", ",")
  .csv("./people.csv").as[Person]
peopleDS.show()
peopleDS.createOrReplaceTempView("people")
spark.sql("select * from people where age > 30").show()

失败代码（添加以下行返回错误）：

val filteredDS = peopleDS.filter(_.age > 30)
filteredDS.show()

返回空错误

java.lang.RuntimeException: Null value appeared in non-nullable field:
- field (class: "scala.Long", name: "age")
- root class: "com.gcp.model.Person"
If the schema is inferred from a Scala tuple/case class, or a Java bean, please try to use scala.Option[_] or other nullable types (e.g. java.lang.Integer instead of int/scala.Int).

Answer 1

你得到的异常应该解释一切，但让我们一步一步来：

• 使用csv数据源加载数据时，所有字段都标记为nullable：

  val path: String = ???
  
  val peopleDF = spark.read
    .option("inferSchema","true")
    .option("header", "true")
    .option("delimiter", ",")
    .csv(path)
  
  peopleDF.printSchema
  

  root
  |-- name: string (nullable = true)
  |-- age: integer (nullable = true)
  |-- stat: string (nullable = true)
  

• 缺失字段表示为 SQL NULL

  peopleDF.where($"age".isNull).show
  

  +----+----+----+
  |name| age|stat|
  +----+----+----+
  | xyz|null|   s|
  +----+----+----+
  

• 接下来将Dataset[Row] 转换为Dataset[Person]，它使用Long 对age 字段进行编码。 Scala 中的Long 不能是null。因为输入模式是nullable，所以输出模式仍然是nullable：

  val peopleDS = peopleDF.as[Person]
  
  peopleDS.printSchema
  

  root
   |-- name: string (nullable = true)
   |-- age: integer (nullable = true)
   |-- stat: string (nullable = true)
  

  请注意，as[T] 根本不会影响架构。

• 当您使用 SQL（在注册表上）或 DataFrame 查询 Dataset 时，API Spark 不会反序列化对象。由于 schema 仍然是 nullable 我们可以执行：

  peopleDS.where($"age" > 30).show
  

  +----+---+----+
  |name|age|stat|
  +----+---+----+
  +----+---+----+
  

  没有任何问题。这只是一个普通的 SQL 逻辑，NULL 是一个有效值。

• 当我们使用静态类型的Dataset API 时：

  peopleDS.filter(_.age > 30)
  

  Spark 必须反序列化对象。因为 Long 不能是 null (SQL NULL)，所以它会失败，但您已经看到了异常。

  如果不是这样，您将获得 NPE。

• 正确的数据静态类型表示应使用Optional 类型：

  case class Person(name: String, age: Option[Long], stat: String)
  

  带调整过滤功能：

  peopleDS.filter(_.age.map(_ > 30).getOrElse(false))
  

  +----+---+----+
  |name|age|stat|
  +----+---+----+
  +----+---+----+
  

  如果您愿意，可以使用模式匹配：

  peopleDS.filter {
    case Some(age) => age > 30
    case _         => false     // or case None => false
  }
  

  请注意，您不必（但无论如何都会推荐）为name 和stat 使用可选类型。因为 Scala String 只是一个 Java String 它可以是 null。当然，如果您采用这种方法，您必须明确检查访问的值是否为null。

相关Spark 2.0 Dataset vs DataFrame