title: Using numpy inside pandas udf pyspark
body: <p>I am trying to define a pandas udf that would compute the skew of a lognormal distribution per period.</p>
  
  <p>I currently have done the following:</p>
  
  <pre><code>@pandas_udf("double", PandasUDFType.GROUPED_AGG)  
  def lognormal_skew(v):
    return (np.exp(v.std()) + 2) * np.sqrt(np.exp(v.std()) - 1)
  
  my_df.groupBy('period').agg(lognormal_skew(my_df['my_columns'])).show()
  </code></pre>
  
  <p>However I get an error:</p>
  
  <pre><code>rg.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 3047.0 failed 4 times, most recent failure: Lost task 0.3 in stage 3047.0 (TID 208812, 10.139.64.8, executor 82): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  </code></pre>
  
  <p>My guess is that this has to do something with <code>numpy</code> since if I try to define a skew as follows:</p>
  
  <pre><code>@pandas_udf("double", PandasUDFType.GROUPED_AGG)  
  def skew(v):
    return v.skew()
  
  my_df.groupBy('period').agg(skew(my_df['my_columns'])).show()
  </code></pre>
  
  <p>It outputs a <code>DataFrame</code> and it does not error.</p>