title: Using a method from a Python class as a PySpark user-defined function
body: <p>I am trying to write a Python utility function that accepts an object of a locally-defined class, and uses one of that class's methods as a user-defined function (UDF), in a PySpark DataFrame <code>withColumn</code> call.  The utility function signature is:</p>
  
  <pre><code> def spark_analyze(lp: LogProcessor):
  </code></pre>
  
  <p>In the <code>LogProcessor</code> class, I have a method that I'd like to use as a UDF.  The method definition is:</p>
  
  <pre><code>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></pre>
  
  <p>In the <code>spark_analyze</code>, I do the following, where <code>lp</code> is the passed-in object of type <code>LogProcessor</code>:</p>
  
  <pre><code>@udf(lp.schema)
  def lpf(lcm_file):
      lp.ProcessLog(lcm_file)
  return (df.withColumn('results', lpf(col('logfile_dir')))
  ...
  </code></pre>
  
  <p>The produces a long Python stack trace, which starts like this:</p>
  
  <p>/home/david/libs.zip/pyspark_utils.py in spark_analyze(lp)
      132     def lpf(lcm_file):
      133         lp.ProcessLog(lcm_file)
  --> 134     return (df.withColumn('results', lpf(col('logfile_dir')))
      135               .withColumn('log name', spark_get_dataset_name(col('logfile_dir')))
      136               .select('log name', 'results.*')</p>
  
  <p>/usr/hdp/current/spark2-client/python/lib/pyspark.zip/pyspark/sql/functions.py in wrapper(*args)
     1955         @functools.wraps(f)
     1956         def wrapper(*args):
  -> 1957             return udf_obj(*args)
     1958 
     1959         wrapper.func = udf_obj.func</p>
  
  <p>and ends with:</p>
  
  <p>/home/david/libs.zip/pyspark_utils.py in spark_analyze(lp)
      132     def lpf(lcm_file):
      133         lp.ProcessLog(lcm_file)
  --> 134     return (df.withColumn('results', lpf(col('logfile_dir')))
      135               .withColumn('log name', spark_get_dataset_name(col('logfile_dir')))
      136               .select('log name', 'results.*')</p>
  
  <p>/usr/hdp/current/spark2-client/python/lib/pyspark.zip/pyspark/sql/functions.py in wrapper(*args)
     1955         @functools.wraps(f)
     1956         def wrapper(*args):
  -> 1957             return udf_obj(*args)
     1958 
     1959         wrapper.func = udf_obj.func</p>
  
  <p>I did some testing, and found that things work fine if I define my UDF right above the place where I pass it to <code>col</code>.  I also tried redefining the <code>ProcessLog</code> to just <code>return [0,0]</code>, and found that the problem did not go away.  So the problem seems to be that I'm using a passed-in class object's method as a UDF.  Is there another way to have a UDF be a method in a class?  Thanks for any help here!</p>