How to access stateSnapshot in mapGroupsWithState or share the GroupState between streams?

user_4341887 0
Viewed 970

With the DStream API it's possible to access the snapshot state of a stateful stream using MapWithStateDStream.stateSnapshots(). In the new Structured Streaming API it seems to me that only the function passed to mapGroupsWithState is able to access and update the state.

I'd like to create an in-memory distributed state of a stream based on it's input events. Then enrich events from another stream by joining them to the first event's (complete) state, i.e. not to the first stream itself.

Using the DStream API I'd simply join the second stream with the first stream's stateSnapshot. Is this feature missing on the new Structured Streaming API or is there a new better/cleaner way of sharing the GroupState between two streams this?

spark-structured-streaming apache-spark
2 Answers

Is this feature missing on the new Structured Streaming API or is there a new better/cleaner way of doing this?

There is no stateSnapshot provided out of the box as part of Structured Streaming (SS). I'm assuming this could be done, perhaps in a later version of SS. I'm not sure it fits their design goals, since state is completely masked away from the end user for arbitrary streams, although this could be useful for people using custom state via (flat)mapGroupsWithState.

In order to roll your own "snapshot", you can always output the intermediate state you have in GroupState[S] at for every batch generated, i.e:

def updateSessionEvents(
  id: Int,
  userEvents: Iterator[UserEvent],
  state: GroupState[UserSession]): Option[UserSession] = {
    // Do stuff
    val someState = ??? // update state
    someState
}

Then, you're exposed to the entire state all the time. What this means is that you'll now need to maintain some flag indicating if the state is actually complete or not, in order not to send incomplete state downstream where it isn't mean to be.

user_18708030

mapGroupsWithState is like the functional map where for every input you have to give an output. That's the contract of map in its broad sense and mapGroupsWithState follows it.

If I understood you correctly, what you really need is KeyValueGroupedDataset.flatMapGroupsWithState as you seem to be building a so-called arbitrary stateful stream aggregation.

flatMapGroupsWithState[S, U](outputMode: OutputMode, timeoutConf: GroupStateTimeout)(func: (K, Iterator[V], GroupState[S]) ⇒ Iterator[U])(implicit arg0: Encoder[S], arg1: Encoder[U]): Dataset[U] applies the given function to each group of data, while maintaining a user-defined per-group state.

This seemingly complex flatMapGroupsWithState function gives you the most powerful abstraction for stateful aggregations and my understanding is that if it's not capable of giving you the expected result it's unlikely there's a better function (I'd love being told that I'm wrong).

user_13053440
Related