Определение окна на основе нескольких столбцов

Question

У меня есть следующий фрейм данных:

from pyspark.sql import functions as fn
from pyspark.sql.types import ArrayType, StringType, StructType, StructField, \
    MapType, FloatType, BooleanType, DoubleType, LongType, DateType, IntegerType
from pyspark.sql import SQLContext

sqlc = SQLContext(sc)

schema = StructType([
    StructField('key', StructType([
                StructField('event_name', StringType()),
                StructField('event_type', StringType())])), 
    StructField('uuid', StringType(), False),
    StructField('timestamp', LongType(), False),
    ])

df_test = sqlc.createDataFrame([
    ["1",['name_1', "type_1"], 1532428811 ], 
    ["1", ['name_2', "type_2"],1532428812], 
    ["1", ['name_2', "type_1"],1532428813], 
    ["2",['name_3', "type_3"], 1532428814],
    ["2",['name_4', "type_3"], 1532428815], 
    ["2",['name_5', "type_3"], 1532428880]], 
    schema= schema)

+----+----------------+----------+
|uuid|             key| timestamp|
+----+----------------+----------+
|   1|[name_1, type_1]|1532428811|
|   1|[name_2, type_2]|1532428812|
|   1|[name_2, type_1]|1532428813|
|   2|[name_3, type_3]|1532428814|
|   2|[name_4, type_3]|1532428815|
|   2|[name_5, type_3]|1532428880|
+----+----------------+----------+

Я бы хотел сгруппировать данные так, чтобы у меня была строка для каждого uuid, когда разница между любыми временными метками (из того же uuid)меньше порога отсечки, а типы события одинаковы.

Кроме того, я хочу, чтобы начальная и конечная отметки времени были минимальной и максимальной отметками времени строк, которые удовлетворяли условиям.

Выходные данные должны выглядеть следующим образом, при условии, что пороговое значение отсечки равно 50 (обратите внимание, что последняя строка ниже отделяется от последней до последней, поскольку разница временных меток> 50):

+----+-----------------------------------------------------------+-----------+-----------+
|uuid|      key                                                  | start     | end       |
+----+-----------------------------------------------------------+-----------+-----------+
|   1|[[name_1|type_1, 1532428811], [name_2|type_1, 1532428813]] |1532428811 | 1532428813|
|   1|[[name_2|type_2, 1532428812]]                              |1532428812 | 1532428812|
|   2|[[name_3|type_3, 1532428814], [name_4|type_3, 1532428815]] |1532428814 | 1532428815|
|   2|[[name_5|type_3, 1532428880]]                              |1532428880 | 1532428880|
+----+-----------------------------------------------------------+-----------+-----------+