spark-sql-streaming-IncrementalExecution.adoc

IncrementalExecution — QueryExecution of Streaming Datasets

IncrementalExecution is a QueryExecution of a streaming Dataset that StreamExecution creates when incrementally executing the logical query plan (every trigger).

Figure 1. StreamExecution creates IncrementalExecution (every trigger / streaming batch)

Tip	Details on QueryExecution contract can be found in the Mastering Apache Spark 2 gitbook.

IncrementalExecution registers state physical preparation rule with the parent QueryExecution's preparations that prepares the streaming physical plan (using batch-specific execution properties).

IncrementalExecution is created when:

• StreamExecution plans a streaming query

• ExplainCommand is executed (for explain operator)

Table 1. IncrementalExecution’s Internal Registries and Counters (in alphabetical order)

Name	Description
planner	SparkPlanner with the following extra planning strategies (in the order of execution): StatefulAggregationStrategy FlatMapGroupsWithStateStrategy StreamingRelationStrategy StreamingDeduplicationStrategy Note planner is used to plan (aka convert) an optimized logical plan into a physical plan (that is later available as sparkPlan). sparkPlan physical plan is then prepared for execution using preparations physical optimization rules. The result is later available as executedPlan physical plan.
state	State preparation rule (i.e. Rule[SparkPlan]) that transforms a streaming physical plan (i.e. SparkPlan with StateStoreSaveExec, StreamingDeduplicateExec and FlatMapGroupsWithStateExec physical operators) and fills missing properties that are batch-specific, e.g. StateStoreSaveExec and StateStoreRestoreExec operator pair, state assigns: StateStoreSaveExec operator gets nextStatefulOperationStateInfo, OutputMode and batchWatermarkMs StateStoreRestoreExec operator gets nextStatefulOperationStateInfo that was used for StateStoreSaveExec operator StreamingDeduplicateExec operator gets nextStatefulOperationStateInfo and batchWatermarkMs FlatMapGroupsWithStateExec gets nextStatefulOperationStateInfo, batchWatermarkMs and batchWatermarkMs Used when IncrementalExecution prepares a physical plan (i.e. SparkPlan) for execution (which is when StreamExecution runs a streaming batch and plans a streaming query).
statefulOperatorId	Java’s AtomicInteger 0 when IncrementalExecution is created Incremented…​FIXME

nextStatefulOperationStateInfo Internal Method

nextStatefulOperationStateInfo(): StatefulOperatorStateInfo

nextStatefulOperationStateInfo creates a StatefulOperatorStateInfo with checkpointLocation, runId, the next statefulOperatorId and currentBatchId.

Note	All the properties of StatefulOperatorStateInfo are specified when IncrementalExecution is created.

Note	nextStatefulOperationStateInfo is used exclusively when IncrementalExecution is requested to transform a streaming physical plan using state preparation rule.

Creating IncrementalExecution Instance

IncrementalExecution takes the following when created:

• SparkSession

• Logical query plan (i.e. LogicalPlan with the logical plans of the data sources that have new data and new column attributes)

• OutputMode (as specified using outputMode method of DataStreamWriter)

• state checkpoint directory (as specified using checkpointLocation option or spark.sql.streaming.checkpointLocation Spark property with queryName option)

• Run id

• Current batch id

• OffsetSeqMetadata

IncrementalExecution initializes the internal registries and counters.