Pyspark - 按两个日期列分组的 UDAF 函数，UDAF 用于计算实际值和预测值之间的 RMSE

Posted 2023-04-13

【中文标题】Pyspark - 按两个日期列分组的 UDAF 函数，UDAF 用于计算实际值和预测值之间的 RMSE

【英文标题】：Pyspark - UDAF function for a groupby two date columns, UDAF to calculate RMSE between actuals and predictions

【发布时间】：2020-04-11 19:56:04

【问题描述】：

在接下来的几年里，我在这样的 pyspark 数据框中有数据。 week_start_dt 是我开始预测的时间。 start_month 是前 12 个月。

+--------------------+------------------+----------------------+----------------+
|     start_month    |     week_start_dt|           predictions|       actuals  |
+--------------------+------------------+----------------------+----------------+
|             2019-01|        2019-11-11|                    12|              11|
|             2018-12|        2019-11-11|                    13|              11|
|             2019-08|        2019-11-11|                     9|              11|
|             2019-11|        2019-11-11|                    12|              11|
|             2019-11|        2019-11-11|                  1970|            1440|
|             2019-11|        2019-11-11|                   478|             501|
+--------------------+------------------+----------------------+----------------+

我想用groupby 在start_month 和week_start_dt 上计算RMSE。我认为它需要一个用户定义的聚合函数。 大熊猫中类似的东西：Python Dataframe: Calculating R^2 and RMSE Using Groupby on One Column

我使用以下代码来获取 groupby 的实际值和预测值。

df_startmonth_week = actuals_compare.groupby('start_month', 'week_start_dt').agg(f.sum('predictions'), f.sum('actuals'))

在计算预测值和实际值之间的 RMSE 时，我应该在聚合步骤中进行哪些更改？我需要 UDF 来执行此操作吗？

这是我在 excel 中制定的最终目标的示例

| week_start_dt | start_month | RMSE |
|---------------|-------------|------|
| 20-01-2020    | 2019-02     | 2345 |
| 20-01-2020    | 2019-03     | 2343 |
| 20-01-2020    | 2019-04     | 2341 |
| 20-01-2020    | 2019-05     | 2100 |
| 20-01-2020    | 2019-06     | 1234 |

【问题讨论】：

我会选择 udf - 另外，请查看 spark.apache.org/docs/1.6.3/…

【参考方案1】：

我看不出与problem here 的区别，所以我将解决方案调整为稍微不同的变量名：

import pyspark.sql.functions as psf

def compute_RMSE(expected_col, actual_col):

  rmse = old_df.withColumn("squarederror",
                           psf.pow(psf.col(actual_col) - psf.col(expected_col),
                                   psf.lit(2)
                           ))
  .groupby('start_month', 'week_start_dt')
  .agg(psf.avg(psf.col("squarederror")).alias("mse"))
  .withColumn("rmse", psf.sqrt(psf.col("mse")))

  return(rmse)


compute_RMSE("predictions", "actuals")

如果我错过了问题的细微差别，请告诉我