Цикл и поиск нескольких строк в другой таблице в pyspark

Question

У меня есть два кадра данных, таблица 1: приобретенный пользователем товар в день 0, таблица 2: цена товара в течение x дней (колеблется изо дня в день)

Я хочу соответствовать, когда пользователь покупает цену навещь.Есть ли лучший способ сделать это без зацикливания каждой строки, а затем применить функцию?

Мой окончательный вывод - я хочу знать, каково среднее значение для яблок в среднем за 3 дня, когда Джон купил его на 1/1?

Первая таблица: таблица Джона (может быть больше пользователей)

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

Пример для Apple:

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

Если вы просто хотите сгруппировать по определенному элементу (установив таблицу цен на 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

Ответ немного отличается, если вы хотите ограничиться определенными группами дат.

Answer 2

поэтому, если я правильно понимаю проблему, вы хотите рассчитать среднее значение за 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