python - 使用 Python 类中的方法作为 PySpark 用户定义函数

Question

我正在尝试编写一个 Python 实用函数，它接受本地定义类的对象，并在 PySpark DataFrame withColumn 中使用该类的方法之一作为用户定义函数 (UDF)调用。实用函数签名为:

 def spark_analyze(lp: LogProcessor):

在 LogProcessor 类中，我有一个想要用作 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]

在 spark_analyze 中，我执行以下操作，其中 lp 是 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/pyspark_utils.py 132 def lpf(lcm_文件): 133 lp.ProcessLog(lcm_文件)--> 134 return (df.withColumn('结果', lpf(col('logfile_dir'))) 135 .withColumn('日志名称',spark_get_dataset_name(col('logfile_dir'))) 136 .select('日志名称', '结果.*')

包装器中的/usr/hdp/current/spark2-client/python/lib/pyspark.zip/pyspark/sql/functions.py(*args) 第1955章 1956 def 包装器(*args):-> 1957 返回 udf_obj(*args) 1958年 第 1959 章

并以以下内容结尾:

spark_analyze(lp) 中的/home/david/libs.zip/pyspark_utils.py 132 def lpf(lcm_文件): 133 lp.ProcessLog(lcm_文件)--> 134 return (df.withColumn('结果', lpf(col('logfile_dir'))) 135 .withColumn('日志名称',spark_get_dataset_name(col('logfile_dir'))) 136 .select('日志名称', '结果.*')

包装器中的/usr/hdp/current/spark2-client/python/lib/pyspark.zip/pyspark/sql/functions.py(*args) 第1955章 1956 def 包装器(*args):-> 1957 返回 udf_obj(*args) 1958年 第 1959 章

我做了一些测试，发现如果我将 UDF 定义在将其传递给 col 的位置的正上方，那么事情就可以正常工作。我还尝试将 ProcessLog 重新定义为仅返回 [0,0] ，发现问题并没有消失。所以问题似乎是我使用传入的类对象的方法作为 UDF。还有另一种方法可以让 UDF 成为类中的方法吗？感谢您的帮助!

Answer 1

乌斯曼·阿扎尔建议的方法可能有效。我最终通过简单地将 UDF 的定义作为库函数的参数传递来解决了这个问题。