Разделите данные на начало / окончание сеанса * значение sh

Question

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

usr_id, event_time, type_of_event, options
1, 2019-01-01 20:00:00, session_started, option1
1, 2019-01-01 20:00:01, session_continue, option2
1, 2019-01-01 20:00:02, session_finished
1, 2019-01-01 20:00:01, session_started, option3
1, 2019-01-01 20:00:02, session_finished

и в качестве вывода я хочу сделать что-то вроде этого:

usr_id, session_id, options
1, unique_session_id1, [option1, option2]
1, unique_session_id2, [option3]

Я пробовал некоторые оконные функции, но это не так помощь.

Answer 1

Я использовал concate для уникального session_id. Вы можете заменить его своим логином c для уникального session_id

scala> import org.apache.spark.sql.expressions.Window
scala> val W1  = Window.partitionBy("usr_id").orderBy("usr_id")
scala> val W2  = Window.partitionBy("usr_id").rowsBetween(Window.unboundedPreceding, Window.currentRow)
scala> df.show()
+------+------------------+----------------+-------+
|usr_id|        event_time|   type_of_event|options|
+------+------------------+----------------+-------+
|     1|2019-01-0120:00:00| session_started|option1|
|     1|2019-01-0120:00:01|session_continue|option2|
|     1|2019-01-0120:00:02|session_finished|   null|
|     1|2019-01-0120:00:01| session_started|option3|
|     1|2019-01-0120:00:02|session_finished|   null|
+------+------------------+----------------+-------+

scala> df.withColumn("temp", when(lag(col("type_of_event"),1).over(W1) === "session_finished", lit(1)).otherwise(lit(0)))
         .withColumn("temp1", sum("temp").over(W2))
         .groupBy("usr_id", "temp1")
         .agg(concat(concat_ws("_",collect_list(col("options"))),lit("_"), col("temp1")).alias("session_id"), collect_list(col("options")).alias("options"))
         .drop("temp1")
         .show()
+------+-----------------+------------------+
|usr_id|       session_id|           options|
+------+-----------------+------------------+
|     1|option1_option2_0|[option1, option2]|
|     1|        option3_1|         [option3]|
+------+-----------------+------------------+