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

Question

我在想,如果有可能创建一个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"无法解析'Bonsanto'给定的输入列名称,年龄,余额;"

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

另一方面,我知道可以在where子句中使用一些运算符(但实际上我想知道它是否可以使用UDF),如下所示:

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

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

Answer 1

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

1. 使用currying传递参数:

   def comparatorUDF(n):
       return udf(lambda c: c == n, BooleanType())
   
   df.where(comparatorUDF("Bonsanto")(col("name")))
   

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

2. 使用SQL文字和当前实现:

   from pyspark.sql.functions import lit
   
   df.where(comparatorUDF(col("name"), lit("Bonsanto")))
   

   这仅适用于受支持的类型(字符串,数字,布尔值).对于非原子类型,请参阅如何在Spark DataFrame中添加常量列？