使用来自 Python 类的方法作为 PySpark 用户定义函数

Question

我正在尝试编写一个Python实用函数，它接受本地定义类的对象，并在PySpark DataFrame＜code＞withColumn＜code＞调用中使用该类的一个方法作为用户定义函数（UDF）。实用功能签名为：

 def spark_analyze(lp: LogProcessor):

在LogProcess类中，我有一个想用作UDF的方法。方法定义是：

schema = StructType([
  StructField("total", IntegerType(), False),
  StructField("other", IntegerType(), False)
])

def ProcessLog(self, log_file):
    self.PrepareForLog()
    for event in pyspark_utils.spark_events_from_file(log_file):
      self.ProcessEvent(event)
      return [total, other]

在< code>spark_analyze中，我执行以下操作，其中< code>lp是传入的< code>LogProcessor类型的对象:

@udf(lp.schema)
def lpf(lcm_file):
    lp.ProcessLog(lcm_file)
return (df.withColumn('results', lpf(col('logfile_dir')))
...

生成一个很长的 Python 堆栈跟踪，其开始如下所示：

spark_analyze(lp)中的/home/David/libs . zip/py spark _ utils . py 132 def LPF(LCM _ file):133 LP。进程日志(lcm_file)

/usr/HDP/current/spark 2-client/python/lib/py spark . zip/py spark/SQL/functions . py in wrapper(* args)1955 @ functools . wrappers(f)1956 def wrapper(* args):-

并以以下内容结尾：

spark_analyze(lp)中的/home/David/libs . zip/py spark _ utils . py 132 def LPF(LCM _ file):133 LP。进程日志(lcm_file)

/usr/HDP/current/spark 2-client/python/lib/py spark . zip/py spark/SQL/functions . py in wrapper(* args)1955 @ functools . wrappers(f)1956 def wrapper(* args):-

我做了一些测试，发现如果我在传递给col的位置上方定义了我的UDF，事情就会很好。我还尝试将＜code＞ProcessLog，发现问题并没有消失。因此，问题似乎是我使用传入类对象的方法作为UDF。还有其他方法可以让UDF成为类中的方法吗？谢谢你的帮助！

Answer 1

奥斯曼·阿兹哈尔建议的方法可能行得通。我通过简单地将UDF的定义作为库函数的参数来解决这个问题。