[SPARK-54346][SS] Introduce state repartition API and repartition runner #53056
Conversation
| }, | ||
| "sqlState" : "55019" | ||
| }, | ||
| "STATE_REPARTITION_INVALID_PARAMETER" : { |
There was a problem hiding this comment.
nit: INVALID_OPTIONS ?
There was a problem hiding this comment.
I want to avoid confusion, since options in spark means .option(). So using parameter instead.
| /** | ||
| * Returns a `StreamingCheckpointManager` that allows managing any streaming checkpoint. | ||
| */ | ||
| private[spark] def streamingCheckpointManager: StreamingCheckpointManager = |
There was a problem hiding this comment.
We need to add to Spark connect also ?
There was a problem hiding this comment.
Spark connect and pyspark will be added in subsequent PRs.
| /** @inheritdoc */ | ||
| override private[spark] def repartition( | ||
| checkpointLocation: String, | ||
| numPartitions: Int, |
There was a problem hiding this comment.
I guess the underlying recorded value is Int, but should we consider bumping this to Long eventually - probably unlikely for users to have those many partitions though
There was a problem hiding this comment.
lets keep it as int since that is what we record in checkpoint. Also using Long is very unrealistic.
| import org.apache.hadoop.conf.Configuration | ||
| import org.apache.hadoop.fs.Path | ||
|
|
||
| object StreamingUtils { |
There was a problem hiding this comment.
can we put this under this directory -
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/utils/
| throw OfflineStateRepartitionErrors.parameterIsNotGreaterThanZeroError("numPartitions") | ||
| } | ||
|
|
||
| val runner = new OfflineStateRepartitionRunner( |
There was a problem hiding this comment.
Should we encapsulate this whole block in a try-catch in case we want to catch and log any warnings ?
| numPartitions, | ||
| enforceExactlyOnceSink | ||
| ) | ||
| runner.run() |
There was a problem hiding this comment.
Can we also add some logging to indicate the repartition started/ended and the time it took to complete the operation along with other identifying information about the query ?
There was a problem hiding this comment.
see the runner run method
|
|
||
| val newBatchId = createNewBatchIfNeeded(lastBatchId, lastCommittedBatchId) | ||
|
|
||
| // todo: Do the repartitioning here, in subsequent PR |
There was a problem hiding this comment.
Can we create SPARK JIRAs and link them here ?
2 participants
What changes were proposed in this pull request?
Introducing the API for offline repartitioning of streaming state. This is currently not exposed, since it is still in development. Also implemented some of the core functionalities of the repartition batch runner, that validates the checkpoint, creates the repartition batch and commits. Subsequent PRs will build on this. Also Spark connect and pyspark APIs will be added in subsequent PRs.
Also introduce the streamingCheckpointManager for performing operations on the streaming checkpoint. This is currently not exposed, since it is still in development.
Why are the changes needed?
Streaming state repartitioning
Does this PR introduce any user-facing change?
No
How was this patch tested?
New test suite added
Was this patch authored or co-authored using generative AI tooling?
No