Найдите количество строк в заданном временном окне в Pyspark

Question

У меня есть фрейм данных PySpark, небольшая часть которого приведена ниже:

+------+-----+-------------------+-----+
|  name| type|          timestamp|score|
+------+-----+-------------------+-----+
| name1|type1|2012-01-10 00:00:00|   11|
| name1|type1|2012-01-10 00:00:10|   14|
| name1|type1|2012-01-10 00:00:20|    2|
| name1|type1|2012-01-10 00:00:30|    3|
| name1|type1|2012-01-10 00:00:40|   55|
| name1|type1|2012-01-10 00:00:50|   10|
| name5|type1|2012-01-10 00:01:00|    5|
| name2|type2|2012-01-10 00:01:10|    8|
| name5|type1|2012-01-10 00:01:20|    1|
|name10|type1|2012-01-10 00:01:30|   12|
|name11|type3|2012-01-10 00:01:40|  512|
+------+-----+-------------------+-----+

Для выбранного временного окна (в качестве примера, скажем, 5 days) я хочу узнать, какмного значений score (скажем, num_values_week) существует для каждого name. То есть сколько значений score существует для name1 между 2012-01-10 - 2012-01-14, затем между 2012-01-15 - 2012-01-29 и т. Д. (И то же самое для всех других имен, таких как name2 и т. Д.)

Я хочу привести эту информацию в новый фрейм данных PySpark, в котором будут столбцы name, type, num_values_week. Как я могу это сделать?

В аналогичном вопросе , который я задавал ранее, я видел, как получить счет (баллов), когда выбирают интервалы в одну неделю. Но в этом вопросе я хочу узнать, как получить количество баллов, когда кто-либо выбирает любое настраиваемое значение во временном окне (например, 5 days или около того).

Любая помощь будет оценена.

Answer 1

Использование pd.Grouper с DataFrame.groupy:

#df['timestamp']=pd.to_datetime(df['timestamp']) #to convert to datetime
new_df=( df.groupby([pd.Grouper(key='timestamp',freq='5D'),'name'],sort=False)
          .score
          .count()
          .rename('num_values_week')
          .reset_index() )
print(new_df)

Выход

   timestamp    name  num_values_week
0 2012-01-10   name1                6
1 2012-01-10   name5                2
2 2012-01-10   name2                1
3 2012-01-10  name10                1
4 2012-01-10  name11                1

---

или GroupBy.resample:

new_df=( df.groupby('name',sort=False)
           .resample('5D',on='timestamp')
           .count()
           .score
           .rename('num_values_week')
           .reset_index() )
print(new_df)

Вывод

     name  timestamp  num_values_week
0   name1 2012-01-10                6
1   name5 2012-01-10                2
2   name2 2012-01-10                1
3  name10 2012-01-10                1
4  name11 2012-01-10                1

Если вы хотите создать новый столбец в исходном использовании dftransform:

df['num_values_week']=df.groupby([pd.Grouper(key='timestamp',freq='5D'),'name']).score.transform('count')
print(df)

      name   type           timestamp  score  num_values_week
0    name1  type1 2012-01-10 00:00:00     11                6
1    name1  type1 2012-01-10 00:00:10     14                6
2    name1  type1 2012-01-10 00:00:20      2                6
3    name1  type1 2012-01-10 00:00:30      3                6
4    name1  type1 2012-01-10 00:00:40     55                6
5    name1  type1 2012-01-10 00:00:50     10                6
6    name5  type1 2012-01-10 00:01:00      5                2
7    name2  type2 2012-01-10 00:01:10      8                1
8    name5  type1 2012-01-10 00:01:20      1                2
9   name10  type1 2012-01-10 00:01:30     12                1
10  name11  type3 2012-01-10 00:01:40    512                1