0

I have a data frame and I want to roll up the data into 7days and do some aggregation on some of the function. I have a pyspark sql dataframe like ------

Sale_Date|P_1|P_2|P_3|G_1|G_2|G_3|Total_Sale|Sale_Amt|Promo_Disc_Amt|
|2013-04-10| 1| 9| 1| 1| 1| 1| 1| 295.0|0.0|
|2013-04-11| 1| 9| 1| 1| 1| 1| 3| 567.0|0.0|
|2013-04-12| 1| 9| 1| 1| 1| 1| 2| 500.0|200.0|
|2013-04-13| 1| 9| 1| 1| 1| 1| 1| 245.0|20.0|
|2013-04-14| 1| 9| 1| 1| 1| 1| 1| 245.0|0.0|
|2013-04-15| 1| 9| 1| 1| 1| 1| 2| 500.0|200.0|
|2013-04-16| 1| 9| 1| 1| 1| 1| 1| 250.0|0.0|
|2013-04-17| 1| 9| 1| 1| 1| 1| 1| 250.0|0.0|
|2013-04-18| 1| 9| 1| 1| 1| 1| 1| 250.0|0.0|

According to this threadI have applied a window function over the data frame as follows -

days = lambda i: i * 86400
windowSp = Window().partitionBy(dataframeOfquery3["P_1"],dataframeOfquery3["P_2"],dataframeOfquery3["P_3"],dataframeOfquery3["G_1"],dataframeOfquery3["G_2"],dataframeOfquery3["G_3"])\
          .orderBy(dataframeOfquery3["Sale_Date"].cast("timestamp").cast("long").desc())\
          .rangeBetween(-(days(7)), 0)

Now I want to perform some aggregation i.e. applying some windows functions like the following --

df = dataframeOfquery3.select(min(dataframeOfquery3["Sale_Date"]).over(windowSp).alias("Sale_Date"),first(dataframeOfquery3["P_1"]).over(windowSp).alias("P_1"))

But I cannot get the desired output. Desired output will be-

Sale_Date,P_1,P_2,P_3,g_1,G-2,G_3,Total_Sale,Sale_Amt,Promo_Disc_Amt
|2013-04-10| 1| 9| 1| 1| 1| 1| 11| 2602.0|420.0|
|2013-04-17| 1| 9| 1| 1| 1| 1| 7| 1902.0|120.0|
|2013-04-24| 1| 9| 1| 1| 1| 1| 10| 2402.0|120.0|

But it is not working. I am stuck on to it. I will be grateful if anyone could help me to sort this out.

4

0 回答 0