StateStore Contract — Kay-Value Store for State Management-spark技术分享

StateStore Contract — Kay-Value Store for State Management

StateStore is the contract of a versioned and possibly fault-tolerant key-value stores for state management (e.g. of streaming aggregations).

StateStore describes a key-value store that lives on every Spark executor for persistent keyed aggregates.

StateStore is identified with the aggregating operator id and the partition id (among other properties for state store identification).

Tip	Read the motivation and design in State Store for Streaming Aggregations.

Method Description

abort



abort(): Unit

abort(): Unit

Aborts the state changes

Used when:

StateStoreOps implicit class is requested to mapPartitionsWithStateStore (when the state store has not been committed for a task that finishes, possibly with an error)
StateStoreHandler (of SymmetricHashJoinStateManager) is requested to abortIfNeeded (when the state store has not been committed for a task that finishes, possibly with an error)

commit



commit(): Long

commit(): Long

Commits state changes

Used when:

FlatMapGroupsWithStateExec, StreamingDeduplicateExec and StreamingGlobalLimitExec physical operators are executed (right after all rows in a partition have been processed)
StreamingAggregationStateManagerBaseImpl is requested to commit (changes to) a state store (exclusively when StateStoreSaveExec physical operator is executed)
StateStoreHandler (of SymmetricHashJoinStateManager) is requested to commit changes to a state store

get



get(key: UnsafeRow): UnsafeRow

get(key: UnsafeRow): UnsafeRow

Used when:

