Apache Scala/Python Spark 2.4.4:按年份范围分组数据以生成/分析新特性

Question

下面是我为特性工程生成的数据框架，现在为了驱动另一个特性，我试图创建PurchaseErrange列，其中我希望创建一个具有3年范围的列，并通过Itemno、modelnumber和valueClass列聚合ModelRatio、PurchaseRatio值。

例如：对于项目编号7010032，我们将在新的PurchaseErrange列和ModelRatio、PurchaseRatio列中有一个具有1995-1996-1997值的行，这些年的PurchaseRatio值将在相应的行中汇总。接下来，我将在接下来的3年中也这样做，即1996-1997-1998，1997-1998-1999等。

此外，该项目有一行itemclass-rgr，对于该行，我们将只有具有1996年数据的该行。

基本上，在dataframe中检查3年的数据，如果存在，则对3年范围窗口进行modelratio、purchaseratio求和。如果三年的数据不存在，那么根据数据可用性对两年或一年进行求和。

null

| itemNo|modelnumber|itemClass |purchaseYear|          ModelRatio|      PurchaseRatio|
+-------+-----------+----------+------------+--------------------+-------------------+
|1321457|  A99AA25CA|       ATV|        1995|   1.801325096248545|                2.0|
|7010032|  A99AA25CB|       ATV|        1995|  1.0048348106365834|                2.0|
|7010032|  A99AA25CB|       ATV|        1996|  0.8899632912525741|                2.0|
|7010032|  A99AA25CB|       RGR|        1996|  0.7899632912525741|                1.0| 
|7010032|  A99AA25CB|       ATV|        1997|   1.669710806697108|                2.0|  
|7010032|  A99AA25CB|       ATV|        1998|  0.9982988629241651|                2.0| 
|7010032|  A99AA25CB|       ATV|        1999|0.006535947712418301|                1.0|
|7552901|  A99AA25CD|       ATV|        1995|   37.83901871250784|               12.0| 
|7552901|  A99AA25CD|       ATV|        1996|0.026143790849673203|                1.0|
|7552901|  A99AA25CD|       ATV|        1997|  0.9375951293759512|                2.0|

我是scala spark的新手，使用.createorReplaceTempView（“test”）尝试了它，然后应用SQL操作，但是这种方式非常复杂。你能分享一下我如何完成这件事吗？请随时用Python和或Scala提出解决方案。

Answer 1

IIUC，您可以使用Spark SQL窗口函数如下所示：（确保purchaseYear是一个数字列或时间戳列）

编辑：根据注释，添加了all_puchase_years以包括3年序列。请注意，order BY itemNo,purchaseyear子句仅用于演示目的。

spark.sql(""" 
       SELECT itemNo 
       ,      modelnumber 
       ,      itemClass 
       ,      concat_ws('-', sort_array(collect_set(purchaseYear) OVER w1)) AS purchase_years 
       ,      concat_ws('-', sequence(purchaseYear, purchaseYear+2)) AS all_purchase_years
       ,      sum(PurchaseRatio) OVER w1 AS sum_PurchaseRatio 
       ,      sum(ModelRatio) OVER w1 AS sum_ModelRatio 
       FROM test 
       ORDER BY itemNo, purchaseYear 
       WINDOW w1 AS (
           PARTITION BY (itemNo, modelnumber, itemClass)  
           ORDER BY purchaseYear 
           RANGE BETWEEN CURRENT ROW AND 2 FOLLOWING
       ) 
""").show() 
#+-------+-----------+---------+--------------+-----------------+--------------------+
#| itemNo|modelnumber|itemClass|purchase_years|sum_PurchaseRatio|      sum_ModelRatio|
#+-------+-----------+---------+--------------+-----------------+--------------------+
#|1321457|  A99AA25CA|      ATV|          1995|              2.0|   1.801325096248545|
#|7010032|  A99AA25CB|      ATV|1995-1996-1997|              6.0|   3.564508908586266|
#|7010032|  A99AA25CB|      RGR|          1996|              1.0|  0.7899632912525741|
#|7010032|  A99AA25CB|      ATV|1996-1997-1998|              6.0|  3.5579729608738475|
#|7010032|  A99AA25CB|      ATV|1997-1998-1999|              5.0|  2.6745456173336914|
#|7010032|  A99AA25CB|      ATV|     1998-1999|              3.0|  1.0048348106365834|
#|7010032|  A99AA25CB|      ATV|          1999|              1.0|0.006535947712418301|
#|7552901|  A99AA25CD|      ATV|1995-1996-1997|             15.0|   38.80275763273346|
#|7552901|  A99AA25CD|      ATV|     1996-1997|              3.0|  0.9637389202256245|
#|7552901|  A99AA25CD|      ATV|          1997|              2.0|  0.9375951293759512|
#+-------+-----------+---------+--------------+-----------------+--------------------+