首先，您要创建符合条件的分组。要创建它，一般的提示是创建一个标志，当你想要分离一个组时，标志为1，当你想要合并到前一个组时，标志为0。然后，cumsum在这个标志将导致你想要的分组。
你的条件是

from pyspark.sql import functions as F
# covert Time to timestamp
df = df.withColumn('timestamp', F.to_timestamp('Time', 'M/d/yyyy H:mm'))

w = Window.partitionBy('Name').orderBy('timestamp')

# If previous Flag is different from current Flag
F.lag('Flag').over(w) != F.col('Flag'))

# OR previous timestamp is more than 1 minute ago
| (((F.col('timestamp').cast('long') - F.lag('timestamp').over(w).cast('long')) / 60) > 1)

在这些条件下，将分组创建为grp列，并使用该列进行聚合。

w = Window.partitionBy('Name').orderBy('timestamp')
df = (df.withColumn('timestamp', F.to_timestamp('Time', 'M/d/yyyy H:mm'))
      .withColumn('grp', (F.lag('Flag').over(w).isNull() 
                          | (F.lag('Flag').over(w) != F.col('Flag'))
                          | (((F.col('timestamp').cast('long') - F.lag('timestamp').over(w).cast('long')) / 60) > 1)).cast('int'))
      .withColumn('grp', F.sum('grp').over(w))
      .groupby('Name', 'grp')
      .agg(F.min('Time').alias('Start'), F.max('Time').alias('End'), F.first('Flag').alias('Flag')))