如何将常量值传递给 Python UDF？

Posted 2023-04-15

【中文标题】如何将常量值传递给 Python UDF？

【英文标题】：How to pass a constant value to Python UDF?

【发布时间】：2016-02-13 02:06:49

【问题描述】：

我在想是否可以创建一个UDF，它接收两个参数Column 和另一个变量（Object、Dictionary 或任何其他类型），然后执行一些操作并返回结果.

实际上，我试图这样做，但我遇到了一个例外。因此，我想知道是否有任何方法可以避免这个问题。

df = sqlContext.createDataFrame([("Bonsanto", 20, 2000.00), 
                                 ("Hayek", 60, 3000.00), 
                                 ("Mises", 60, 1000.0)], 
                                ["name", "age", "balance"])

comparatorUDF = udf(lambda c, n: c == n, BooleanType())

df.where(comparatorUDF(col("name"), "Bonsanto")).show()

我收到以下错误：

AnalysisException: u"cannot resolve 'Bonsanto' given input columns 姓名、年龄、余额；"

所以很明显UDF“看到”string“Bonsanto”作为列名，实际上我正在尝试将记录值与第二个参数进行比较。

另一方面，我知道可以在where 子句中使用一些运算符（但实际上我想知道使用UDF 是否可以实现），如下所示：

df.where(col("name") == "Bonsanto").show()

#+--------+---+-------+
#|    name|age|balance|
#+--------+---+-------+
#|Bonsanto| 20| 2000.0|
#+--------+---+-------+

【参考方案1】：

传递给 UDF 的所有内容都被解释为列/列名。如果你想传递一个文字，你有两个选择：

使用柯里化传递参数：

def comparatorUDF(n):
    return udf(lambda c: c == n, BooleanType())

df.where(comparatorUDF("Bonsanto")(col("name")))

这可以与任何类型的参数一起使用，只要它是可序列化的。

使用 SQL 文字和当前实现：

from pyspark.sql.functions import lit

df.where(comparatorUDF(col("name"), lit("Bonsanto")))

这仅适用于支持的类型（字符串、数字、布尔值）。对于非原子类型，请参阅How to add a constant column in a Spark DataFrame?