在 pyspark 的另一个表中循环并查找几行

Question

我有两个数据框， 表 1：用户在第 0 天购买的商品 表 2：x 天内的商品价格（每天波动）

我想匹配用户购买商品的价格。有没有更好的方法来做到这一点而不循环每一行然后应用一个函数？

我的最终输出是我想知道 john 在 1/1 购买苹果时的 rolling_average 3 天平均值是多少？

第一个表： 约翰的桌子（可能会有更多用户）

Date    Item    Price
1/1/2018    Apple   1
2/14/2018   Grapes  1.99
1/25/2018   Pineapple   1.5
5/25/2018   Apple   0.98

参考表：价格表

Date    Item    Price
1/1/2018    Apple   1
1/2/2018    Apple   0.98
1/3/2018    Apple   0.88
1/4/2018    Apple   1.2
1/5/2018    Apple   1.3
1/6/2018    Apple   1.5
1/7/2018    Apple   1.05
1/8/2018    Apple   1.025
2/10/2018   Grapes  3.10
2/11/2018   Grapes  0.10
2/12/2018   Grapes  5.00
2/13/2018   Grapes  0.40
2/14/2018   Grapes  1.00
2/15/2018   Grapes  2.70
2/16/2018   Grapes  0.40
2/17/2018   Grapes  0.40
1/23/2018   Pineapple   0.50
1/24/2018   Pineapple   0.60
1/25/2018   Pineapple   0.70
1/26/2018   Pineapple   0.60
1/27/2018   Pineapple   0.60
1/28/2018   Pineapple   0.50
1/29/2018   Pineapple   0.70
1/30/2018   Pineapple   0.50
5/21/2018   Apple   7.00
5/22/2018   Apple   6.00
5/23/2018   Apple   5.00
5/24/2018   Apple   6.00
5/25/2018   Apple   5.00

苹果的例子：

Date    Item    Price       
1/1/2018    Apple   1       #bought on this date
1/2/2018    Apple   0.98    #so next 3 days
1/3/2018    Apple   0.88    0.953333333
1/4/2018    Apple   1.2     1.02
1/5/2018    Apple   1.3     1.126666667
1/6/2018    Apple   1.5     1.333333333
1/7/2018    Apple   1.05    1.283333333
1/8/2018    Apple   1.025   1.191666667

df_price.withColumn('rolling_Average', f.avg("Price").over(Window.partitionBy(f.window("Date", "3 days"))))

Answer 1

所以如果我正确理解问题，您想计算每个项目的 3 天平均值。然后，您只需将表 1 连接到表 2，即可仅获得平均价格与实际价格相邻的已售商品。 您可以通过使用窗口功能来做到这一点。 在 pyspark 中可能是这样的：

import pyspark.sql.functions as F
from pyspark.sql.window import Window

df_price = df_price.withColumn(
    'rolling_average',
    F.avg(df_price.price).over(
        Window.partitionBy(df_price.item).orderBy(
            df_price.date.desc()
        ).rowsBetween(0, 3)
    )
)

然后你只需将你的表加入到这个结果中。 在 SQL 中是这样的：

WITH b as (
SELECT '1/1/2018' as date_p,  'Apple' as item, 1 as price
UNION ALL SELECT '1/2/2018' as date_p,  'Apple' as item, 0.98 as price
UNION ALL SELECT '1/3/2018' as date_p,  'Apple' as item, 0.88 as price
UNION ALL SELECT '1/4/2018' as date_p,  'Apple' as item, 1.2 as price
UNION ALL SELECT '1/5/2018' as date_p,  'Apple' as item, 1.3 as price
UNION ALL SELECT '1/6/2018' as date_p,  'Apple' as item, 1.5 as price
UNION ALL SELECT '1/7/2018' as date_p,  'Apple' as item, 1.05 as price
UNION ALL SELECT '1/8/2018' as date_p,  'Apple' as item, 1.025 as price
UNION ALL SELECT '2/10/2018' as date_p, 'Grape' as item, 3.10 as price)
SELECT *, AVG(price) OVER (
  PARTITION BY item ORDER BY date_p DESC
  ROWS BETWEEN CURRENT ROW AND 2 FOLLOWING
) from b

Answer 2

如果您只是想按特定项目进行分组（将您的价格表设置为df2）：

df2['Date'] = pd.to_datetime(df2['Date'])
df2 = df2.set_index('Date')

df2['Rolling'] = df2[df2['Item']=='Apple']['Price'].rolling(3).mean()

打印df2[df2['Item']=='Apple'] 将产生：

             Item  Price   Rolling
Date                              
2018-01-01  Apple  1.000       NaN
2018-01-02  Apple  0.980       NaN
2018-01-03  Apple  0.880  0.953333
2018-01-04  Apple  1.200  1.020000
2018-01-05  Apple  1.300  1.126667
2018-01-06  Apple  1.500  1.333333
2018-01-07  Apple  1.050  1.283333
2018-01-08  Apple  1.025  1.191667
2018-05-21  Apple  7.000  3.025000
2018-05-22  Apple  6.000  4.675000
2018-05-23  Apple  5.000  6.000000
2018-05-24  Apple  6.000  5.666667
2018-05-25  Apple  5.000  5.333333

如果您想限制某些日期分组，答案会略有不同。