(Pyspark — группировка по пользователю в течение определенного периода времени ⇐ Python

[Anonymous]

Я обрабатываю множество файлов журналов и хотел бы перенести эту работу в Spark, но не могу понять, как агрегировать события по временному окну на основе событий, как это легко сделать в Pandas.< /p>

Вот именно то, что я хочу сделать:

Для файла журнала (смоделированного ниже) пользователей, которые столкнулись с какое-то событие, я хотел бы вернуться на семь дней назад и вернуть статистические данные для всех остальных столбцов.

Вот это в 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

Я хотел бы сгруппировать этот DataFrame по user_id, а затем исключить из агрегирования любую строку, возраст которой превышает семь дней с момента «события».

В Пандах так:

Код: Выделить всё

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 

Подробнее здесь: [url]https://stackoverflow.com/questions/28707987/py-spark-group-by-user-over-a-window-of-time[/url]