PySpark: как рассчитать дни между выполнением последнего условия (положительное или отрицательное)

Question

Current DF (фильтр по одному userId, флаг 1, если убыток> 0, -1, если <= 0): </p>

display(df):
+------+----------+---------+----+
|  user|Date      |RealLoss |flag|
+------+----------+---------+----+
|100364|2019-02-01|    -16.5|   1|
|100364|2019-02-02|     73.5|  -1|
|100364|2019-02-03|       31|  -1|
|100364|2019-02-09|     -5.2|   1|
|100364|2019-02-10|    -34.5|   1|
|100364|2019-02-13|     -8.1|   1|
|100364|2019-02-18|     5.68|  -1|
|100364|2019-02-19|     5.76|  -1|
|100364|2019-02-20|     9.12|  -1|
|100364|2019-02-26|      9.4|  -1|
|100364|2019-02-27|    -30.6|   1|
+----------+------+---------+----+

желаемый результат df должен показывать количество дней с момента lastwin ('RecencyLastWin') и с момента lastloss ('RecencyLastLoss')

display(df):
+------+----------+---------+----+--------------+---------------+
|  user|Date      |RealLoss |flag|RecencyLastWin|RecencyLastLoss|
+------+----------+---------+----+--------------+---------------+
|100364|2019-02-01|    -16.5|   1|          null|           null|
|100364|2019-02-02|     73.5|  -1|             1|           null|
|100364|2019-02-03|       31|  -1|             2|              1|     
|100364|2019-02-09|     -5.2|   1|             8|              6|
|100364|2019-02-10|    -34.5|   1|             1|              7|
|100364|2019-02-13|     -8.1|   1|             1|             10|
|100364|2019-02-18|     5.68|  -1|             5|             15|         
|100364|2019-02-19|     5.76|  -1|             6|              1|
|100364|2019-02-20|     9.12|  -1|             7|              1|
|100364|2019-02-26|      9.4|  -1|            13|              6|
|100364|2019-02-27|    -30.6|   1|            14|              1|
+----------+------+---------+----+--------------+---------------+

Мой подход был следующим:

from pyspark.sql.window import Window

w = Window.partitionBy("userId", 'PlayerSiteCode').orderBy("EventDate")

last_positive = check.filter('flag = "1"').withColumn('last_positive_day' , F.lag('EventDate').over(w))
last_negative = check.filter('flag = "-1"').withColumn('last_negative_day' , F.lag('EventDate').over(w))

finalcheck = check.join(last_positive.select('userId', 'PlayerSiteCode', 'EventDate', 'last_positive_day'), ['userId', 'PlayerSiteCode', 'EventDate'], how = 'left')\
                  .join(last_negative.select('userId', 'PlayerSiteCode', 'EventDate', 'last_negative_day'), ['userId', 'PlayerSiteCode', 'EventDate'], how = 'left')\
                  .withColumn('previous_date_played'         , F.lag('EventDate').over(w))\
                  .withColumn('last_positive_day_count', F.datediff(F.col('EventDate'), F.col('last_positive_day')))\
                  .withColumn('last_negative_day_count', F.datediff(F.col('EventDate'), F.col('last_negative_day')))

затем я попытался добавить (несколько попыток ..) но не смог «идеально» вернуть то, что я хочу.

finalcheck = finalcheck.withColumn('previous_last_pos'           , F.last('last_positive_day_count', True).over(w2))\
                        .withColumn('previous_last_neg'           , F.last('last_negative_day_count', True).over(w2))\
                        .withColumn('previous_last_pos_date'      , F.last('last_positive_day', True).over(w2))\
                        .withColumn('previous_last_neg_date'      , F.last('last_negative_day', True).over(w2))\
                        .withColumn('recency_last_positive'       , F.datediff(F.col('EventDate'), F.col('previous_last_pos_date')))\
                        .withColumn('day_since_last_negative_v1'  , F.datediff(F.col('EventDate'), F.col('previous_last_neg_date')))\
                        .withColumn('days_off'                    , F.datediff(F.col('EventDate'), F.col('previous_date_played')))\
                        .withColumn('recency_last_negative'       , F.when((F.col('day_since_last_negative_v1').isNull()), F.col('days_off')).otherwise(F.col('day_since_last_negative_v1')))\
                        .withColumn('recency_last_negative_v2'    , F.when((F.col('last_negative_day').isNull()), F.col('days_off')).otherwise(F.col('day_since_last_negative_v1')))\
                        .withColumn('recency_last_positive_v2'    , F.when((F.col('last_positive_day').isNull()), F.col('days_off')).otherwise(F.col('recency_last_positive')))

Есть предложения / советы? (Я нашел аналогичный вопрос, но не понял, как подать заявку в моем случае c): Как рассчитать количество дней между выполнением последнего условия?

Answer 1

Вот моя попытка.

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

Наконец, объедините эти результаты в столбец.

from pyspark.sql.functions import lag, col, sum
from pyspark.sql import Window

w1 = Window.orderBy('Date')
w2 = Window.partitionBy('groupLossCheck').orderBy('Date')
w3 = Window.partitionBy('groupWinCheck').orderBy('Date')

