Scala Spark - 不支持 udf 列

Posted 2023-04-13

【中文标题】Scala Spark - 不支持 udf 列

【英文标题】：Scala Spark - udf Column is not supported

【发布时间】：2017-11-19 15:15:41

【问题描述】：

我正在尝试使用udf，它相当于：

df.select(when(col("abc").isNotNull and col("abc") =!= "" and col("age") <= 18, 1).otherwise(0).alias("something"))

我将udf 声明为：

//return Int 0 or 1 if conditions are true 
val myudf_x = udf((col_name: String, col_value: String, num: Int) => 
  when(col_name.isNotNull and col_name =!= "" and col_value < num, 1).otherwise(0)

)

用法：

  df.select(
  "col_abc",
  myudf(col("col_abc"), col("age"), 18).alias("something")
)

但我得到一个错误：

不支持 org.apache.spark.sql.Column 类型的架构

我也尝试过使用 String 类型而不是 column 类型的 udf

有什么问题？

谢谢

【参考方案1】：

一个简单的区别：

表达式对 SQL 类型 (Columns) 进行操作。 udfs 对外部 (Scala) 类型进行操作。

如果你想要一个使用表达式 DSL 的函数：

import org.apache.spark.sql.Column

// You can use function:
// def f(col_name: Column, col_value: Column, num: Column) = ???
// I used closure syntax to highlight difference in types
val f: (Column, Column, Column) => Column = 
  (col_name: Column, col_value: Column, num: Column) =>  when(
    col_name.isNotNull and col_name =!= "unknown" and col_value < num, 
    1
  ).otherwise(0)

否则：

val g: UserDefinedFunction = udf(
  (col_name: String, col_value: String, num: Int) => 
    if (col_name != null && col_name != "unknown" && col_value < num) 1 else 0
  
)

但在当前形式下，udf 不会进行类型检查（col_value 是 String 和 num 是 Int - 它们无法与 < 进行比较）。

也许你想要col_value.cast("int") < num / col_value.toInt < num？