您可以使用lead并将其与现有值相减，然后找到窗口的最大值，完成后，设置一个条件return 0 is max is>1，否则返回1

w = Window.partitionBy("group").orderBy(F.monotonically_increasing_id())

(foo.withColumn("Diff",F.lead("value").over(w)-F.col("value"))
 .withColumn("is_consecutive",F.when(F.max("Diff").over(w)>1,0).otherwise(1))
.drop("Diff")).show()

+-----+-----+--------------+
|group|value|is_consecutive|
+-----+-----+--------------+
|    a|    1|             1|
|    a|    2|             1|
|    a|    3|             1|
|    b|    4|             1|
|    b|    5|             1|
|    c|    2|             0|
|    c|    4|             0|
|    c|    5|             0|
+-----+-----+--------------+

你可以用 lag 要将值与前一行进行比较并检查它们是否连续，请使用 min 确定给定组中的所有行是否连续。

from pyspark.sql import functions as F, Window

df2 = df.withColumn(
    'consecutive', 
    F.coalesce(
        F.col('value') - F.lag('value').over(Window.partitionBy('group').orderBy('value')) == 1, 
        F.lit(True)
    ).cast('int')
).withColumn(
    'all_consecutive', 
    F.min('consecutive').over(Window.partitionBy('group'))
)

df2.show()
+-----+-----+-----------+---------------+
|group|value|consecutive|all_consecutive|
+-----+-----+-----------+---------------+
|    c|    2|          1|              0|
|    c|    4|          0|              0|
|    c|    5|          1|              0|
|    b|    4|          1|              1|
|    b|    5|          1|              1|
|    a|    1|          1|              1|
|    a|    2|          1|              1|
|    a|    3|          1|              1|
+-----+-----+-----------+---------------+