如何在 pyspark 数据框列上拟合内核密度估计并将其用于创建具有估计的新列

Posted 2023-04-13

【中文标题】如何在 pyspark 数据框列上拟合内核密度估计并将其用于创建具有估计的新列

【英文标题】：How to fit a kernel density estimate on a pyspark dataframe column and use it for creating a new column with the estimates

【发布时间】：2018-03-08 11:10:58

【问题描述】：

我的用途如下。考虑我有一个具有以下格式的 pyspark 数据框： df.columns： 1. hh：包含一天中的小时（类型 int） 2. userId : 一些唯一标识符。

我想做的是找出在页面上有异常点击的用户 ID 列表。所以我首先做一个 groupby ： df=df.groupby("hh","userId).count().alias("LoginCounts)

现在数据框的格式为： 1.hh 2.用户ID 3.LoginCounts：特定用户在特定时间登录的次数。

我想使用pyspark kde功能如下：

from pyspark.mllib.stat import KernelDensity
kd=KernelDensity()
kd.setSample(df.select("LoginCounts").rdd)
kd.estimate([13.0,14.0]).

我得到错误： Py4JJavaError：调用 o647.estimateKernelDensity 时出错。 ：org.apache.spark.SparkException：作业因阶段失败而中止

现在我的最终目标是在一天中基于小时的数据上拟合 kde，然后使用第二天的数据来获得每个登录计数的概率估计值。 例如：我想实现这种性质的东西：

df.withColumn("kdeProbs",kde.estimate(col("LoginCounts)))

所以 kdeProbs 列将包含 P(LoginCount=x | 估计 kde)。

我曾尝试搜索相同的示例，但总是被重定向到 spark.apache.org 页面上的标准 kde 示例，这并不能解决我的问题。

【问题讨论】：

mllib 设计用于 RDD，而不是 DataFrames。对于 DataFrames，使用 ml

@Mike 在 ml 中没有核密度估计函数，只有在 mllib 中

你说得对，好评

【参考方案1】：

仅选择一列并将其转换为 RDD 是不够的；您还需要选择该列中的实际数据才能使其正常工作。试试这个：

from pyspark.mllib.stat import KernelDensity

dat_rdd = df.select("LoginCounts").rdd

# actually select data from RDD
dat_rdd_data = dat_rdd.map(lambda x: x[0])

kd = KernelDensity()
kd.setSample(dat_rdd_data)
kd.estimate([13.0,14.0])