df2 = df.withColumn('lastFlag', lag('flag', 1).over(w1)) \
  .withColumn('lastDate', lag('Date', 1).over(w1)) \
  .withColumn('dateDiff', expr('datediff(Date, lastDate)')) \
  .withColumn('consecutiveLoss', expr('if(flag =  1 or lastFlag =  1, 0, 1)')) \
  .withColumn('consecutiveWin' , expr('if(flag = -1 or lastFlag = -1, 0, 1)')) \
  .withColumn('groupLossCheck', sum('consecutiveLoss').over(w1)) \
  .withColumn('groupWinCheck' , sum('consecutiveWin' ).over(w1)) \
  .withColumn('daysLastLoss', sum(when((col('consecutiveLoss') == 0) & (col('groupLossCheck') != 0), col('dateDiff'))).over(w2)) \
  .withColumn('daysLastwin' , sum(when((col('consecutiveWin' ) == 0) & (col('groupWinCheck' ) != 0), col('dateDiff'))).over(w3)) \
  .withColumn('lastLoss', expr('if(lastFlag = -1, datediff, null)')) \
  .withColumn('lastWin' , expr('if(lastFlag =  1, dateDiff, null)')) \
  .withColumn('RecencyLastLoss', coalesce('lastLoss', 'daysLastLoss')) \
  .withColumn('RecencyLastWin',  coalesce('lastWin' , 'daysLastwin' )) \
  .orderBy('Date')

df2.show(11, False)

+------+----------+--------+----+--------+----------+--------+---------------+--------------+--------------+-------------+------------+-----------+--------+-------+---------------+--------------+
|user  |Date      |RealLoss|flag|lastFlag|lastDate  |dateDiff|consecutiveLoss|consecutiveWin|groupLossCheck|groupWinCheck|daysLastLoss|daysLastwin|lastLoss|lastWin|RecencyLastLoss|RecencyLastWin|
+------+----------+--------+----+--------+----------+--------+---------------+--------------+--------------+-------------+------------+-----------+--------+-------+---------------+--------------+
|100364|2019-02-01|-16.5   |1   |null    |null      |null    |0              |1             |0             |1            |null        |null       |null    |null   |null           |null          |
|100364|2019-02-02|73.5    |-1  |1       |2019-02-01|1       |0              |0             |0             |1            |null        |1          |null    |1      |null           |1             |
|100364|2019-02-03|31.0    |-1  |-1      |2019-02-02|1       |1              |0             |1             |1            |null        |2          |1       |null   |1              |2             |
|100364|2019-02-09|-5.2    |1   |-1      |2019-02-03|6       |0              |0             |1             |1            |6           |8          |6       |null   |6              |8             |
|100364|2019-02-10|-34.5   |1   |1       |2019-02-09|1       |0              |1             |1             |2            |7           |null       |null    |1      |7              |1             |
|100364|2019-02-13|-8.1    |1   |1       |2019-02-10|3       |0              |1             |1             |3            |10          |null       |null    |3      |10             |3             |
|100364|2019-02-18|5.68    |-1  |1       |2019-02-13|5       |0              |0             |1             |3            |15          |5          |null    |5      |15             |5             |
|100364|2019-02-19|5.76    |-1  |-1      |2019-02-18|1       |1              |0             |2             |3            |null        |6          |1       |null   |1              |6             |
|100364|2019-02-20|9.12    |-1  |-1      |2019-02-19|1       |1              |0             |3             |3            |null        |7          |1       |null   |1              |7             |
|100364|2019-02-26|9.4     |-1  |-1      |2019-02-20|6       |1              |0             |4             |3            |null        |13         |6       |null   |6              |13            |
|100364|2019-02-27|-30.6   |1   |-1      |2019-02-26|1       |0              |0             |4             |3            |1           |14         |1       |null   |1              |14            |
+------+----------+--------+----+--------+----------+--------+---------------+--------------+--------------+-------------+------------+-----------+--------+-------+---------------+--------------+

df2.select(*df.columns, 'RecencyLastLoss', 'RecencyLastWin').show(11, False)

+------+----------+--------+----+---------------+--------------+
|user  |Date      |RealLoss|flag|RecencyLastLoss|RecencyLastWin|
+------+----------+--------+----+---------------+--------------+
|100364|2019-02-01|-16.5   |1   |null           |null          |
|100364|2019-02-02|73.5    |-1  |null           |1             |
|100364|2019-02-03|31.0    |-1  |1              |2             |
|100364|2019-02-09|-5.2    |1   |6              |8             |
|100364|2019-02-10|-34.5   |1   |7              |1             |
|100364|2019-02-13|-8.1    |1   |10             |3             |
|100364|2019-02-18|5.68    |-1  |15             |5             |
|100364|2019-02-19|5.76    |-1  |1              |6             |
|100364|2019-02-20|9.12    |-1  |1              |7             |
|100364|2019-02-26|9.4     |-1  |6              |13            |
|100364|2019-02-27|-30.6   |1   |1              |14            |
+------+----------+--------+----+---------------+--------------+