StreamingDeduplicateExec physical operator is requested to doExecute
StreamingGlobalLimitExec physical operator is requested to doExecute
StateManagerImplBase (of FlatMapGroupsWithStateExecHelper) is requested to getState
StreamingAggregationStateManagerImplV1 and StreamingAggregationStateManagerImplV2 are requested to get the current value of a non-null key
KeyToNumValuesStore is requested to get
KeyWithIndexToValueStore` is requested to get and getAll

getRange



getRange(start: Option[UnsafeRow], end: Option[UnsafeRow]): Iterator[UnsafeRowPair]

getRange(start: Option[UnsafeRow], end: Option[UnsafeRow]): Iterator[UnsafeRowPair]

Gets the key-value pairs with optional approximate start and end extents

Used when:

WatermarkSupport is requested to removeKeysOlderThanWatermark
StateManagerImplBase is requested to getAllState
StreamingAggregationStateManagerBaseImpl is requested to keys
KeyToNumValuesStore and KeyWithIndexToValueStore are requested to iterator

hasCommitted



hasCommitted: Boolean

hasCommitted: Boolean

Used when:

StateStoreOps implicit class is requested to mapPartitionsWithStateStore (when a task finishes, possibly with an error, to check whether to abort state updates)
StateStoreHandler (of SymmetricHashJoinStateManager) is requested to abortIfNeeded (when a task finishes, possibly with an error, to check whether to abort state updates)

id



id: StateStoreId

id: StateStoreId

The ID of the state store (for logging purposes only)

Used when:

HDFSBackedStateStore state store is requested for the textual representation
StateStoreHandler (of SymmetricHashJoinStateManager) is requested to abortIfNeeded and getStateStore

iterator



iterator(): Iterator[UnsafeRowPair]

iterator(): Iterator[UnsafeRowPair]

Used when:

StateStoreRestoreExec physical operator is requested to doExecute
HDFSBackedStateStore state store in particular and any StateStore in particular are requested to getRange
StreamingAggregationStateManagerImplV1 state manager is requested to iterator and values
StreamingAggregationStateManagerImplV2 state manager is requested to iterator and values

metrics



metrics: StateStoreMetrics

metrics: StateStoreMetrics

The StateStoreMetrics

Used when:

StateStoreWriter stateful physical operator is requested to setStoreMetrics
StateStoreHandler (of SymmetricHashJoinStateManager) is requested to commit and for the metrics

put



put(key: UnsafeRow, value: UnsafeRow): Unit

put(key: UnsafeRow, value: UnsafeRow): Unit

Stores a given value for a given non-null key

Used when:

StreamingDeduplicateExec and StreamingGlobalLimitExec physical operators are executed
StateManagerImplBase is requested to putState
StreamingAggregationStateManagerImplV1 and StreamingAggregationStateManagerImplV2 are requested to store a row in a state store
KeyToNumValuesStore and KeyWithIndexToValueStore are requested to store a new value for a key

remove



remove(key: UnsafeRow): Unit

remove(key: UnsafeRow): Unit

Removes a given key from the state store

Used when:

Physical operators with WatermarkSupport are requested to removeKeysOlderThanWatermark
StateManagerImplBase is requested to removeState
StreamingAggregationStateManagerBaseImpl is requested to remove a key from a state store
KeyToNumValuesStore is requested to remove a key
KeyWithIndexToValueStore is requested to remove a key and removeAllValues

version



version: Long

version: Long

Used exclusively when HDFSBackedStateStore state store is requested for a new version (that simply the current version incremented)

Table 2. StateStores
StateStore	Description
HDFSBackedStateStore	Uses a HDFS-compatible file system for state persistence
MemoryStateStore

Table 3. StateStore’s Internal Registries and Counters
Name	Description
`loadedProviders`	Registry of StateStoreProviders per `StateStoreProviderId` Used in…FIXME
`_coordRef`	StateStoreCoordinator RPC endpoint (a `RpcEndpointRef` to StateStoreCoordinator). Used in…FIXME

Note	`StateStore` was introduced in [SPARK-13809][SQL] State store for streaming aggregations.

Tip

Enable DEBUG, INFO or WARN logging level for org.apache.spark.sql.execution.streaming.state.StateStore$ logger to see what happens inside.

Add the following line to conf/log4j.properties:



log4j.logger.org.apache.spark.sql.execution.streaming.state.StateStore$=DEBUG

log4j.logger.org.apache.spark.sql.execution.streaming.state.StateStore$=DEBUG

Refer to Logging.

Creating (and Caching) RPC Endpoint Reference to StateStoreCoordinator for Executors — `coordinatorRef` Internal Object Method



coordinatorRef: Option[StateStoreCoordinatorRef]

coordinatorRef: Option[StateStoreCoordinatorRef]

coordinatorRef requests the SparkEnv helper object for the current SparkEnv.

If the SparkEnv is available and the _coordRef is not assigned yet, coordinatorRef prints out the following DEBUG message to the logs followed by requesting the StateStoreCoordinatorRef for the StateStoreCoordinator endpoint.



Getting StateStoreCoordinatorRef

Getting StateStoreCoordinatorRef

If the SparkEnv is available, coordinatorRef prints out the following INFO message to the logs:



Retrieved reference to StateStoreCoordinator: [_coordRef]

Retrieved reference to StateStoreCoordinator: [_coordRef]

Note	`coordinatorRef` is used when `StateStore` helper object is requested to reportActiveStoreInstance (when `StateStore` object helper is requested to find the StateStore by StateStoreProviderId) and verifyIfStoreInstanceActive (when `StateStore` object helper is requested to doMaintenance).

Unloading State Store Provider — `unload` Method



unload(storeProviderId: StateStoreProviderId): Unit

unload(storeProviderId: StateStoreProviderId): Unit

unload…FIXME

Note	`unload` is used when `StateStore` helper object is requested to stop and doMaintenance.

`stop` Object Method



stop(): Unit

stop(): Unit

stop…FIXME

Note	`stop` seems only be used in tests.

Announcing New StateStoreProvider — `reportActiveStoreInstance` Internal Object Method



reportActiveStoreInstance(storeProviderId: StateStoreProviderId): Unit

reportActiveStoreInstance(storeProviderId: StateStoreProviderId): Unit

reportActiveStoreInstance takes the current host and executorId (from the BlockManager on the Spark executor) and requests the StateStoreCoordinatorRef to reportActiveInstance.

Note	`reportActiveStoreInstance` uses `SparkEnv` to access the `BlockManager`.

In the end, reportActiveStoreInstance prints out the following INFO message to the logs:



Reported that the loaded instance [storeProviderId] is active

Reported that the loaded instance [storeProviderId] is active

Note	`reportActiveStoreInstance` is used exclusively when `StateStore` helper object is requested to find the StateStore by StateStoreProviderId.

`MaintenanceTask` Daemon Thread

MaintenanceTask is a daemon thread that triggers maintenance work of every registered StateStoreProvider.

When an error occurs, MaintenanceTask clears loadedProviders registry.

MaintenanceTask is scheduled on state-store-maintenance-task thread pool that runs periodically every spark.sql.streaming.stateStore.maintenanceInterval configuration property (default: 60s).

Retrieving StateStore by ID and Version — `get` Factory Method



get(
  storeProviderId: StateStoreProviderId,
  keySchema: StructType,
  valueSchema: StructType,
  indexOrdinal: Option[Int],
  version: Long,
  storeConf: StateStoreConf,
  hadoopConf: Configuration): StateStore

get(

storeProviderId: StateStoreProviderId,

keySchema: StructType,

valueSchema: StructType,

indexOrdinal: Option[Int],

version: Long,

storeConf: StateStoreConf,

hadoopConf: Configuration): StateStore

get finds StateStore for StateStoreProviderId.

Internally, get looks up the StateStoreProvider (for storeProviderId) in loadedProviders registry. If unavailable, get creates and initializes one.

get will also start the periodic maintenance task (unless already started) and announce the new StateStoreProvider.

In the end, get gets the StateStore (for the version).

Note	`get` is used when: `StateStoreRDD` is requested to compute `StateStoreHandler` (of SymmetricHashJoinStateManager) is requested to getStateStore

Starting Periodic Maintenance Task (Unless Already Started) — `startMaintenanceIfNeeded` Internal Object Method



startMaintenanceIfNeeded(): Unit

startMaintenanceIfNeeded(): Unit

startMaintenanceIfNeeded schedules MaintenanceTask to start after and every spark.sql.streaming.stateStore.maintenanceInterval (defaults to 60s).

Note	`startMaintenanceIfNeeded` does nothing when the maintenance task has already been started and is still running.

Note	`startMaintenanceIfNeeded` is used exclusively when `StateStore` is requested to find the StateStore by StateStoreProviderId.

Performing Maintenance of Registered State Store Providers — `doMaintenance` Internal Object Method



doMaintenance(): Unit

doMaintenance(): Unit

Internally, doMaintenance prints the following DEBUG message to the logs:



Doing maintenance

Doing maintenance

doMaintenance then requests every StateStoreProvider (registered in loadedProviders) to do its own internal maintenance (only when a StateStoreProvider is still active).

When a StateStoreProvider is inactive, doMaintenance removes it from the provider registry and prints the following INFO message to the logs:



Unloaded [provider]

Unloaded [provider]

Note	`doMaintenance` is used exclusively in MaintenanceTask daemon thread.

`verifyIfStoreInstanceActive` Internal Object Method



verifyIfStoreInstanceActive(storeProviderId: StateStoreProviderId): Boolean

verifyIfStoreInstanceActive(storeProviderId: StateStoreProviderId): Boolean

verifyIfStoreInstanceActive…FIXME

Note	`verifyIfStoreInstanceActive` is used exclusively when `StateStore` helper object is requested to doMaintenance (from a running MaintenanceTask daemon thread).

StateStore Contract — Kay-Value Store for State Management