Как найти максимальную и минимальную метку времени, когда значение становится ниже минимального порога в pyspark?

В Pandas вы можете сделать это:

data = {
    "time_is_seconds": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12],
    "value": [4.5, 4, 3, 5, 6, 7, 6, 5, 4.5, 4.2, 3, 3.5],
}

df = pd.DataFrame(data)

m = df["value"].lt(5)
g = m.ne(m.shift(1)).cumsum()

df.loc[m, "min_time"] = df.groupby(g)["time_is_seconds"].transform(min)
df.loc[m, "max_time"] = df.groupby(g)["time_is_seconds"].transform(max)

    time_is_seconds  value  min_time  max_time
0                 1    4.5       1.0       3.0
1                 2    4.0       1.0       3.0
2                 3    3.0       1.0       3.0
3                 4    5.0       NaN       NaN
4                 5    6.0       NaN       NaN
5                 6    7.0       NaN       NaN
6                 7    6.0       NaN       NaN
7                 8    5.0       NaN       NaN
8                 9    4.5       9.0      12.0
9                10    4.2       9.0      12.0
10               11    3.0       9.0      12.0
11               12    3.5       9.0      12.0

У меня нет опыта работы с Pyspark, но, возможно, это поможет вам, если вы будете следовать той же логике.

Вот способ:

m = df['value'].ge(5)
g = df.groupby(m.cumsum().mask(m))['time_is_seconds']

df.assign(**pd.concat([g.transform('min'),g.transform('max')],keys = ['min_time','max_time'],axis=1))

Выход:

    time_is_seconds  value  min_time  max_time
0                 1    4.5       1.0       3.0
1                 2    4.0       1.0       3.0
2                 3    3.0       1.0       3.0
3                 4    5.0       NaN       NaN
4                 5    6.0       NaN       NaN
5                 6    7.0       NaN       NaN
6                 7    6.0       NaN       NaN
7                 8    5.0       NaN       NaN
8                 9    4.5       9.0      12.0
9                10    4.2       9.0      12.0
10               11    3.0       9.0      12.0
11               12    3.5       9.0      12.0

Основываясь на предыдущих ответах, я применил аналогичную логику и получил ответ:

w=orderBy('time_in_seconds')
w1=Window.partitionBy('cum_sum_ne')
df1=df\
.withColumn( 'bool_less_than_5',col('value')<5)\
.withColumn('lag_bool_less_than_5',lag(col('bool_less_than_5')).over(w))\          
.withColumn('ne_bool_less_than_5',when(col('bool_less_than_5')==col('lag_bool_less_than_5'),0).otherwise(1))\
.withColumn('cum_sum_ne',sum(col('ne_bool_less_than_5')).over(w))\  
.withColumn('min_time',min(col('time_in_seconds')).over(w1))\
.withColumn('max_time',max(col('time_in_seconds')).over(w1))\
.withColumn('min_time',when(col('bool_less_than_5'),col('min_time')))\
.withColumn('max_time',when(col('bool_less_than_5'),col('max_time')))