Проблема с производительностью при использовании Window

Question

У меня есть данные временного ряда, я хочу получить интервал данных таким образом, чтобы, если 1 обнаружен в столбце детектора, это будет конец одного интервала и начало другого интервала. Я могу сделать это с groupby, но я хочу, чтобы альтернативный метод делал это из-за проблемы с производительностью при использовании groupby, а также одновременно определял интервал таким образом, если разница между временем для двух последовательных строк больше или равна 15.

Для простоты мы можем взять пример как ниже

time | detector
5    | 0
10   | 0
15   | 0
20   | 0
25   | 1
35   | 0
40   | 0
56   | 0
57   | 0
55   | 0
60   | 1
65   | 0
70   | 0
75   | 0
80   | 1
85   | 0

Вывод, который я хочу, это

interval 
[5,25]   
[25,60]  
[40,56]  
[60,80]  
[80,85]

обновление 1:

val wAll = Window.partitionBy(col("imei")).orderBy(col("time").asc)  

 val test= df.withColumn("lead_time", lead("time", 1, null).over(wAll)).withColumn("runningTotal", sum("detector").over(wAll))
        .groupBy("runningTotal").agg(struct(min("time"), max("lead_time")).as("interval"))

Это для расчета точек данных больше, чем равных 15 мин

val unreachable_df=df
    .withColumn("lag_time",lag("time", 1, null).over(wAll)) 
    .withColumn("diff_time",abs((col("time") - col("lag_time"))/60D))
   .withColumn("unreachable",when(col("diff_time")>=15.0,0).otherwise(1))
    .drop(col("diff_time"))
    .drop(col("lag_time"))
    .withColumn("runningTotal", sum("unreachable").over(wAll)) 
    .groupBy("runningTotal") 
    .agg(struct(min("time"), max("time")).as("interval"))
        .withColumn("diff_interval",abs((unix_timestamp(col("interval.col1"))-unix_timestamp(col("interval.col2"))))) 
        .filter(col("diff_interval")>0) .drop("diff_interval")
        .withColumn("type",lit("Unreachable")).drop("runningTotal")

Затем я объединил два кадра данных, чтобы получить вышеуказанный результат

val merged_df=test.union(unreachable_df).sort(col("interval.col1"))