Spark UDF 空值处理

Question

我正在努力处理 UDF 中的空值，该 UDF 在由浮点数结构组成的数据帧（源自配置单元 table）上运行：

数据框 (points) 具有以下架构：

root
 |-- point: struct (nullable = true)
 |    |-- x: float (nullable = true)
 |    |-- y: float (nullable = true)

例如，我想计算x和y的总和。请注意，我在以下示例中没有 "handle" 空值，但我希望能够检查我的 udf point、x 或 y 是否为 null.

第一种方法：

val sum = udf((x:Float,y:Float) => x+y)

points.withColumn("sum",sum($"point.x",$"point.y"))

如果 struct 点是 null，这将不起作用，在这种情况下，永远不会评估 udf（永远不会执行 udf 中的代码！），结果为 null。此外，我无法检查 x 或 y 是否为 null，因为 Floats 在 scala 中不能为 null。

第二种方法：

val sum = udf((pt:Row) => pt.getFloat(0)+pt.getFloat(1))
points.withColumn("sum",sum($"point"))

这种方法，我可以在我的 udf 中检查 pt 是否为 null，但我无法检查 x 和 y，因为 Floats 不能无效的。在这种情况下，我得到一个 NullPointerException。

我如何编写一个 udf win 来检查结构以及 x 和 y 是否为空？

我正在使用 spark 1.6.1

更新：与相比，我处理的是浮点数而不是字符串（字符串在 Scala 中可以为空，浮点数不能）

Answer 1

您可以使用 Row.isNullAt(i) 检查第 i 字段是否为空。在你的情况下，你应该把你的 udf 写成，

sum = udf((point: Row) => point.match {
  case p if (p.isNullAt(0) && p.isNullAt(0)) => 0f
  case p if p.isNullAt(0) => p.getFloat(1)
  case p if p.isNullAt(1) => p.getFloat(0)
  case p => p.getFloat(0) + p.getFloat(1)
})

Spark UDF 空值处理

Spark UDF Null handling

scala

apache-spark

udf