Python Pyspark 根据条件重置累计和列

我有以下数据框:

+----+----------+-----+------+
|  id|      date|reset|cumsum|
+----+----------+-----+------+
|1001|2023-04-01|false|     0|
|1001|2023-04-02|false|     0|
|1001|2023-04-03|false|     1|
|1001|2023-04-04|false|     1|
|1001|2023-04-05| true|     4|
|1001|2023-04-06|false|     4|
|1001|2023-04-07|false|     4|
|1001|2023-04-08|false|    10|
|1001|2023-04-09| true|    10|
|1001|2023-04-10|false|    12|
|1001|2023-04-11|false|    13|
+----+----------+-----+------+

我正在try 创建一个新列，该列在特定日期的reset列为True时重置cumsum.

预期输出:

+----+----------+-----+------+----------+
|  id|      date|reset|cumsum|new_cumsum|
+----+----------+-----+------+----------+
|1001|2023-04-01|false|     0|         0|
|1001|2023-04-02|false|     0|         0|
|1001|2023-04-03|false|     1|         1|
|1001|2023-04-04|false|     1|         1|
|1001|2023-04-05| true|     4|         3|
|1001|2023-04-06|false|     4|         3|
|1001|2023-04-07|false|     4|         3|
|1001|2023-04-08|false|    10|         6|
|1001|2023-04-09| true|    10|         0|
|1001|2023-04-10|false|    12|         2|
|1001|2023-04-11|false|    13|         3|
+----+----------+-----+------+----------+

对上面的DF中正在发生的事情的解释:

1. 从4/01到4/04，没有重置标志，因此每个日期的新值都应该反映原始的值.
2. 在4/05，我们进行了第一次重置，如果原始值保持为1，则新值将为0.但因为它在那一天从1增加到4，所以新值就是差值(3)
3. 从4/05到4/07，原始值保持在4不变，因此新值应该保持在3不变.
4. 在4/08时，原来的值从4增加到10，因此新值应该是差值(6).
5. 在09年4月4日，我们进行了第二次重置，因为该值保持在10不变，该日期的新值应该是0.
6. 在4/10时，原始值从10增加到12，因此新值应该是差值(2).
7. 在4/11时，原始值从12增加到13，因此新值应为4/11时的值与最近重置日期4/09(3)时的值之间的差值.

Df代码:

df = spark.createDataFrame(
    [
        (1001, "2023-04-01", False, 0, 0),
        (1001, "2023-04-02", False, 0, 0),
        (1001, "2023-04-03", False, 1, 1),
        (1001, "2023-04-04", False, 1, 1),
        (1001, "2023-04-05", True, 4, 3),
        (1001, "2023-04-06", False, 4, 3),
        (1001, "2023-04-07", False, 4, 3),
        (1001, "2023-04-08", False, 10, 6),
        (1001, "2023-04-09", True, 10, 0),
        (1001, "2023-04-10", False, 12, 2),
        (1001, "2023-04-11", False, 13, 3),
    ],
    ["id", "date", "reset", "cumsum", "new_cumsum"],
)

我希望这是有道理的.解释起来有点棘手.提前感谢您的任何答复.