How to compute the sum of orders over a 12 months period sliding by 1 month per customer in Spark

Viewed 2248

I am relatively new to spark with Scala. currently I am trying to aggregate order data in spark over a 12 months period that slides monthly.

Below is a simple sample of my data, I tried to format it so you can easily test it

import spark.implicits._
import org.apache.spark.sql._
import org.apache.spark.sql.functions._


var sample = Seq(("C1","01/01/2016", 20), ("C1","02/01/2016", 5), 
 ("C1","03/01/2016", 2),  ("C1","04/01/2016", 3), ("C1","05/01/2017", 5),
 ("C1","08/01/2017", 5), ("C1","01/02/2017", 10), ("C1","01/02/2017", 10),  
 ("C1","01/03/2017", 10)).toDF("id","order_date", "orders")

sample = sample.withColumn("order_date",
to_date(unix_timestamp($"order_date", "dd/MM/yyyy").cast("timestamp")))

sample.show 
 +---+----------+------+
 | id|order_date|orders|
 +---+----------+------+
 | C1|2016-01-01|    20|
 | C1|2016-01-02|     5|
 | C1|2016-01-03|     2|
 | C1|2016-01-04|     3|
 | C1|2017-01-05|     5|
 | C1|2017-01-08|     5|
 | C1|2017-02-01|    10|
 | C1|2017-02-01|    10|
 | C1|2017-03-01|    10|
 +---+----------+------+

the imposed upon me outcome is the following.

id      period_start    period_end  rolling
C1      2015-01-01      2016-01-01  30
C1      2016-01-01      2017-01-01  40
C1      2016-02-01      2017-02-01  30
C1      2016-03-01      2017-03-01  40

what I tried to do so far

I collapsed the dates per costumer to the first day of the month

(e.i. 2016-01-[1..31] >> 2016-01-01 )

import org.joda.time._

val collapse_month = (month:Integer, year:Integer ) => {
   var  dt = new DateTime().withYear(year)
                        .withMonthOfYear(month)
                        .withDayOfMonth(1)
   dt.toString("yyyy-MM-dd")
 }

val collapse_month_udf = udf(collapse_month)


sample = sample.withColumn("period_end",
           collapse_month_udf(
           month(col("order_date")),
           year(col("order_date"))
           ).as("date"))

sample.groupBy($"id",  $"period_end")
              .agg(sum($"orders").as("orders"))
              .orderBy("period_end").show
 +---+----------+------+
 | id|period_end|orders|
 +---+----------+------+
 | C1|2016-01-01|    30|
 | C1|2017-01-01|    10|
 | C1|2017-02-01|    20|
 | C1|2017-03-01|    10|
 +---+----------+------+

I tried the provided window function but I was not able to use 12 months sliding by one option.

I am really not sure what is the best way to proceed from this point, that would not take 5 hours given how much data I have to work with.

Any help would be appreciated.

1 Answers
Related