Расчет скользящих сумм в писпарке

Question

У меня есть датафрейм, который содержит информацию о ежедневных продажах и ежедневных кликах. Прежде чем я хочу выполнить свой анализ, я хочу объединить данные. Чтобы было понятнее, я попытаюсь объяснить это на примере фрейма данных

item_id date          Price  Sale   Click   Discount_code     
2       01.03.2019    10       1     10      NULL
2       01.03.2019    8        1     10      Yes
2       02.03.2019    10       0     4       NULL
2       03.03.2019    10       0     6       NULL
2       04.03.2019    6        0     15      NULL 
2       05.03.2019    6        0     14      NULL
2       06.03.2019    5        0     7       NULL 
2       07.03.2019    5        1     11      NULL
2       07.03.2019    5        1     11      NULL
2       08.03.2019    5        0     9       NULL

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

item_id    Price  CSale  Discount_code Cclicks   firstdate   lastdate  
2           10       1      No             20    01.03.2019 03.03.2019
2           8        1      Yes            10    01.03.2019 01.03.2019
2           6        0      NULL           29    04.03.2019 05.03.2019
2           5        2      NULL           38    06.03.2019 08.03.2019 

Если CSale соответствует совокупным продажам для данной цены и данного item_id, Cclicks соответствует совокупным щелчкам для данной цены и данного item_id, firstdate - это первая дата, когда данный товар был доступен для данной цены и последней даты. последняя дата, когда данный предмет был доступен по указанной цене.

Answer 1

В зависимости от проблемы, OP хочет агрегировать DataFrame на основе item_id и Price.

# Creating the DataFrames
from pyspark.sql.functions import col, to_date, sum, min, max, first
df = sqlContext.createDataFrame([(2,'01.03.2019',10,1,10,None),(2,'01.03.2019',8,1,10,'Yes'),
                                 (2,'02.03.2019',10,0,4,None),(2,'03.03.2019',10,0,6,None),
                                 (2,'04.03.2019',6,0,15,None),(2,'05.03.2019',6,0,14,None),
                                 (2,'06.03.2019',5,0,7,None),(2,'07.03.2019',5,1,11,None),
                                 (2,'07.03.2019',5,1,11,None),(2,'08.03.2019',5,0,9,None)],
                                ('item_id','date','Price','Sale','Click','Discount_code'))
# Converting string column date to proper date
df = df.withColumn('date',to_date(col('date'),'dd.MM.yyyy'))
df.show()
+-------+----------+-----+----+-----+-------------+
|item_id|      date|Price|Sale|Click|Discount_code|
+-------+----------+-----+----+-----+-------------+
|      2|2019-03-01|   10|   1|   10|         null|
|      2|2019-03-01|    8|   1|   10|          Yes|
|      2|2019-03-02|   10|   0|    4|         null|
|      2|2019-03-03|   10|   0|    6|         null|
|      2|2019-03-04|    6|   0|   15|         null|
|      2|2019-03-05|    6|   0|   14|         null|
|      2|2019-03-06|    5|   0|    7|         null|
|      2|2019-03-07|    5|   1|   11|         null|
|      2|2019-03-07|    5|   1|   11|         null|
|      2|2019-03-08|    5|   0|    9|         null|
+-------+----------+-----+----+-----+-------------+

. Как видно из printSchema ниже,столбец date фрейма данных имеет формат date.

df.printSchema()
root
 |-- item_id: long (nullable = true)
 |-- date: date (nullable = true)
 |-- Price: long (nullable = true)
 |-- Sale: long (nullable = true)
 |-- Click: long (nullable = true)
 |-- Discount_code: string (nullable = true)

Наконец, агрегирует agg() столбцы ниже.Просто предостережение - поскольку Discount_code является столбцом string, и нам также необходимо его агрегировать, при группировании мы примем first ненулевое значение.

df = df.groupBy('item_id','Price').agg(sum('Sale').alias('CSale'),
                                       first('Discount_code',ignorenulls = True).alias('Discount_code'),
                                       sum('Click').alias('Cclicks'),
                                       min('date').alias('firstdate'),
                                       max('date').alias('lastdate'))
df.show()
+-------+-----+-----+-------------+-------+----------+----------+
|item_id|Price|CSale|Discount_code|Cclicks| firstdate|  lastdate|
+-------+-----+-----+-------------+-------+----------+----------+
|      2|    6|    0|         null|     29|2019-03-04|2019-03-05|
|      2|    5|    2|         null|     38|2019-03-06|2019-03-08|
|      2|    8|    1|          Yes|     10|2019-03-01|2019-03-01|
|      2|   10|    1|         null|     20|2019-03-01|2019-03-03|
+-------+-----+-----+-------------+-------+----------+----------+