StateStoreSaveExec Physical Operator¶

StateStoreSaveExec is a unary physical operator (Spark SQL) that saves (writes) a streaming state (to a state store) with support for streaming watermark.

StateStoreSaveExec is one of the physical operators used for streaming aggregation.

Creating Instance¶

StateStoreSaveExec takes the following to be created:

• Grouping Key Attributes (Spark SQL)
• StatefulOperatorStateInfo
• OutputMode
• Event-time Watermark
• spark.sql.streaming.aggregation.stateFormatVersion
• Child Physical Operator (Spark SQL)

StateStoreSaveExec is created when:

• StatefulAggregationStrategy execution planning strategy is requested to plan a streaming aggregation
• IncrementalExecution is created

OutputMode¶

outputMode: Option[OutputMode]

StateStoreSaveExec can be given an OutputMode when created.

StateStoreSaveExec supports all three OutputModes when executed:

• Append
• Complete
• Update

For Append and Update output modes with event-time watermark defined, StateStoreSaveExec can run another batch if the watermark has advanced.

OutputMode is undefined (None) by default and when AggUtils is requested to planStreamingAggregation (when StatefulAggregationStrategy execution planning strategy is requested to plan a streaming aggregation).

OutputMode is required for executing StateStoreSaveExec. It is specified to be the OutputMode of the IncrementalExecution (when state preparation rule is executed and fills in the execution-specific configuration).

Short Name¶

shortName: String

shortName is part of the StateStoreWriter abstraction.

shortName is the following text:

stateStoreSave

Executing Physical Operator¶

doExecute(): RDD[InternalRow]

doExecute is part of the SparkPlan (Spark SQL) abstraction.

doExecute creates the metrics (to happen on the driver).

doExecute executes the child physical operator (that produces a RDD[InternalRow]) and creates a StateStoreRDD with the following:

storeUpdateFunction¶

storeUpdateFunction: (StateStore, Iterator[T]) => Iterator[U]

storeUpdateFunction branches off based on the given OutputMode:

• Append
• Complete
• Update

Append¶

storeUpdateFunction drops late rows (from the input partition) using the watermark predicate for data.

For every (remaining, non-late) row, storeUpdateFunction stores the row in the input StateStore (using the StreamingAggregationStateManager) and increments the number of updated state rows metric.

storeUpdateFunction tracks the time taken (to drop late rows and store non-late rows) in time to update metric.

storeUpdateFunction requests the StreamingAggregationStateManager for all the aggregates by grouping keys.

storeUpdateFunction creates a new NextIterator:

• When requested for next state value (row), the iterator traverses over the key-value aggregate state pairs until the watermark predicate for keys holds true for a key.

  If so, the iterator removes the key from the state store (via the StreamingAggregationStateManager) and increments the number of removed state rows and number of output rows metrics.

  The value of the removed key is returned as the next element.

• When requested to close, the iterator updates the time to remove metric (to be the time to process all the state rows) and time to commit changes (to be the time taken for the StreamingAggregationStateManager to commit state changes).

  In the end, the iterator records metrics of the StateStore and this stateful operator.

Complete¶

Update¶

AssertionError: outputMode has not been set¶

doExecute makes sure that outputMode is specified or throws an AssertionError:

Incorrect planning in IncrementalExecution, outputMode has not been set

StreamingAggregationStateManager¶

StateStoreSaveExec creates a StreamingAggregationStateManager when created.

The StreamingAggregationStateManager is created using the following:

• Grouping Keys
• Output schema of the child physical operator

The StreamingAggregationStateManager is used in doExecute for the following:

• State Value Schema to mapPartitionsWithStateStore

For Complete:

1. Store all input rows
2. Commit the state changes
3. Fetch the values

For Append:

1. Store all updated rows
2. Fetch all the key-value pairs
3. Remove old "watermarked" aggregates for all removed pairs
4. Commit the state changes

For Update:

1. Store all updated rows
2. removeKeysOlderThanWatermark of old "watermarked" aggregates
3. Commit the state changes

Required Child Output Distribution¶

requiredChildDistribution: Seq[Distribution]

requiredChildDistribution is part of the SparkPlan (Spark SQL) abstraction.

requiredChildDistribution...FIXME

Performance Metrics¶

StateStoreSaveExec is a StateStoreWriter that defines the performance metrics for writes to a state store.

memory used by state¶

Estimated memory used by a StateStore (aka stateMemory) after StateStoreSaveExec finished execution (per the StateStoreMetrics of the StateStore)

number of output rows¶

number of removed state rows¶

See number of removed state rows

number of rows which are dropped by watermark¶

See number of rows which are dropped by watermark

number of updated state rows¶

The metric is computed differently based on the given OutputMode.

StateStoreSaveExec is a StateStoreWriter so consult number of updated state rows, too.

time to commit changes¶

Time taken for the StreamingAggregationStateManager to commit changes to a state store

When requested to commit changes to a state store, a StreamingAggregationStateManager (as StreamingAggregationStateManagerBaseImpl) requests the given StateStore to commit state changes.

For RocksDBStateStore, it means for RocksDB to commit state changes which is made up of the time-tracked phases that are available using the performance metrics.

In other words, the total of the time-tracked phases is close approximation of this metric (there are some file-specific ops though that contribute to the metric but are not part of the phases).

time to remove¶

The metric is computed differently based on the given OutputMode.

StateStoreSaveExec is a StateStoreWriter so consult time to remove, too.

time to update¶

The metric is computed differently based on the given OutputMode.

Logging¶

Enable ALL logging level for org.apache.spark.sql.execution.streaming.StateStoreSaveExec logger to see what happens inside.

Add the following line to conf/log4j.properties:

log4j.logger.org.apache.spark.sql.execution.streaming.StateStoreSaveExec=ALL

Refer to Logging.