(StateStore, Iterator[T]) => Iterator[U]Arbitrary Stateful Streaming Aggregation
Arbitrary Stateful Streaming Aggregation is a streaming aggregation query that uses the following KeyValueGroupedDataset operators:
-
mapGroupsWithState for implicit state logic
-
flatMapGroupsWithState for explicit state logic
KeyValueGroupedDataset represents a grouped dataset as a result of Dataset.groupByKey operator.
mapGroupsWithState and flatMapGroupsWithState operators use GroupState as group streaming aggregation state that is created separately for every aggregation key with an aggregation state value (of a user-defined type).
mapGroupsWithState and flatMapGroupsWithState operators use GroupStateTimeout as an aggregation state timeout that defines when a GroupState can be considered timed-out (expired).
Performance Metrics
Arbitrary Stateful Streaming Aggregation uses performance metrics (of the StateStoreWriter through FlatMapGroupsWithStateExec physical operator).
Internals
One of the most important internal execution components of Arbitrary Stateful Streaming Aggregation is FlatMapGroupsWithStateExec physical operator.
When requested to execute and generate a recipe for a distributed computation (as an RDD[InternalRow]), FlatMapGroupsWithStateExec first validates a selected GroupStateTimeout:
-
For ProcessingTimeTimeout, batch timeout threshold has to be defined
-
For EventTimeTimeout, event-time watermark has to be defined and the input schema has the watermark attribute
|
Note
|
FIXME When are the above requirements met? |
FlatMapGroupsWithStateExec physical operator then mapPartitionsWithStateStore with a custom storeUpdateFunction of the following signature:
While generating the recipe, FlatMapGroupsWithStateExec uses StateStoreOps extension method object to register a listener that is executed on a task completion. The listener makes sure that a given StateStore has all state changes either committed or aborted.
In the end, FlatMapGroupsWithStateExec creates a new StateStoreRDD and adds it to the RDD lineage.
StateStoreRDD is used to properly distribute tasks across executors (per preferred locations) with help of StateStoreCoordinator (that runs on the driver).
StateStoreRDD uses StateStore helper to look up a StateStore by StateStoreProviderId and store version.
FlatMapGroupsWithStateExec physical operator uses state managers that are different than state managers for Streaming Aggregation. StateStore abstraction is the same as in Streaming Aggregation.
One of the important execution steps is when InputProcessor (of FlatMapGroupsWithStateExec physical operator) is requested to callFunctionAndUpdateState. That executes the user-defined state function on a per-group state key object, value objects, and a GroupStateImpl.