我正在使用Pyspark处理以下数据帧，因此它可以适合营销归因模型：
| user_id | 时间戳 | 活动活动名称 | |
|---|---|---|---|
| akalsds124 | 2022-01-01 10:00 | 点击 | 假日活动空 |
| akalsds124 | 2021-12-31 09:00 | 点击假日活动 | >NULL |
| akalsds124 | 2022-01-13 15:59 | 点击 | X活动 | 空
| akalsds124 | 2022-01-10 16:32 | 点击 | 超级活动>>空 |
| akalsds124 | 2022-01-05 22:12 | 点击假日活动 | >空 |
| akalsds124 | 2022-01-30 20:55 | 事件 | 空>购买 |
| akalsds124 | 2022-01-30 22:10 | 事件 | 空购买 |
| akalsds124 | 2022-01-31 10:13 | 事件 | 空>购买 |
| akalsds124 | 2022-02-03 04:55 | 点击 | T8活动空|
| akalsds124 | 2022-02-07 17:30 | 点击 | Y活动空 |
| akalsds124 | 2022-02-12 22:37 | 事件空 | 购买 |
| akalsds124 | 2022-03-31 18:19 | 点击 | U9活动空|
| akalsds124 | 2022-04-02 23:08 | 点击 | 第二次活动空|
| akalsds124 | 2022-03-02 07:00 | 点击T8活动 | 空 |
| ijnbmshs33 | 2022-06-03 17:01 | 点击 | 大型活动空|
| ijnbmshs33 | 2022-05-03 10:31 | 点击新活动 | >>空 |
| ijnbmshs33 | 2022-05-20 17:01 | 点击 | 大型活动空
我认为诀窍是在有界窗口上使用collect_list函数。下面的代码可能是你答案的第一部分
window = W.partitionBy('user_id').orderBy('unixTime').rangeBetween(-3600*24*30, 0)
path_df = (
df
.withColumn('timestamp', F.col('timestamp').cast('timestamp'))
.withColumn('unixTime', F.unix_timestamp('timestamp'))
.withColumn('pathList', F.collect_list('campaign').over(window))
.filter(F.col('event_name') == 'purchase')
)
path_df.sort('timestamp').show(truncate=False)