title: Spark (Scala): Compute session length of user in DataFrame with continuous active hours
body: <p>I like to do the following transformation. Given a data frame that records whether a user is active each hour, and a continuous # of hours that user is active is counted as session, I'm trying to gather the cumulative hour in each session.</p>
  
  <p>For instance, the original DataFrame would look like the following:</p>
  
  <pre><code>scala&gt; val df = sc.parallelize(List(
    ("user1",0,true),
    ("user1",1,true),
    ("user1",2,false),
    ("user1",3,true),
    ("user1",4,false),
    ("user1",5,false),
    ("user1",6,true),
    ("user1",7,true),
    ("user1",8,true)
  )).toDF("user_id","hour_of_day","is_active")
  df: org.apache.spark.sql.DataFrame = [user_id: string, hour_of_day: int, is_active: boolean]
  
    +-------+-----------+---------+
    |user_id|hour_of_day|is_active|
    +-------+-----------+---------+
    |user1  |0          |true     |
    |user1  |1          |true     |
    |user1  |2          |false    |
    |user1  |3          |true     |
    |user1  |4          |false    |
    |user1  |5          |false    |
    |user1  |6          |true     |
    |user1  |7          |true     |
    |user1  |8          |true     |
    +-------+-----------+---------+
  </code></pre>
  
  <p>And I'd like to add two columns that tracks when a session begin and the length of the session. Getting one of these columns would allow me to solve the other so either would work.</p>
  
  <p>Example below:</p>
  
  <pre><code>  +-------+-----------+---------+------------------+--------------+
    |user_id|hour_of_day|is_active|session_begin_hour|session_length|
    +-------+-----------+---------+------------------+--------------+
    |user1  |0          |true     |0                 |1             |
    |user1  |1          |true     |0                 |2             |
    |user1  |2          |false    |null              |0             |
    |user1  |3          |true     |3                 |1             |
    |user1  |4          |false    |null              |0             |
    |user1  |5          |false    |null              |0             |
    |user1  |6          |true     |6                 |1             |
    |user1  |7          |true     |6                 |2             |
    |user1  |8          |true     |6                 |3             |
    +-------+-----------+---------+------------------+--------------+
  </code></pre>
  
  <p>I tried to use WindowSpec to look back a row but that wouldn't allow me to compute the value for column based on the last row, if that column doesn't exist in the original DF.</p>
  
  <p>Are there any elegant solution to solve this problem, preferably in Scala.</p>
  
  <p>Thanks in advance!</p>