【发布时间】：2020-05-26 09:04:21
【问题描述】：
这个问题与我之前的问题有关。 pyspark dataframe aggregate a column by sliding time window
但是，我想创建一个帖子，以澄清我上一个问题中缺少的一些关键点。
原始数据框：
client_id value1 name1 a_date
dhd 561 ecdu 2019-10-8
dhd 561 tygp 2019-10-8
dhd 561 rdsr 2019-10-8
dhd 561 rgvd 2019-8-12
dhd 561 bhnd 2019-8-12
dhd 561 prti 2019-8-12
dhd 561 teuq 2019-5-7
dhd 561 wnva 2019-5-7
dhd 561 pqhn 2019-5-7
我需要为每个“client_id”、每个“value1”以及某个给定的滑动时间窗口找到“name1”的值。
我定义了一个窗口函数：
w = window().partitionBy("client_id", "value1").orderBy("a_date")
但我不知道如何为窗口大小 1、2、6、9、12 选择“name1”的值。
这里，窗口大小是指“a_date”当前月份的月份长度。
例如
client_id value1 names1_within_window_size_1 names1_within_window_size_2
dhd 561 [ecdu,tygp,rdsr] [ecdu,tygp,rdsr]
names1_within_window_size_6
[ecdu,tygp,rdsr, rgvd,bhnd,prti, teuq, wnva,pqhn ]
names1_within_window_size_1 : the month window 2019-10
names1_within_window_size_2 : the month window 2019-10 and 2019-9 (no data in 2019-9 so just keep the data from 2019-10)
names1_within_window_size_6 : the month window 2019-10 and 2019-9 (no data in 2019-9 so just keep the data from 2019-10) but there are data in 2019-8
谢谢
============================================= 更新
from pyspark.sql import functions as F
from pyspark.sql.window import Window
data= [['dhd',589,'ecdu','2020-1-5'],
['dhd',575,'tygp','2020-1-5'],
['dhd',821,'rdsr','2020-1-5'],
['dhd',872,'rgvd','2019-12-1'],
['dhd',619,'bhnd','2019-12-15'],
['dhd',781,'prti','2019-12-18'],
['dhd',781,'prti1','2019-12-18'],
['dhd',781,'prti2','2019-11-18'],
['dhd',781,'prti3','2019-10-31'],
['dhd',781,'prti4','2019-09-30'],
['dhd',781,'prt1','2019-07-31'],
['dhd',781,'pr4','2019-06-30'],
['dhd',781,'pr2','2019-08-31'],
['dhd',781,'prt4','2019-01-31'],
['dhd',781,'prti6','2019-02-28'],
['dhd',781,'prti7','2019-02-02'],
['dhd',781,'prti8','2019-03-29'],
['dhd',781,'prti9','2019-04-29'],
['dhd',781,'prti10','2019-05-04'],
['dhd',781,'prti11','2019-03-01'],
['dhd',781,'prti12','2018-12-17'],
['dhd',781,'prti15','2018-11-21'],
['dhd',781,'prti17','2018-10-31'],
['dhd',781,'prti19','2018-09-5']
]
columns= ['client_id','value1','name1','a_date']
df= spark.createDataFrame(data,columns)
df2 = df.withColumn("year_val", F.year("a_date"))\
.withColumn("month_val", F.month("a_date"))\
.withColumn("year_month", F.year(F.col("a_date")) * 100 +
F.month(F.col("a_date")))\
.groupBy("client_id", "value1", "year_month")\
.agg(F.concat_ws(", ", F.collect_list("name1")).alias("init_list"))
df2.sort(F.col("value1").desc(), F.col("year_month").desc()).show()
w = Window().partitionBy("client_id", "value1")\
.orderBy("year_month")
df4 = df2.withColumn("a_rank", F.dense_rank().over(w))
df4.sort(F.col("value1"), F.col("year_month")).show()
month_range = 3
w = Window().partitionBy("client_id", "value1")\
.orderBy("a_rank")\
.rangeBetween(-(month_range),0)
df5 = df4.withColumn("last_" + str(month_range) + "_month", F.collect_list(F.col("init_list")).over(w))\
.orderBy("value1", "a_rank")
df6 = df5.sort(F.col("value1").desc(), F.col("year_month").desc())
df6.show(100,False)
【问题讨论】：
标签： sql python-3.x dataframe pyspark