Pyspark: función de ventana personalizada

Actualmente estoy tratando de extraer series de ocurrencias consecutivas en un marco de datos PySpark y ordenarlas / clasificarlas como se muestra a continuación (por conveniencia, he ordenado el marco de datos inicial poruser_id ytimestamp):

df_ini

+-------+--------------------+------------+
|user_id|     timestamp      |  actions   |
+-------+--------------------+------------+
| 217498|           100000001|    'A'     |
| 217498|           100000025|    'A'     |
| 217498|           100000124|    'A'     |
| 217498|           100000152|    'B'     |
| 217498|           100000165|    'C'     |
| 217498|           100000177|    'C'     |
| 217498|           100000182|    'A'     |
| 217498|           100000197|    'B'     |
| 217498|           100000210|    'B'     |
| 854123|           100000005|    'A'     |
| 854123|           100000007|    'A'     |
| etc.

a :

expected df_transformed

+-------+------------+------------+------------+
|user_id|  actions   | nb_of_occ  |    order   |
+-------+------------+------------+------------+
| 217498|    'A'     |      3     |     1      |
| 217498|    'B'     |      1     |     2      |
| 217498|    'C'     |      2     |     3      |
| 217498|    'A'     |      1     |     4      |
| 217498|    'B'     |      2     |     5      |
| 854123|    'A'     |      2     |     1      |
| etc.

Supongo que tengo que usar una función de ventana inteligente que particione la tabla por user_id y accionespero solo cuando estas acciones son consecutivas en el tiempo ! Lo cual no puedo entender cómo hacerlo ...

Si alguien encontrara este tipo de transformación en PySpark antes, ¡me alegraría tener una pista!

Salud