Example: Car Fleet Management

Let’s imagine a car fleet management solution in which the vehicles of the fleet are enabled with wireless network capabilities. Each vehicle regularly reports its geographical location and many operational parameters like fuel level, speed, acceleration, bearing, engine temperature, and so on. The stakeholders would like to exploit this stream of telemetry data to implement a range of applications to help them manage the operational and financial aspects of the business.

Using the Structured Streaming features we know so far, we could already implement many use cases, like monitoring kilometers driven per day using event-time windows or finding vehicles with a low-fuel warning by applying filters.

Now, we would like to have the notion of a trip: the driven road segment from a start to a stop. Individually, the notion of a trip is useful to compute fuel efficiency or monitor compliance to geo-fencing agreements. When analyzed in groups, trip information might reveal transportation patterns, traffic hotspots, and, when combined with other sensor information, they can even report road conditions. From our stream-processing perspective, we could see trips as an arbitrary window that opens when the vehicle starts moving and closes when it finally stops. The event-time window aggregations we saw in Chapter 12 use fixed time intervals as windowing criteria, so they are of no help to implement our trip analysis.

We can appreciate that we need a more powerful definition of state that is not purely based on time, but also on arbitrary conditions. In our example, this condition is that the vehicle is driving.

Understanding Group with State Operations

The arbitrary state operations, mapGroupsWithState and flatMapGroupWithState, work exclusively on the typed Dataset API using either the Scala or the Java bindings.

Based on the data that we are processing and the requirements of our stateful transformation, we need to provide three type definitions, typically encoded as a case class (Scala) or a Java Bean (Java):

• The input event (I)

• The arbitrary state to keep (S)

• The output (O) (this type might be the same as the state representation, if suitable)

All of these types must be encodable into Spark SQL types. This means that there should be an Encoder available. The usual import statement

import spark.implicits._

is sufficient for all basic types, tuples, and case classes.

With these types in place, we can formulate the state transformation function that implements our custom state handling logic.

mapGroupsWithState requires that this function returns a single mandatory value:

def mappingFunction(key: K, values: Iterator[I], state: GroupState[S]): O

flatMapGroupsWithState requires that this function returns an Iterator, which might contain zero or more elements:

def flatMappingFunction(
    key: K, values: Iterator[I], state: GroupState[S]): Iterator[O]

GroupState[S] is a wrapper provided by Structured Streaming and used internally to manage the state S across executions. Within the function, GroupState provides mutation access to the state and the ability to check and set timeouts.

Internal State Flow

In Figure 13-1, we illustrate the process that combines the incoming data, in the form of events, with the state maintained internally, to produce a result. In this chart, the mappingFunction (denoted with a Σ) uses the custom logic to process this group of elements, that when combined with the state managed by GroupState[S], leads to a result. In this illustration, we used the stop symbol to indicate a timeout. In the case of MapGroupsWithState, a timeout also triggers the production of an event and should evict the state. Given that the eviction logic is under the control of the programmed logic, the complete state management is under the responsibility of the developer. Structured Streaming provides only the building blocks.

Figure 13-1. Map groups with state dynamics

Using MapGroupsWithState

In “Sliding windows”, we saw how we can compute a moving average based on a time window. This time-based window would produce a result independently of the number of elements found in the window.

Now, suppose that our requirement is to compute a moving average of the last 10 elements received per key. We cannot use a time window, because we don’t know how long it will take us to have the number of elements we need. Instead, we can define our own count-based window using a custom state with MapGroupsWithState.

Let’s begin with the same streaming Dataset that we used in “Sliding windows”. The WeatherEvent case class becomes our input type (I):

// a representation of a weather station event
case class WeatherEvent(stationId: String,
  timestamp: Timestamp,
  location:(Double,Double),
  pressure: Double,
  temp: Double)

val weatherEvents: Dataset[WeatherEvents] = ...

Next, we define the state (S). What we want is to keep the latest n elements in our state and drop anything older. This seems a natural application of a FIFO (First In, First Out) collection, such as a Queue. Newer elements are added to the front of the queue, we keep the most recent n, and drop any older element.

Our state definition becomes a FIFOBuffer backed by a Queue with few helper methods to facilitate its usage:

import scala.collection.immutable.Queue
case class FIFOBuffer[T](
    capacity: Int, data: Queue[T] = Queue.empty
    ) extends Serializable {

  def add(element: T): FIFOBuffer[T] =
    this.copy(data = data.enqueue(element).take(capacity))

  def get: List[T] = data.toList

  def size: Int = data.size
}

Then, we need to define the output type (O) that results from the stateful computation. The desired result of our stateful computation is the moving average of the sensor values present in the input WeatherEvent. We also would like to know the time span of the values used for the computation. With this knowledge, we design our output type, WeatherEventAverage:

import java.sql.Timestamp
case class WeatherEventAverage(stationId: String,
                               startTime: Timestamp,
                               endTime:Timestamp,
                               pressureAvg: Double,
                               tempAvg: Double)

