scala – 如何计算Spark中12个月内每个客户滑动1个月的订单总和

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

您仍然可以使用间隔较长的窗口,但所有参数都必须以天或周表示：

window($"order_date","365 days","28 days")

不幸的是,这个窗口不会尊重月份或年份的界限,因此对您来说不会有用.

就个人而言,我会先汇总数据：

val byMonth = sample
  .groupBy($"id",trunc($"order_date","month").alias("order_month"))
  .agg(sum($"orders").alias("orders"))

+---+-----------+-----------+                                                   
| id|order_month|sum(orders)|
+---+-----------+-----------+
| C1| 2017-01-01|         10|
| C1| 2016-01-01|         30|
| C1| 2017-02-01|         20|
| C1| 2017-03-01|         10|
+---+-----------+-----------+

创建参考日期范围：

import java.time.temporal.ChronoUnit

val Row(start: java.sql.Date,end: java.sql.Date) = byMonth
  .select(min($"order_month"),max($"order_month"))
  .first

val months = (0L to ChronoUnit.MONTHS.between(
    start.toLocalDate,end.toLocalDate))
  .map(i => java.sql.Date.valueOf(start.toLocalDate.plusMonths(i)))
  .toDF("order_month")

并结合独特的ID：

val ref = byMonth.select($"id").distinct.crossJoin(months)

并与来源联系：

val expanded = ref.join(byMonth,Seq("id","order_month"),"leftouter")

+---+-----------+------+ 
| id|order_month|orders|
+---+-----------+------+
| C1| 2016-01-01|    30|
| C1| 2016-02-01|  null|
| C1| 2016-03-01|  null|
| C1| 2016-04-01|  null|
| C1| 2016-05-01|  null|
| C1| 2016-06-01|  null|
| C1| 2016-07-01|  null|
| C1| 2016-08-01|  null|
| C1| 2016-09-01|  null|
| C1| 2016-10-01|  null|
| C1| 2016-11-01|  null|
| C1| 2016-12-01|  null|
| C1| 2017-01-01|    10|
| C1| 2017-02-01|    20|
| C1| 2017-03-01|    10|
+---+-----------+------+

使用这样的数据准备你可以使用窗口函数：

import org.apache.spark.sql.expressions.Window

val w = Window.partitionBy($"id")
     .orderBy($"order_month")
    .rowsBetween(-12,Window.currentRow)

expanded.withColumn("rolling",sum("orders").over(w))
  .na.drop(Seq("orders"))
  .select(
      $"order_month" - expr("INTERVAL 12 MONTHS") as "period_start",$"order_month" as "period_end",$"rolling")

+------------+----------+-------+
|period_start|period_end|rolling|
+------------+----------+-------+
|  2015-01-01|2016-01-01|     30|
|  2016-01-01|2017-01-01|     40|
|  2016-02-01|2017-02-01|     30|
|  2016-03-01|2017-03-01|     40|
+------------+----------+-------+

请注意,这是一项非常昂贵的操作,需要至少两次洗牌：

== Physical Plan ==
*Project [cast(cast(order_month#104 as timestamp) - interval 1 years as date) AS period_start#1387,order_month#104 AS period_end#1388,rolling#1375L]
+- *Filter AtLeastNNulls(n,orders#55L)
   +- Window [sum(orders#55L) windowspecdefinition(id#7,order_month#104 ASC NULLS FIRST,ROWS BETWEEN 12 PRECEDING AND CURRENT ROW) AS rolling#1375L],[id#7],[order_month#104 ASC NULLS FIRST]
      +- *Sort [id#7 ASC NULLS FIRST,order_month#104 ASC NULLS FIRST],false,0
         +- Exchange hashpartitioning(id#7,200)
            +- *Project [id#7,order_month#104,orders#55L]
               +- *BroadcastHashJoin [id#7,order_month#104],[id#181,order_month#49],LeftOuter,BuildRight
                  :- BroadcastNestedLoopJoin BuildRight,Cross
                  :  :- *HashAggregate(keys=[id#7],functions=[])
                  :  :  +- Exchange hashpartitioning(id#7,200)
                  :  :     +- *HashAggregate(keys=[id#7],functions=[])
                  :  :        +- *HashAggregate(keys=[id#7,trunc(order_date#14,month)#1394],functions=[])
                  :  :           +- Exchange hashpartitioning(id#7,month)#1394,200)
                  :  :              +- *HashAggregate(keys=[id#7,month) AS trunc(order_date#14,functions=[])
                  :  :                 +- LocalTableScan [id#7,order_date#14]
                  :  +- BroadcastExchange IdentityBroadcastMode
                  :     +- LocalTableScan [order_month#104]
                  +- BroadcastExchange HashedRelationBroadcastMode(List(input[0,string,true],input[1,date,true]))
                     +- *HashAggregate(keys=[id#181,month)#1395],functions=[sum(cast(orders#183 as bigint))])
                        +- Exchange hashpartitioning(id#181,month)#1395,200)
                           +- *HashAggregate(keys=[id#181,functions=[partial_sum(cast(orders#183 as bigint))])
                              +- LocalTableScan [id#181,order_date#14,orders#183]

也可以使用rangeBetween帧来表达这一点,但您必须首先对数据进行编码：

val encoded = byMonth
  .withColumn("order_month_offset",// Choose "zero" date appropriate in your scenario
      months_between($"order_month",to_date(lit("1970-01-01"))))


val w = Window.partitionBy($"id")
  .orderBy($"order_month_offset")
  .rangeBetween(-12,Window.currentRow)

encoded.withColumn("rolling",sum($"orders").over(w))

+---+-----------+------+------------------+-------+                             
| id|order_month|orders|order_month_offset|rolling|
+---+-----------+------+------------------+-------+
| C1| 2016-01-01|    30|             552.0|     30|
| C1| 2017-01-01|    10|             564.0|     40|
| C1| 2017-02-01|    20|             565.0|     30|
| C1| 2017-03-01|    10|             566.0|     40|
+---+-----------+------+------------------+-------+

这将使参考的连接过时并简化执行计划.