您可以创建一个滞后列，然后在两列之间使用按位OR。

from pyspark.sql.functions import when, lag, col, monotonically_increasing_id
from pyspark.sql.window import Window

df = spark.createDataFrame(
  [[1000,1000],
  [2600,2600],
  [3600,3600],
  [3600,4500],
  [3600,4500]],['col1','col2']
)

df = df.withColumn('flag', when((df['col1'] <= 4000) & (df['col2'] <= 4000), 1).otherwise(0))

df = df.withColumn('idx', monotonically_increasing_id())

w = Window().partitionBy().orderBy(col('idx'))
df = df.withColumn('lag', lag('flag', 1).over(w))
df = df.fillna(0, subset='lag')

df = df.withColumn('flag', df.flag.bitwiseOR(df.lag))
df.select('col1','col2','flag').show()

输出量

+----+----+----+
|col1|col2|flag|
+----+----+----+
|1000|1000|   1|
|2600|2600|   1|
|3600|3600|   1|
|3600|4500|   1|
|3600|4500|   0|
+----+----+----+