With these types defined, we can proceed to create the mappingFunction that combines the existing state and the new elements into a result. We can see the implementation of the mapping function in Example 13-1. Remember that this function is also responsible for updating the internal state through the functions provided by the GroupState wrapper. It’s important to note that the state cannot be updated with a null value. An attempt to do so will throw an IllegalArgumentException. To remove a state, use the method state.remove().

import org.apache.spark.sql.streaming.GroupState
def mappingFunction(
    key: String,
    values: Iterator[WeatherEvent],
    state: GroupState[FIFOBuffer[WeatherEvent]]
  ): WeatherEventAverage = {

  // the size of the window
  val ElementCountWindowSize = 10

  // get current state or create a new one if there's no previous state
  val currentState = state.getOption
    .getOrElse(
      new FIFOBuffer[WeatherEvent](ElementCountWindowSize)
    )

  // enrich the state with the new events
  val updatedState = values.foldLeft(currentState) {
    case (st, ev) => st.add(ev)
  }

  // update the state with the enriched state
  state.update(updatedState)

  // if we have enough data, create a WeatherEventAverage from the state
  // otherwise, make a zeroed record
  val data = updatedState.get
  if (data.size > 2) {
    val start = data.head
    val end = data.last
    val pressureAvg = data
      .map(event => event.pressure)
      .sum / data.size
    val tempAvg = data
      .map(event => event.temp)
      .sum / data.size
    WeatherEventAverage(
      key,
      start.timestamp,
      end.timestamp,
      pressureAvg,
      tempAvg
    )
  } else {
    WeatherEventAverage(
      key,
      new Timestamp(0),
      new Timestamp(0),
      0.0,
      0.0
    )
  }
}

Now, we use the mappingFunction to declare the stateful transformation of the streaming Dataset:

import org.apache.spark.sql.streaming.GroupStateTimeout
val weatherEventsMovingAverage = weatherEvents
  .groupByKey(record => record.stationId)
  .mapGroupsWithState(GroupStateTimeout.ProcessingTimeTimeout)(mappingFunction)

Note that we first create groups out of the key identifiers in our domain. In this example, this is the stationId. The groupByKey operation creates an intermediate structure, a KeyValueGroupedDataset that becomes the entry point for the [map|flatMap]GroupWithState operations.

Besides the mapping function, we also need to provide a timeout type. A timeout type can be either a ProcessingTimeTimeout or an EventTimeTimeout. Because we are not relying on the timestamp of the events for our state management, we chose the ProcessingTimeTimeout. We discuss timeout management in detail later in this chapter.

Finally, we can easily observe the results of the query by using the Console sink:

val outQuery = weatherEventsMovingAverage.writeStream
  .format("console")
  .outputMode("update")
  .start()

+---------+-------------------+-------------------+------------+------------+
|stationId|startTime          |endTime            |pressureAvg |tempAvg     |
+---------+-------------------+-------------------+------------+------------+
|d1e46a42 |2018-07-08 19:20:31|2018-07-08 19:20:36|101.33375295|19.753225782|
|d1e46a42 |2018-07-08 19:20:31|2018-07-08 19:20:44|101.33667584|14.287718525|
|d60779f6 |2018-07-08 19:20:38|2018-07-08 19:20:48|101.59818386|11.990002708|
|d1e46a42 |2018-07-08 19:20:31|2018-07-08 19:20:49|101.34226429|11.294964619|
|d60779f6 |2018-07-08 19:20:38|2018-07-08 19:20:51|101.63191940|8.3239282534|
|d8e16e2a |2018-07-08 19:20:40|2018-07-08 19:20:52|101.61979385|5.0717571842|
|d4c162ee |2018-07-08 19:20:34|2018-07-08 19:20:53|101.55532969|13.072768358|
+---------+-------------------+-------------------+------------+------------+
// (!) output edited to fit in the page

Using FlatMapGroupsWithState

Our previous implementation has a flaw. Can you spot it?

When we start processing the stream, and before we have collected all the elements that we deem required to compute the moving average, the operation of mapGroupsWithState produces zeroed-out values:

+---------+-------------------+-------------------+-----------+-------+
|stationId|startTime          |endTime            |pressureAvg|tempAvg|
+---------+-------------------+-------------------+-----------+-------+
|d2e710aa |1970-01-01 01:00:00|1970-01-01 01:00:00|0.0        |0.0    |
|d1e46a42 |1970-01-01 01:00:00|1970-01-01 01:00:00|0.0        |0.0    |
|d4a11632 |1970-01-01 01:00:00|1970-01-01 01:00:00|0.0        |0.0    |
+---------+-------------------+-------------------+-----------+-------+

As we mentioned earlier, mapGroupsWithState requires the state handling function to produce a single record for each group processed at every trigger interval. This is fine when the arrival of new data corresponding to each key naturally updates its state.

But there are cases for which our state logic requires a series of events to occur before we can produce a result. In our current example, we need n elements before we can start producing an average over them. In other scenarios, it might be that a single incoming event might complete several temporary states and therefore produce more than one result. For example, the arrival of a single mass transport to its destination might update the traveling state of all of its passengers, potentially producing a record for each of them.

