scala - SparkSQL : How to deal with null values in user defined function?

Question

给定表 1，其中有一列“x”，类型为字符串。我想创建表 2，其中“y”列是“x”中给出的日期字符串的整数表示形式。

基本是在“y”列中保留null值。

表 1(数据帧 df1):

+----------+
|         x|
+----------+
|2015-09-12|
|2015-09-13|
|      null|
|      null|
+----------+
root
 |-- x: string (nullable = true)

表 2(数据帧 df2):

+----------+--------+                                                                  
|         x|       y|
+----------+--------+
|      null|    null|
|      null|    null|
|2015-09-12|20150912|
|2015-09-13|20150913|
+----------+--------+
root
 |-- x: string (nullable = true)
 |-- y: integer (nullable = true)

将“x”列的值转换为“y”列的用户定义函数(udf)是:

val extractDateAsInt = udf[Int, String] (
  (d:String) => d.substring(0, 10)
      .filterNot( "-".toSet)
      .toInt )

并且可以工作，处理空值是不可能的。

尽管如此，我可以做类似的事情

val extractDateAsIntWithNull = udf[Int, String] (
  (d:String) => 
    if (d != null) d.substring(0, 10).filterNot( "-".toSet).toInt 
    else 1 )

我没有找到通过 udfs“生成”null 值的方法(当然，因为 Int 不能为 null) 。

我当前创建 df2(表 2)的解决方案如下:

// holds data of table 1  
val df1 = ... 

// filter entries from df1, that are not null
val dfNotNulls = df1.filter(df1("x")
  .isNotNull)
  .withColumn("y", extractDateAsInt(df1("x")))
  .withColumnRenamed("x", "right_x")

// create df2 via a left join on df1 and dfNotNull having 
val df2 = df1.join( dfNotNulls, df1("x") === dfNotNulls("right_x"), "leftouter" ).drop("right_x")

问题:

• 当前的解决方案似乎很麻烦(并且可能在性能方面效率不高)。有更好的办法吗？
• @Spark-developers:是否有计划/可用的类型 NullableInt ，以便可以使用以下 udf(请参阅代码摘录)？

代码摘录

val extractDateAsNullableInt = udf[NullableInt, String] (
  (d:String) => 
    if (d != null) d.substring(0, 10).filterNot( "-".toSet).toInt 
    else null )

Answer 1

这就是Option派上用场的地方:

val extractDateAsOptionInt = udf((d: String) => d match {
  case null => None
  case s => Some(s.substring(0, 10).filterNot("-".toSet).toInt)
})

或者在一般情况下使其更加安全:

import scala.util.Try

val extractDateAsOptionInt = udf((d: String) => Try(
  d.substring(0, 10).filterNot("-".toSet).toInt
).toOption)

所有功劳归Dmitriy Selivanov谁指出此解决方案是(缺失？)编辑 here .

替代方法是在 UDF 外部处理 null:

import org.apache.spark.sql.functions.{lit, when}
import org.apache.spark.sql.types.IntegerType

val extractDateAsInt = udf(
   (d: String) => d.substring(0, 10).filterNot("-".toSet).toInt
)

df.withColumn("y",
  when($"x".isNull, lit(null))
    .otherwise(extractDateAsInt($"x"))
    .cast(IntegerType)
)