您还可以选择最后一个索引的 Bool 将值转换为变量，并在表达式中使用它，如下所示：

from pyspark.sql import functions as F

# or actually, if the index is always = 4 you can just filter without ordering

last_bool = df.orderBy(F.desc("index")).limit(1).select("Bool").first().Bool

df2 = df.withColumn(
    'New_Bool',
    F.when(
        F.col('index') == 1, 
        F.lit(last_bool) & F.col("New_bool") # keep New_bool or update to false if last is false  
    ).otherwise(
        F.col("New_bool") & F.col("Bool") # keep New_bool or update to false if Bool is false 
    )
)

df2.show()

# +-----+-----+--------+

# |index| Bool|New_Bool|

# +-----+-----+--------+

# |    1| true|   false|

# |    2| true|    true|

# |    3|false|   false|

# |    4|false|   false|

# +-----+-----+--------+

你可以用 lead 当index=1时检查index=4的行。对于其他索引，只需保留bool列。

from pyspark.sql import functions as F, Window

df2 = df.withColumn(
    'New_Bool',
    F.when(
        (F.col('index') == 1) & 
        (F.lead('Bool', 3).over(Window.orderBy('index')) == False),
        False
    ).otherwise(F.col('Bool'))
)

df2.show()
+-----+-----+--------+
|index| Bool|New_Bool|
+-----+-----+--------+
|    1| true|   false|
|    2| true|    true|
|    3|false|   false|
|    4|false|   false|
+-----+-----+--------+

另一种方法是交叉连接到索引为4的行的bool：

df2 = df.crossJoin(
    df.filter('index = 4')
      .select(F.col('Bool').alias('id_4'))
).withColumn(
    'New_Bool',
    F.when(
        (F.col('index') == 1) & (F.col('id_4') == False), 
        False
    ).otherwise(F.col('Bool'))
).drop('id_4')

如果4总是最后一个索引，那么可以使用 last :

df2 = df.withColumn(
    'New_Bool',
    F.when(
        (F.col('index') == 1) & 
        (F.last('Bool').over(Window.orderBy('index').rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)) == False),
        False
    ).otherwise(F.col('Bool'))
)