scala - 如何将 Option 与 Spark UDF 结合使用

Question

我有一个像这样的数据集:

+----+------+
|code|status|
+-----------+
|   1| "new"|
|   2|  null|
|   3|  null|
+----+------+

我想编写一个依赖于这两列的 UDF。

我按照this answer中的第二种方法让它工作了也就是在UDF之外处理null，并编写myFn以 bool 值作为第二个参数:

df.withColumn("new_column",
  when(pst_regs("status").isNull, 
    myFnUdf($"code", lit(false))
  )
  .otherwise(
    myFnUdf($"code", lit(true))
  )
)

为了处理 UDF 中的 null，我查看的方法是 this answer其中讨论了“用 Options 包装参数”。我尝试过这样的代码:

df.withColumn("new_column", myFnUdf($"code", $"status"))

def myFn(code: Int, status: String) = (code, Option(status)) match {
  case (1, "new") => "1_with_new_status"
  case (2, Some(_)) => "2_with_any_status"
  case (3, None) => "3_no_status"
}

但是带有 null 的行会导致类型不匹配；发现:None.type required String。我还尝试在 udf 创建期间使用 Option 包装参数，但没有成功。其基本形式(不带选项)如下所示:

myFnUdf = udf[String, Int, String](myFn(_:Int, _:String))

我是 Scala 新手，所以我确信我错过了一些简单的东西。我的部分困惑可能是从函数创建 udf 的语法不同(例如每个 https://jaceklaskowski.gitbooks.io/mastering-apache-spark/content/spark-sql-udfs.html )，所以我不确定我是否使用了最好的方法。任何帮助表示赞赏!

编辑

已编辑以根据 @user6910411 和 @sgvd 评论添加缺少的 (1, "new") 案例。

Answer 1

首先，您正在使用的一些代码可能是我们在这里丢失的。当我尝试您的示例 myFn 时，使用 val myFnUdf = udf(myFn _) 制作成 UDF 并使用 df.withColumn("new_column", myFnUdf 运行它($"code", $"status")).show，我没有得到类型不匹配，而是得到 MatchError，正如 user6910411 也指出的那样。这是因为没有匹配 (1, "new") 的模式。

除此之外，虽然通常使用 Scala 的选项比原始 null 值更好，但在本例中您不必这样做。以下示例直接使用 null:

val my_udf = udf((code: Int, status: String) => status match {
    case null => "no status"
    case _ => "with status"
})

df.withColumn("new_column", my_udf($"code", $"status")).show

结果:

+----+------+-----------+
|code|status| new_column|
+----+------+-----------+
|   1|   new|with status|
|   2|  null|  no status|
|   2|  null|  no status|
+----+------+-----------+

用选项包装仍然有效:

val my_udf = udf((code: Int, status: String) => Option(status) match {
    case None => "no status"
    case Some(_) => "with status"
})

这给出了相同的结果。