Tengo algunos datos que quiero agrupar por una determinada columna, luego agrego una serie de campos basados en una ventana de tiempo variable del grupo.

Aquí hay algunos datos de ejemplo:

df = spark.createDataFrame([Row(date='2016-01-01', group_by='group1', get_avg=5, get_first=1),
                            Row(date='2016-01-10', group_by='group1', get_avg=5, get_first=2),
                            Row(date='2016-02-01', group_by='group2', get_avg=10, get_first=3),
                            Row(date='2016-02-28', group_by='group2', get_avg=20, get_first=3),
                            Row(date='2016-02-29', group_by='group2', get_avg=30, get_first=3),
                            Row(date='2016-04-02', group_by='group2', get_avg=8, get_first=4)])

Quiero agrupar porgroup_by, luego cree ventanas de tiempo que comiencen en la fecha más temprana y se extiendan hasta que haya 30 días sin entrada para ese grupo. Una vez que hayan terminado esos 30 días, la próxima ventana comenzaría con la fecha de la siguiente fila que no se encontraba en la ventana anterior.

Luego quiero agregar, por ejemplo, obtener el promedio deget_avgy el primer resultado deget_first.

Entonces la salida para este ejemplo debería ser:

group_by    first date of window    get_avg  get_first
group1      2016-01-01              5        1
group2      2016-02-01              20       3
group2      2016-04-02              8        4

editar: lo siento, me di cuenta de que mi pregunta no se especificó correctamente. De hecho, quiero una ventana que termine después de 30 días de inactividad. He modificado la parte group2 del ejemplo en consecuencia.