我认为这个逻辑是可能的。在2周范围内计数bool = 1，如果小于1，则为目标值。

from pyspark.sql import functions as f
from pyspark.sql import Window

w = Window.partitionBy('ID').orderBy('timestamp').rangeBetween(0, 14 * 86400)
​
df.withColumn('timestamp', f.unix_timestamp('date', 'y-M-d')) \
  .withColumn('bool_in_2weeks', f.count(f.when(f.col('bool') == f.lit(1), True)).over(w)) \
  .withColumn('D', f.expr('bool = 1 and bool_in_2weeks > 1').cast('int')) \
  .show(truncate=False)
+-------+----+----------+----------+--------------+---+
|ID     |bool|date      |timestamp |bool_in_2weeks|D  |
+-------+----+----------+----------+--------------+---+
|1092829|0   |2019-10-29|1572307200|0             |0  |
|1092829|1   |2019-12-14|1576281600|1             |0  |
|1092829|1   |2020-07-15|1594771200|2             |1  |
|1092829|1   |2020-07-17|1594944000|1             |0  |
|1092829|0   |2020-07-19|1595116800|0             |0  |
|1092829|1   |2020-08-15|1597449600|1             |0  |
|1092829|1   |2020-09-10|1599696000|1             |0  |
+-------+----+----------+----------+--------------+---+

字符串

我以你为例做了一些测试。这是我的想法

df = spark.createDataFrame(
    [
        {"ID": 1092829, "date": "2019-10-29", "bool": 0},
        {"ID": 1092829, "date": "2019-12-14", "bool": 1},
        {"ID": 1092829, "date": "2020-07-15", "bool": 1},
        {"ID": 1092829, "date": "2020-07-17", "bool": 1},
        {"ID": 1092829, "date": "2020-07-19", "bool": 0},
        {"ID": 1092829, "date": "2020-08-15", "bool": 1},
        {"ID": 1092829, "date": "2020-09-10", "bool": 1},
    ]
)

字符串
对于您的情况，使用Window函数既简单又合适。该窗口是通过根据“ID”列对DataFrame进行分区来构造的。这将保证在应用滞后时，不会有两个不同的“ID”值将它们的日期/布尔值混淆。在窗口中，您需要为datediff操作按升序对日期进行排序。

from pyspark.sql.window import Window
import pyspark.sql.functions as F

window = Window.partitionBy("ID").orderBy(F.col("ID").asc(), F.col("date").asc())

transformed_df = df.withColumn(
    "D",
    F.coalesce(
        # Apply lag on "date" by -1 (bring next row's value to current row's level) to do `datediff` for the 2-week condition.
        (F.datediff(F.lag("date", -1, None).over(window), F.col("date")) < 14)
        # Apply lag on "bool" by -1 to compare, for a given row, if the next row has a 1 or not.
        & (
            F.lag("bool", -1, None).over(window) == F.lit(1)
        ),  # The comparisons and the "&" produce a boolean column
        F.lit(
            False
        ),  # For the case when the last `date` value for a given "ID" is NULL because of the lag, we set it to False by default.
    ).cast(
        "integer"
    ),  # Cast the boolean result into an integer for a binary value.
)

transformed_df.show()

型
结果 Dataframe ：

#+-------+----------+----+---+
#|     ID|      date|bool|  D|
#+-------+----------+----+---+
#|1092829|2019-10-29|   0|  0|
#|1092829|2019-12-14|   1|  0|
#|1092829|2020-07-15|   1|  1|
#|1092829|2020-07-17|   1|  0|
#|1092829|2020-07-19|   0|  0|
#|1092829|2020-08-15|   1|  0|
#|1092829|2020-09-10|   1|  0|
#+-------+----------+----+---+

型
我希望这能回答你的问题。