Spark 2数据集空值异常

Question

在spark数据集中获取此空错误。滤器

输入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)

缺失的字段表示为SQLNULL

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

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

接下来，将Dataset[Row]转换为Dataset[Person]，其中使用Long对age字段进行编码。因为输入模式是nullable，所以输出模式仍然是nullable，尽管：

val peopleDS = peopleDF.as[Person]

peopleDS.printSchema

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

请注意，它作为[T]完全不会影响模式。

当您使用SQL（在注册表上）或DataFrameAPI查询Dataset时，Spark不会反序列化对象。由于模式仍然是可为空的，我们可以执行：

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

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

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

当我们使用静态类型的DatasetAPI时：

peopleDS.filter(_.age > 30)

Spark必须反序列化对象。因为Long不能是null（SQLnull），所以它会失败，出现异常情况。

如果不是因为这个，你会得到NPE。

数据的正确静态类型表示应使用可选类型：

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使用可选类型。因为ScalaString只是一个JavaString，所以它可以是null。当然，如果使用这种方法，则必须明确检查访问的值是否为null。

相关火花2.0数据集vs数据帧