我是 PySpark 的新手，正在努力处理简单的数据帧操作。我有一个类似于以下内容的数据框:
product period rating product_Desc1 product_Desc2 ..... more columns
a 1 60 foo xx
a 2 70 foo xx
a 3 59 foo xx
b 1 50 bar yy
b 2 55 bar yy
c 1 90 foo bar xy
c 2 100 foo bar xy
我想按产品分组，添加列来计算评级的算术、几何和调和平均值，同时还维护数据框中的其余列，这些列在每个产品中都是一致的。
我尝试结合内置函数和 UDF 来实现此目的。例如:
a_means = df.groupBy("product").agg(mean("rating").alias("a_mean")
g_means = df.groupBy("product").agg(udf_gmean("rating").alias("g_mean")
地点:
def g_mean(x):
gm = reduce(mul,x)**(1/len(x))
return gm
udf_gmean = udf(g_mean, FloatType())
然后，我将 a_means 和 g_means 输出与产品上的原始数据帧连接起来并删除重复项。然而，这个方法返回一个错误，对于g_means，指出“评分”不参与groupBy，也不是用户定义的聚合函数......
我也尝试过使用 SciPy 的 gmean 模块，但我收到的错误消息指出 ufunc 'log' 不适合输入类型，尽管据我所知，所有评级列都是整数类型。
网站上有类似的问题，但我找不到任何可以解决我遇到的问题的方法。我真的很感激你的帮助，因为它让我发疯!
提前致谢，如果我没有提供足够的信息，我今天应该能够快速提供任何进一步的信息。
值得注意的是，为了提高效率，我无法像使用 Pandas 数据帧那样简单地转换为 Pandas 并进行转换...而且我使用的是 Spark 2.2 并且无法更新!
最佳答案
像这样怎么样
from pyspark.sql.functions import avg
df1 = df.select("product","rating").rdd.map(lambda x: (x[0],(1.0,x[1]*1.0))).reduceByKey(lambda x,y: (x[0]+y[0], x[1]*y[1])).toDF(['product', 'g_mean'])
gdf = df1.select(df1['product'],pow(df1['g_mean._2'],1.0/df1['g_mean._1']).alias("rating_g_mean"))
display(gdf)
+-------+-----------------+
|product| rating_g_mean|
+-------+-----------------+
| a|62.81071936240795|
| b|52.44044240850758|
| c|94.86832980505137|
+-------+-----------------+
df1 = df.withColumn("h_mean", 1.0/df["rating"])
hdf = df1.groupBy("product").agg(avg(df1["rating"]).alias("rating_mean"), (1.0/avg(df1["h_mean"])).alias("rating_h_mean"))
sdf = hdf.join(gdf, ['product'])
display(sdf)
+-------+-----------+-----------------+-----------------+
|product|rating_mean| rating_h_mean| rating_g_mean|
+-------+-----------+-----------------+-----------------+
| a| 63.0|62.62847514743051|62.81071936240795|
| b| 52.5|52.38095238095239|52.44044240850758|
| c| 95.0|94.73684210526315|94.86832980505137|
+-------+-----------+-----------------+-----------------+
fdf = df.join(sdf, ['product'])
display(fdf.sort("product"))
+-------+------+------+-------------+-------------+-----------+-----------------+-----------------+
|product|period|rating|product_Desc1|product_Desc2|rating_mean| rating_h_mean| rating_g_mean|
+-------+------+------+-------------+-------------+-----------+-----------------+-----------------+
| a| 3| 59| foo| xx| 63.0|62.62847514743051|62.81071936240795|
| a| 2| 70| foo| xx| 63.0|62.62847514743051|62.81071936240795|
| a| 1| 60| foo| xx| 63.0|62.62847514743051|62.81071936240795|
| b| 2| 55| bar| yy| 52.5|52.38095238095239|52.44044240850758|
| b| 1| 50| bar| yy| 52.5|52.38095238095239|52.44044240850758|
| c| 2| 100| foo bar| xy| 95.0|94.73684210526315|94.86832980505137|
| c| 1| 90| foo bar| xy| 95.0|94.73684210526315|94.86832980505137|
+-------+------+------+-------------+-------------+-----------+-----------------+-----------------+
关于python - PySpark 中的 Groupby 和 UDF/UDAF，同时保持 DataFrame 结构，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/53031808/