How to extract timed-out sessions using mapWithState
<p>I am updating my code to switch from <code>updateStateByKey</code> to <code>mapWithState</code> in order to get users' sessions based on a time-out of 2 minutes (2 is used for testing purpose only). Each session should aggregate all the streaming data (JSON string) within a session before time-out.</p>

<p>This was my old code:</p>

<pre><code>val membersSessions = stream.map[(String, (Long, Long, List[String]))](eventRecord =&gt; {
  val parsed = Utils.parseJSON(eventRecord)
  val member_id = parsed.getOrElse("member_id", "")
  val timestamp = parsed.getOrElse("timestamp", "").toLong
  //The timestamp is returned twice because the first one will be used as the start time and the second one as the end time
  (member_id, (timestamp, timestamp, List(eventRecord)))
})

val latestSessionInfo = membersSessions.map[(String, (Long, Long, Long, List[String]))](a =&gt; {
  //transform to (member_id, (time, time, counter, events within session))
  (a._1, (a._2._1, a._2._2, 1, a._2._3))
}).
  reduceByKey((a, b) =&gt; {
    //transform to (member_id, (lowestStartTime, MaxFinishTime, sumOfCounter, events within session))
    (Math.min(a._1, b._1), Math.max(a._2, b._2), a._3 + b._3, a._4 ++ b._4)
  }).updateStateByKey(Utils.updateState)
</code></pre>

<p>The problems of <code>updateStateByKey</code> are nicely explained <a href="http://asyncified.io/2016/07/31/exploring-stateful-streaming-with-apache-spark/" rel="nofollow noreferrer">here</a>. One of the key reasons why I decided to use <code>mapWithState</code> is because <code>updateStateByKey</code> was unable to return <strong>finished sessions (the ones that have timed out)</strong> for further processing.</p>

<p>This is my first attempt to transform the old code to the new version:</p>

<pre><code>val spec = StateSpec.function(updateState _).timeout(Minutes(1))
val latestSessionInfo = membersSessions.map[(String, (Long, Long, Long, List[String]))](a =&gt; {
  //transform to (member_id, (time, time, counter, events within session))
  (a._1, (a._2._1, a._2._2, 1, a._2._3))
})
val userSessionSnapshots = latestSessionInfo.mapWithState(spec).snapshotStream()
</code></pre>

<p>I slightly misunderstand what shoud be the content of <code>updateState</code>, because as far as I understand the time-out should not be calculated manually (it was previously done in my function <code>Utils.updateState</code>) and <code>.snapshotStream</code> should return the timed-out sessions.</p>