Как увеличить / уменьшить столбец на основе значения другого столбца в Pyspark

Question

У меня есть данные журнала аудита в Pyspark, которые я пытаюсь выяснить, сколько людей вошли в систему в любой момент времени.

Так, например, рабочий кадр, на котором я работаю, имеет следующую структуру

time    src_comp    auth_orient
1       C1          LogOn
2       C1          LogOn
2       C1          LogOn
3       C2          LogOn
3       C1          LogOff
4       C1          LogOn
4       C2          LogOff
5       C1          LogOn
6       C2          LogOn
6       C3          LogOff

Затем из своего кода я хочу создать еще один столбец, который показывает, сколько людей вошли в систему, как показано ниже.Я также не хочу, чтобы столбец loggedon когда-либо становился отрицательным, поэтому, если кто-то выходит из системы, которую я не отслеживал, я не хочу уменьшать столбец:

Ожидаемый результат:

time    src_comp    LoggedOn
1       C1          1
2       C1          2
3       C2          1
3       C1          1
3       C1          2
4       C2          0
5       C1          3
6       C2          1
6       C3          0

У меня есть следующий код, который я попробовал и получил ошибку.

auth_dataset.groupby(auth_dataset.time,auth_dataset.src_comp).agg(F.when(auth_dataset.strt=='LogOn',number+=1).otherwise(number-=1))

Answer 1

Попробуйте следующий код.Он суммирует все значения (1 и -1), а функции greatest(0,col) гарантируют, что он никогда не станет отрицательным.

auth_dataset.groupby('time','src_comp')\
.agg(F.sum(F.when(auth_dataset.auth_orient == 'LogOn', F.lit(1)).otherwise(F.lit(-1))).alias('LoggedOn'))\
.withColumn('LoggedOn', F.greatest(F.lit(0),'LoggedOn'))\
.sort('time','src_comp').show()