Spark SQL using a window - collect data from rows after current row based on a column condition

Question

I have a spark DataFrame (in Scala) like this:

+---------+-------------+------+---------+------------+
|  user_id|      item_id|  mood|     time|sessionBegin|
+---------+-------------+------+---------+------------+
|        1|            A| Happy|        0|           0|
|        1|            B| Happy|        1|           0|
|        1|            C| Happy|        3|           0|
|        1|            D| Happy|        5|           0|
|        1|            C| Happy|        6|           0|
|        1|            D|   Sad|        6|           0|
|        1|            C|   Sad|       10|           0|
|        1|            A| Happy|       28|           0|
|        1|            E| Happy|       35|           0|
|        1|            E|   Sad|       60|           0|
|        2|            F| Happy|        6|           6|
|        2|            E| Happy|       17|           6|
|        2|            D| Happy|       20|           6|
|        2|            D|   Sad|       21|           6|
|        2|            E| Happy|       27|           6|
|        2|            G| Happy|       37|           6|
|        2|            H| Happy|       39|           6|
|        2|            G|   Sad|       45|           6|
+---------+-------------+------+---------+------------+

I have defined a Window over columns (user_id, sessionBegin) and ordered by time

val window = Window.partitionBy("user_id","sessionBegin").orderBy("time")

Now I want to add a column result which :

1) Checks if the Mood is Happy then collect all the item_id ONLY AFTER the current row && mood = Sad. Else if Mood is sad: put empty array.

