在 pyspark 中添加具有主题内排序交互日的列
Add column with within-subject rank-ordered interaction day in pyspark
我有一个大型 pyspark 数据框,其中包含多年的用户交互数据。有很多列,但对这道题有用的三个是 userid
、interaction_date
和 interaction_timestamp
。假设 table 中的给定用户有多个条目。
我需要编写一个函数来添加一个列,该列将指示 table 中给定客户的最新观察到的交互之前的天数。例如,对于输入 table
我想添加一个从该用户最近的交互日期开始计数的列(例如,最近的交互日期是 1,下一个最近的交互日期是 2,等等):
任何人都可以指导我以正确的方式做到这一点吗?
您可以使用 window function like dense_rank 来实现。看看下面的评论:
from pyspark.sql.window import Window
import pyspark.sql.functions as F
cols = ['userid','interaction_timestamp']
data =[( '1' ,'2018-01-02' ),
( '2' , '2018-01-03' ),
( '1' , '2018-01-03' ),
( '1' , '2018-01-04' ),
( '2' , '2018-01-02' ),
( '3' , '2018-01-03' ),
( '4' , '2018-01-03' )]
df = spark.createDataFrame(data, cols)
df = df.withColumn('interaction_timestamp', F.to_date('interaction_timestamp', 'yyyy-MM-dd'))
#rows with the same userid become part of the the same partition
#these partitions will be ordered descending by interaction_timestamp
w = Window.partitionBy('userid').orderBy(F.desc('interaction_timestamp'))
#dense_rank will assign a number to each row according to the defined order
df.withColumn("interaction_date_order", F.dense_rank().over(w)).show()
输出:
+------+---------------------+----------------------+
|userid|interaction_timestamp|interaction_date_order|
+------+---------------------+----------------------+
| 3| 2018-01-03| 1|
| 1| 2018-01-04| 1|
| 1| 2018-01-03| 2|
| 1| 2018-01-02| 3|
| 4| 2018-01-03| 1|
| 2| 2018-01-03| 1|
| 2| 2018-01-02| 2|
+------+---------------------+----------------------+
我有一个大型 pyspark 数据框,其中包含多年的用户交互数据。有很多列,但对这道题有用的三个是 userid
、interaction_date
和 interaction_timestamp
。假设 table 中的给定用户有多个条目。
我需要编写一个函数来添加一个列,该列将指示 table 中给定客户的最新观察到的交互之前的天数。例如,对于输入 table
我想添加一个从该用户最近的交互日期开始计数的列(例如,最近的交互日期是 1,下一个最近的交互日期是 2,等等):
任何人都可以指导我以正确的方式做到这一点吗?
您可以使用 window function like dense_rank 来实现。看看下面的评论:
from pyspark.sql.window import Window
import pyspark.sql.functions as F
cols = ['userid','interaction_timestamp']
data =[( '1' ,'2018-01-02' ),
( '2' , '2018-01-03' ),
( '1' , '2018-01-03' ),
( '1' , '2018-01-04' ),
( '2' , '2018-01-02' ),
( '3' , '2018-01-03' ),
( '4' , '2018-01-03' )]
df = spark.createDataFrame(data, cols)
df = df.withColumn('interaction_timestamp', F.to_date('interaction_timestamp', 'yyyy-MM-dd'))
#rows with the same userid become part of the the same partition
#these partitions will be ordered descending by interaction_timestamp
w = Window.partitionBy('userid').orderBy(F.desc('interaction_timestamp'))
#dense_rank will assign a number to each row according to the defined order
df.withColumn("interaction_date_order", F.dense_rank().over(w)).show()
输出:
+------+---------------------+----------------------+
|userid|interaction_timestamp|interaction_date_order|
+------+---------------------+----------------------+
| 3| 2018-01-03| 1|
| 1| 2018-01-04| 1|
| 1| 2018-01-03| 2|
| 1| 2018-01-02| 3|
| 4| 2018-01-03| 1|
| 2| 2018-01-03| 1|
| 2| 2018-01-02| 2|
+------+---------------------+----------------------+