Pyspark 高级 window 函数

Question

这是我的数据框：

FlightDate=[20,40,51,50,60,15,17,37,36,50]
IssuingDate=[10,15,44,45,55,10,2,30,32,24]
Revenue = [100,50,40,70,60,40,30,100,200,100]
Customer = ['a','a','a','a','a','b','b','b','b','b']
df = spark.createDataFrame(pd.DataFrame([Customer,FlightDate,IssuingDate, Revenue]).T, schema=["Customer",'FlightDate', 'IssuingDate','Revenue'])
df.show()

+--------+----------+-----------+-------+
|Customer|FlightDate|IssuingDate|Revenue|
+--------+----------+-----------+-------+
|       a|        20|         10|    100|
|       a|        40|         15|     50|
|       a|        51|         44|     40|
|       a|        50|         45|     70|
|       a|        60|         55|     60|
|       b|        15|         10|     40|
|       b|        27|          2|     30|
|       b|        37|         30|    100|
|       b|        36|         32|    200|
|       b|        50|         24|    100|
+--------+----------+-----------+-------+

为了方便，我用了天数。

对于每个客户，我想对研究的 FlightDate 和研究的 FlightDate + 10 天之间的所有发行日期的收入求和。

也就是说：

对于第一行：我将 IssuingDate 在第 20 天和第 30 天之间的所有收入相加...这里给出 0。
对于第二行：我将 IssuingDate 在第 40 天和第 50 天之间的所有收入相加，也就是说 40+70 = 110

这是想要的结果：

+--------+----------+-----------+-------+------+
|Customer|FlightDate|IssuingDate|Revenue|Result|
+--------+----------+-----------+-------+------+
|       a|        20|         10|    100|     0|
|       a|        40|         15|     50|   110|
|       a|        51|         44|     40|    60|
|       a|        50|         45|     70|    60|
|       a|        60|         55|     60|     0|
|       b|        15|         10|     40|   100|
|       b|        27|          2|     30|   300|
|       b|        37|         30|    100|     0|
|       b|        36|         32|    200|     0|
|       b|        50|         24|    100|     0|
+--------+----------+-----------+-------+------+

我知道它会涉及到一些 window 功能，但这一个似乎有点棘手。谢谢

Answer 1

不需要 window 功能。它只是一个连接和一个聚合：

df.alias("df").join(
    df.alias("df_2"),
    on=F.expr(
        "df.Customer = df_2.Customer "
        "and df_2.issuingdate between df.flightdate and df.flightdate+10"
    ), 
    how='left'
).groupBy(
    *('df.{}'.format(c) 
      for c 
      in df.columns)
).agg(
    F.sum(F.coalesce(
        "df_2.revenue", 
        F.lit(0))
    ).alias("result")
).show()

+--------+----------+-----------+-------+------+                                
|Customer|FlightDate|IssuingDate|Revenue|result|
+--------+----------+-----------+-------+------+
|       a|        20|         10|    100|     0|
|       a|        40|         15|     50|   110|
|       a|        50|         45|     70|    60|
|       a|        51|         44|     40|    60|
|       a|        60|         55|     60|     0|
|       b|        15|         10|     40|   100|
|       b|        27|          2|     30|   300|
|       b|        36|         32|    200|     0|
|       b|        37|         30|    100|     0|
|       b|        50|         24|    100|     0|
+--------+----------+-----------+-------+------+

Answer 2

如果您想保留当前行和接下来 10 天的收入，则可以使用以下代码。

例如

第一行：flightDate = 20，您需要 20 到 30 之间的收入（包括首尾两日），这意味着总收入 = 100。

第二行：flightDate = 40，您需要 40 到 50 之间的收入（包括两个日期），这意味着总收入 = 50（对于日期 40）+ 50（对于日期 50）= 120。

第三行：flightDate = 50 并且您需要 50 到 60 之间的收入（包括两个日期），这意味着总收入 = 70（对于日期 50）+ 40（对于日期 51）+ 60（对于日期 60）= 170

from pyspark.sql import *
from pyspark.sql.functions import *
import pandas as pd

FlightDate=[20,40,51,50,60,15,17,37,36,50]
IssuingDate=[10,15,44,45,55,10,2,30,32,24]
Revenue = [100,50,40,70,60,40,30,100,200,100]
Customer = ['a','a','a','a','a','b','b','b','b','b']
df = spark.createDataFrame(pd.DataFrame([Customer,FlightDate,IssuingDate, Revenue]).T, schema=["Customer",'FlightDate', 'IssuingDate','Revenue'])


windowSpec = Window.partitionBy("Customer").orderBy("FlightDate").rangeBetween(0,10)
df.withColumn("Sum", sum("Revenue").over(windowSpec)).sort("Customer").show()

Result as mentioned below

+--------+----------+-----------+-------+---+
|Customer|FlightDate|IssuingDate|Revenue|Sum|
+--------+----------+-----------+-------+---+
|       a|        20|         10|    100|100|
|       a|        40|         15|     50|120|
|       a|        50|         45|     70|170|
|       a|        51|         44|     40|100|
|       a|        60|         55|     60| 60|
|       b|        15|         10|     40| 70|
|       b|        17|          2|     30| 30|
|       b|        36|         32|    200|300|
|       b|        37|         30|    100|100|
|       b|        50|         24|    100|100|
+--------+----------+-----------+-------+---+

Pyspark 高级 window 函数

Pyspark advanced window function

python

window-functions

pyspark

pyspark-sql