StateStoreRDD — RDD for Updating State (in StateStores Across Spark Cluster)-spark技术分享

StateStoreRDD — RDD for Updating State (in StateStores Across Spark Cluster)

StateStoreRDD is an RDD for executing storeUpdateFunction with StateStore (and data from partitions of a new batch RDD).

StateStoreRDD is created when the following stateful physical operators are executed (using StateStoreOps.mapPartitionsWithStateStore):

StateStoreRDD SparkPlans LogicalPlans operators.png

Figure 1. StateStoreRDD, Physical and Logical Plans, and operators

StateStoreRDD uses StateStoreCoordinator for preferred locations for job scheduling.

Figure 2. StateStoreRDD and StateStoreCoordinator

getPartitions is exactly the partitions of the data RDD.

Table 1. StateStoreRDD’s Internal Registries and Counters
Name	Description
`hadoopConfBroadcast`
`storeConf`	Configuration parameters (as `StateStoreConf`) using the current `SQLConf` (from `SessionState`)

Computing Partition (in TaskContext) — `compute` Method



compute(partition: Partition, ctxt: TaskContext): Iterator[U]

compute(partition: Partition, ctxt: TaskContext): Iterator[U]

Note	`compute` is a part of the RDD Contract to compute a given partition in a `TaskContext`.

compute computes dataRDD passing the result on to storeUpdateFunction (with a configured StateStore).

Internally, (and similarly to getPreferredLocations) compute creates a StateStoreProviderId with StateStoreId (using checkpointLocation, operatorId and the index of the input partition) and queryRunId.

compute then requests StateStore for the store for the StateStoreProviderId.

In the end, compute computes dataRDD (using the input partition and ctxt) followed by executing storeUpdateFunction (with the store and the result).

Getting Placement Preferences of Partition — `getPreferredLocations` Method



getPreferredLocations(partition: Partition): Seq[String]

getPreferredLocations(partition: Partition): Seq[String]

Note	`getPreferredLocations` is a part of the RDD Contract to specify placement preferences (aka preferred task locations), i.e. where tasks should be executed to be as close to the data as possible.

getPreferredLocations creates a StateStoreProviderId with StateStoreId (using checkpointLocation, operatorId and the index of the input partition) and queryRunId.

Note	checkpointLocation and operatorId are shared across different partitions and so the only difference in `StateStoreProviderIds` is the partition index.

In the end, getPreferredLocations requests StateStoreCoordinatorRef for the location of the state store for StateStoreProviderId.

Note	StateStoreCoordinator coordinates instances of `StateStores` across Spark executors in the cluster, and tracks their locations for job scheduling.

Creating StateStoreRDD Instance

StateStoreRDD takes the following when created:

RDD with the new streaming batch data (to update the aggregates in a state store)
Store update function (i.e. (StateStore, Iterator[T]) ⇒ Iterator[U] with T being the type of the new batch data)
The path to the checkpoint location
queryRunId
Operator id
Store version
Schema of the keys
Schema of the values
Optional index
SessionState
Optional StateStoreCoordinatorRef

StateStoreRDD initializes the internal registries and counters.

StateStoreRDD — RDD for Updating State (in StateStores Across Spark Cluster)