Pyspark，如何使用 udf 计算泊松分布?

问题描述

我有一个如下所示的数据框:

I have a dataframe looks like this:

df_schema = StructType([StructField("date", StringType(), True),\
                              StructField("col1", FloatType(), True),\
                             StructField("col2", FloatType(), True)])
df_data = [('2020-08-01',0.09,0.8),\
                 ('2020-08-02',0.0483,0.8)]
rdd = sc.parallelize(df_data)
df = sqlContext.createDataFrame(df_data, df_schema)
df = df.withColumn("date",to_date("date", 'yyyy-MM-dd'))
df.show()

+----------+------+----+
|      date|  col1|col2|
+----------+------+----+
|2020-08-01|  0.09| 0.8|
|2020-08-02|0.0483| 0.8|
+----------+------+----+

我想使用 col1 和 col2 计算泊松 CDF.

And I want to calculate Poisson CDF using col1 and col2.

我们可以很容易地在 pandas 数据框中使用 from scipy.stats import poisson 但我不知道如何处理 pyspark.

we can easily use from scipy.stats import poisson in pandas dataframe but I don't know how to deal with pyspark.

prob = poisson.cdf(x, mu) 其中 x= col1 和 mu = col2 在我们的例子中.

prob = poisson.cdf(x, mu) where x= col1 , and mu = col2 in our case.

尝试 1:

from scipy.stats import poisson
from pyspark.sql.functions import udf,col
def poisson_calc(a,b):
    return poisson.cdf(a,b,axis=1)

poisson_calc = udf(poisson_calc, FloatType())

df_new = df.select(
  poisson_calc(col('col1'),col('col2')).alias("want") )

df_new.show()

给我一​​个错误:类型错误:_parse_args() 得到了一个意外的关键字参数轴"

Got me an error :TypeError: _parse_args() got an unexpected keyword argument 'axis'

推荐答案

我发现您的尝试存在一些问题.

I see some issues with your attempt.

• 您将 udf 命名为与底层函数相同的名称.令人惊讶的是，这本身实际上并不是问题，但我会避免它.
• scipy.stats.poisson.cdf
没有 axis 关键字参数
• 您必须将输出显式转换为 float，否则您将遇到 这个错误

• You named the udf the same as the underlying function. Surprisingly this actually isn't a problem per se but I would avoid it.
• There's no axis keyword argument to scipy.stats.poisson.cdf
• You'll have to explicitly convert the output to float or you'll run into this error

解决所有问题，以下应该有效:

Fixing that all up, the following should work:

from scipy.stats import poisson
from pyspark.sql.functions import udf,col

def poisson_calc(a,b):
    return float(poisson.cdf(a,b))

poisson_calc_udf = udf(poisson_calc, FloatType())

df_new = df.select(
  poisson_calc_udf(col('col1'),col('col2')).alias("want")
)

df_new.show()
#+----------+
#|      want|
#+----------+
#|0.44932896|
#|0.44932896|
#+----------+

这篇关于Pyspark，如何使用 udf 计算泊松分布?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！