2) This has to be over the window I specified above. (For eg. this dataframe has two windows -> first is (user_id = 1 , sessionBegin = 0) and second is (user_id = 2, sessionBegin = 6)

Hence the resulting DF will be :

+---------+-------------+------+---------+------------+---------+
|  user_id|      item_id|  mood|     time|sessionBegin|   result|
+---------+-------------+------+---------+------------+---------+
|        1|            A| Happy|        0|           0|  [D,C,E]|
|        1|            B| Happy|        1|           0|  [D,C,E]|
|        1|            C| Happy|        3|           0|  [D,C,E]|
|        1|            D| Happy|        5|           0|  [D,C,E]|
|        1|            C| Happy|        6|           0|  [D,C,E]|
|        1|            D|   Sad|        6|           0|       []|
|        1|            C|   Sad|       10|           0|       []|
|        1|            A| Happy|       28|           0|      [E]|
|        1|            E| Happy|       35|           0|      [E]|
|        1|            E|   Sad|       60|           0|       []|
|        2|            F| Happy|        6|           6|    [D,G]|
|        2|            E| Happy|       17|           6|    [D,G]|
|        2|            D| Happy|       20|           6|    [D,G]|
|        2|            D|   Sad|       21|           6|       []|
|        2|            E| Happy|       27|           6|      [G]|
|        2|            G| Happy|       37|           6|      [G]|
|        2|            H| Happy|       39|           6|      [G]|
|        2|            G|   Sad|       45|           6|       []|
+---------+-------------+------+---------+------------+---------+

I used collect_set method using when..otherwise over the window, but I was unable to figure out two things:

How to ONLY consider the rows after the current row
For all the rows with mood=Happy how to collect_set the item_id ONLY when mood=sad?

Any points to how to approach this?

You can use row frame in window spec by calling rowsBetween function. See https://databricks.com/blog/2015/07/15/introducing-window-functions-in-spark-sql.html — sramalingam24, Nov 29 '18 at 18:06
Any reason why "B" is not included in the very first row?.. your explanation and expected results are not matching... — stack0114106, Nov 30 '18 at 09:04
@stack0114106 "B" should not be included in the `result` column, because I want to collect only those item_id which have `Mood= Sad` . And I have edited the expected results now. The first row should contain [D,C,E]. — Jeet Banerjee, Nov 30 '18 at 10:27

stack0114106 · Accepted Answer · 2018-12-31T23:58:19.157

I'm not able to give row between next row and next row following the end of the partition. So I used current row and unbounded following, then removed the first Array element using udf. I have used all - spark.sql, udf and df manipulations.. check this out

val df = Seq((1,"A","Happy","0","0"),(1,"B","Happy","1","0"),(1,"C","Happy","3","0"),(1,"D","Happy","5","0"),(1,"C","Happy","6","0"),(1,"D","Sad","6","0"),(1,"C","Sad","10","0"),(1,"A","Happy","28","0"),(1,"E","Happy","35","0"),(1,"E","Sad","60","0"),(2,"F","Happy","6","6"),(2,"E","Happy","17","6"),(2,"D","Happy","20","6"),(2,"D","Sad","21","6"),(2,"E","Happy","27","6"),(2,"G","Happy","37","6"),(2,"H","Happy","39","6"),(2,"G","Sad","45","6")).toDF("user_id","item_id","mood","time","sessionBegin")
val df2 = df.withColumn("time", 'time.cast("int"))
df2.createOrReplaceTempView("user")

val df3 = spark.sql(
  """
    select user_id, item_id, mood, time, sessionBegin,
    case when mood='Happy' then
    collect_list(case when mood='Happy' then ' ' when mood='Sad' then item_id end) over(partition by user_id order by time rows between current row  and unbounded following )
    when mood='Sad' then array()
    end as result from user

  """)
def sliceResult(x:Seq[String]):Seq[String]={
  val y = x.drop(1).filter( _ != " ")
  y.toSet.toSeq
}
val udf_sliceResult = udf ( sliceResult(_:Seq[String]):Seq[String]  )
df3.withColumn("result1", udf_sliceResult('result) ).show(false)

Results:

+-------+-------+-----+----+------------+------------------------------+---------+
|user_id|item_id|mood |time|sessionBegin|result                        |result1  |
+-------+-------+-----+----+------------+------------------------------+---------+
|1      |A      |Happy|0   |0           |[ ,  ,  ,  ,  , D, C,  ,  , E]|[D, C, E]|
|1      |B      |Happy|1   |0           |[ ,  ,  ,  , D, C,  ,  , E]   |[D, C, E]|
|1      |C      |Happy|3   |0           |[ ,  ,  , D, C,  ,  , E]      |[D, C, E]|
|1      |D      |Happy|5   |0           |[ ,  , D, C,  ,  , E]         |[D, C, E]|
|1      |C      |Happy|6   |0           |[ , D, C,  ,  , E]            |[D, C, E]|
|1      |D      |Sad  |6   |0           |[]                            |[]       |
|1      |C      |Sad  |10  |0           |[]                            |[]       |
|1      |A      |Happy|28  |0           |[ ,  , E]                     |[E]      |
|1      |E      |Happy|35  |0           |[ , E]                        |[E]      |
|1      |E      |Sad  |60  |0           |[]                            |[]       |
|2      |F      |Happy|6   |6           |[ ,  ,  , D,  ,  ,  , G]      |[D, G]   |
|2      |E      |Happy|17  |6           |[ ,  , D,  ,  ,  , G]         |[D, G]   |
|2      |D      |Happy|20  |6           |[ , D,  ,  ,  , G]            |[D, G]   |
|2      |D      |Sad  |21  |6           |[]                            |[]       |
|2      |E      |Happy|27  |6           |[ ,  ,  , G]                  |[G]      |
|2      |G      |Happy|37  |6           |[ ,  , G]                     |[G]      |
|2      |H      |Happy|39  |6           |[ , G]                        |[G]      |
|2      |G      |Sad  |45  |6           |[]                            |[]       |
+-------+-------+-----+----+------------+------------------------------+---------+

EDIT1:

As OP mentioned, the ' ' can be replaced with null and df3 itself will be the final result. Thus udf() can be avoided

scala> :paste
// Entering paste mode (ctrl-D to finish)

val df3 = spark.sql(
  """
    select user_id, item_id, mood, time, sessionBegin,
    case when mood='Happy' then
    collect_list(case when mood='Happy' then null when mood='Sad' then item_id end) over(partition by user_id order by time rows between current row  and unbounded following )
    when mood='Sad' then array()
    end as result from user
  """)

// Exiting paste mode, now interpreting.

df3: org.apache.spark.sql.DataFrame = [user_id: int, item_id: string ... 4 more fields]

scala> df3.show(false)
+-------+-------+-----+----+------------+---------+
|user_id|item_id|mood |time|sessionBegin|result   |
+-------+-------+-----+----+------------+---------+
|1      |A      |Happy|0   |0           |[D, C, E]|
|1      |B      |Happy|1   |0           |[D, C, E]|
|1      |C      |Happy|3   |0           |[D, C, E]|
|1      |D      |Happy|5   |0           |[D, C, E]|
|1      |C      |Happy|6   |0           |[D, C, E]|
|1      |D      |Sad  |6   |0           |[]       |
|1      |C      |Sad  |10  |0           |[]       |
|1      |A      |Happy|28  |0           |[E]      |
|1      |E      |Happy|35  |0           |[E]      |
|1      |E      |Sad  |60  |0           |[]       |
|2      |F      |Happy|6   |6           |[D, G]   |
|2      |E      |Happy|17  |6           |[D, G]   |
|2      |D      |Happy|20  |6           |[D, G]   |
|2      |D      |Sad  |21  |6           |[]       |
|2      |E      |Happy|27  |6           |[G]      |
|2      |G      |Happy|37  |6           |[G]      |
|2      |H      |Happy|39  |6           |[G]      |
|2      |G      |Sad  |45  |6           |[]       |
+-------+-------+-----+----+------------+---------+


scala>

I get `org.apache.spark.sql.AnalysisException:` in your sql statement, it says that `THEN and ELSE expressions should all be same type or coercible to a common type` — Jeet Banerjee, Nov 30 '18 at 13:10
2.1.0... I fixed it by adding `null` in the else part instead of `array()`. it works now! Thanks !! — Jeet Banerjee, Nov 30 '18 at 13:31
yeah, and if you put `null` instead of `whitespace` , then you don't need to slice result/ drop first element of array - so `df3` will be the final output. — Jeet Banerjee, Nov 30 '18 at 14:59

Spark SQL using a window - collect data from rows after current row based on a column condition

1 Answers1

Linked