flatMapGroupsWithState is a generalization of mapGroupsWithState in which the state handling function produces an Iterator of results, which might contain zero or more elements.

Let’s see how we can use this function to improve our moving average computation over n-elements.

We need to update the mapping function to return an Iterator of results. In our case, this Iterator will contain zero elements when we don’t have enough values to compute the average, and a value otherwise. Our changed function looks like Example 13-2.

import org.apache.spark.sql.streaming._
def flatMappingFunction(
    key: String,
    values: Iterator[WeatherEvent],
    state: GroupState[FIFOBuffer[WeatherEvent]]
  ): Iterator[WeatherEventAverage] = {

  val ElementCountWindowSize = 10

  // get current state or create a new one if there's no previous state
  val currentState = state.getOption
    .getOrElse(
      new FIFOBuffer[WeatherEvent](ElementCountWindowSize)
    )

  // enrich the state with the new events
  val updatedState = values.foldLeft(currentState) {
    case (st, ev) => st.add(ev)
  }

  // update the state with the enriched state
  state.update(updatedState)

  // only when we have enough data, create a WeatherEventAverage from the state
  // before that, we return an empty result.
  val data = updatedState.get
  if (data.size == ElementCountWindowSize) {
    val start = data.head
    val end = data.last
    val pressureAvg = data
      .map(event => event.pressure)
      .sum / data.size
    val tempAvg = data
      .map(event => event.temp)
      .sum / data.size
    Iterator(
      WeatherEventAverage(
        key,
        start.timestamp,
        end.timestamp,
        pressureAvg,
        tempAvg
      )
    )
  } else {
    Iterator.empty
  }
}

val weatherEventsMovingAverage = weatherEvents
  .groupByKey(record => record.stationId)
  .flatMapGroupsWithState(
    OutputMode.Update,
    GroupStateTimeout.ProcessingTimeTimeout
  )(flatMappingFunction)

Using flatMapGroupsWithState, we no longer need to produce artificial zeroed records. In addition to that, our state management definition is now strict in having n elements to produce a result.

import org.apache.spark.sql.streaming.GroupStateTimeout
val weatherEventsMovingAverage = weatherEvents
  .groupByKey(record => record.stationId)
  .mapGroupsWithState(GroupStateTimeout.ProcessingTimeTimeout)(mappingFunction)

val weatherEventsMovingAverage = weatherEvents
  .withWatermark("timestamp", "2 minutes")
  .groupByKey(record => record.stationId)
  .mapGroupsWithState(GroupStateTimeout.EventTimeTimeout)(mappingFunction)

def stateToAverageEvent(
    key: String,
    data: FIFOBuffer[WeatherEvent]
  ): Iterator[WeatherEventAverage] = {
  if (data.size == ElementCountWindowSize) {
    val events = data.get
    val start = events.head
    val end = events.last
    val pressureAvg = events
      .map(event => event.pressure)
      .sum / data.size
    val tempAvg = events
      .map(event => event.temp)
      .sum / data.size
    Iterator(
      WeatherEventAverage(
        key,
        start.timestamp,
        end.timestamp,
        pressureAvg,
        tempAvg
      )
    )
  } else {
    Iterator.empty
  }
}

import org.apache.spark.sql.streaming.GroupState
def flatMappingFunction(
    key: String,
    values: Iterator[WeatherEvent],
    state: GroupState[FIFOBuffer[WeatherEvent]]
  ): Iterator[WeatherEventAverage] = {
  // first check for timeout in the state
  if (state.hasTimedOut) {
    // when the state has a timeout, the values are empty
    // this validation is only to illustrate the point
    assert(
      values.isEmpty,
      "When the state has a timeout, the values are empty"
    )
    val result = stateToAverageEvent(key, state.get)
    // evict the timed-out state
    state.remove()
    // emit the result of transforming the current state into an output record
    result
  } else {
    // get current state or create a new one if there's no previous state
    val currentState = state.getOption.getOrElse(
      new FIFOBuffer[WeatherEvent](ElementCountWindowSize)
    )
    // enrich the state with the new events
    val updatedState = values.foldLeft(currentState) {
      case (st, ev) => st.add(ev)
    }
    // update the state with the enriched state
    state.update(updatedState)
    state.setTimeoutDuration("30 seconds")
    // only when we have enough data,
    // create a WeatherEventAverage from the accumulated state
    // before that, we return an empty result.
    stateToAverageEvent(key, updatedState)
  }
}

Summary

In this chapter, we learned about the arbitrary stateful processing API in Structured Streaming. We explored the details of and differences between the mapGroupsWithState and flatMapGroupsWithState with relation to the events produced and the output modes supported. At the end, we also learned about the timeout settings and became aware of its semantics.

Although this API is more complex to use than the regular SQL-like constructs of the structured APIs, it provides us with a powerful toolset to implement arbitrary state management to address the development of the most demanding streaming use cases.