如何在 Spark UDF 中使用 Option

Question

我有一个这样的数据集：

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

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

我按照这个答案中的第二种方法让它工作，即null在 UDF 之外处理，并写入myFn将布尔值作为第二个参数：

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

为了在 UDF 中处理 null，我看到的一种方法是根据这个答案讨论“用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给出了type mismatch; found :None.type required String。我还尝试Option在 udf 创建过程中使用参数包装，但没有成功。其基本形式（不带 Option）如下所示：

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

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

编辑

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

Answer 1

首先，可能有一些您正在使用的代码我们在这里遗漏了。当我尝试使用您的示例myFn制作成 UDFval myFnUdf = udf(myFn _)并使用 运行它时df.withColumn("new_column", myFnUdf($"code", $"status")).show，我没有得到类型不匹配，而是 a 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"
})

这给出了相同的结果。