python之(Pyspark之在一段时间内按用户分组

我正在处理大量日志文件，我想将工作转移到 Spark，但我不知道如何像在 Pandas 中那样轻松地在基于事件的时间窗口内聚合事件。

这正是我想要做的:

对于经历过某些事件的用户的日志文件(下面模拟)，我想及时返回 7 天，并返回所有其他列的聚合。

这是在 Pandas 中。有什么想法可以将其移植到 PySpark 吗？

import pandas as pd 
df = pd.DataFrame({'user_id':[1,1,1,2,2,2], 'event':[0,1,0,0,0,1], 'other':[12, 20, 16, 84, 11, 15] , 'event_date':['2015-01-01 00:02:43', '2015-01-04 00:02:03', '2015-01-10 00:12:26', '2015-01-01 00:02:43', '2015-01-06 00:02:43', '2015-01-012 18:10:09']}) 
df['event_date'] = pd.to_datetime(df['event_date']) 
df 

给予:

    event  event_date           other  user_id 
0   0      2015-01-01 00:02:43  12     1 
1   1      2015-01-04 00:02:03  20     1 
2   0      2015-01-10 00:12:26  16     1 
3   0      2015-01-01 00:02:43  84     2 
4   0      2015-01-06 00:02:43  11     2 
5   1      2015-01-12 18:10:09  15     2 

我想按 user_id 对这个 DataFrame 进行分组，然后从聚合中排除该行距离“事件”超过 7 天的任何行。

在 Pandas 中，像这样:

def f(x): 
    # Find event 
    win = x.event == 1 
 
    # Get the date when event === 1 
    event_date = list(x[win]['event_date'])[0] 
 
    # Construct the window 
    min_date = event_date - pd.DateOffset(days=7)  
 
    # Set x to this specific date window 
    x = x[(x.event_date > min_date) & (x.event_date <= event_date)] 
 
    # Aggregate other 
    x['other'] = x.other.sum() 
 
    return x[win] #, x[z]]) 
 
 
df.groupby(by='user_id').apply(f).reset_index(drop=True) 

提供所需的输出(每个用户一行，其中 event_date 对应于 event==1):

    event   event_date          other   user_id 
0   1       2015-01-04 00:02:03 32      1 
1   1       2015-01-12 18:10:09 26      2 

有人知道在 Spark 中从哪里开始得到这个结果吗？

请您参考如下方法:

相当 SQLish 但你可以这样做:

from pyspark.sql.functions import sum, col, udf 
from pyspark.sql.types import BooleanType 
 
# With raw SQL you can use datediff but it looks like it is not 
# available as a function yet 
def less_than_n_days(n):                                                        
    return udf(lambda dt1, dt2: 0 <= (dt1 - dt2).days < n, BooleanType()) 
 
# Select only events 
events = df.where(df.event == 1).select( 
        df.event_date.alias("evd"), df.user_id.alias("uid")) 
 
(events 
    .join(df, (events.uid == df.user_id) & (events.evd >= df.event_date)) 
    .where(less_than_n_days(7)(col("evd"), col("event_date"))) 
    .groupBy("evd", "user_id")  
    .agg(sum("other").alias("other")) 
    .withColumnRenamed("evd", "event_date")) 

很遗憾，我们不能在 join 子句中包含 less_than_n_days，因为 udf 只能访问单个表中的列。由于它不适用于内置的 datediff，您可能更喜欢这样的原始 SQL:

df.registerTempTable("df") 
events.registerTempTable("events") 
 
sqlContext.sql(""" 
    SELECT evd AS event_date, user_id, SUM(other) AS other 
    FROM df JOIN events ON 
        df.user_id = events.uid AND 
        datediff(evd, event_date) BETWEEN 0 AND 6 
    GROUP by evd, user_id""")