[#36841] Add streaming dispatch seam to the Spark Structured Streaming runner - #39906
Merged
Abacn merged 1 commit intoAug 28, 2026
Merged
Conversation
Prepares the structured streaming runner for a Spark 4 streaming translator without changing batch behavior. - Add PipelineTranslatorFactory, the seam the Spark 4 module shadows to dispatch streaming pipelines. The shared base rejects streaming with a clear message instead of the previous generic checkArgument. - Open up EvaluationContext (non final, protected ctor, leaves()) and add a no-op stop() so a streaming context can override evaluation. - Add a createEvaluationContext hook to PipelineTranslator and skip the persist and lineage breaking optimizations for streaming datasets. - Plumb the EvaluationContext into SparkStructuredStreamingPipelineResult so cancel() can stop a running streaming query. - Default the state store provider to RocksDB, required by Spark 4 transformWithState and inert for batch. - Add watermarkDelayMillis, maxRecordsPerMicroBatch, maxBatchDurationMillis and streamingStopAfterIdleBatches options. - Narrow the Spark 4 test source override exclude to the legacy DStream package. The previous glob also matched structuredstreaming and would have silently dropped the structured streaming tests.
Contributor
|
Assigning reviewers: R: @Abacn added as fallback since no labels match configuration Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
Abacn
reviewed
Aug 28, 2026
| // Note: deliberately NOT named getMaxRecordsPerBatch. The legacy Spark runner's | ||
| // SparkPipelineOptions already declares Long getMaxRecordsPerBatch(); a same-name getter with a | ||
| // different return type breaks proxy generation for every registered PipelineOptions interface. | ||
| @Description( |
Contributor
There was a problem hiding this comment.
If Spark structured streaming runner does not use the legacy "getMaxRecordsPerBatch", we can use the same existing pipeline option. This make user migration effort (from legacy Spark to Spark streaming runner) smoothier. Long return type should be fine.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
First slice of the Spark 4 Structured Streaming work, split out of the POC #39576 as announced there. Addresses #36841.
This prepares the structured streaming runner for a streaming translator without changing batch behavior on either Spark version.
No behavior change for Spark 3, proven by the full :runners:spark:3:test suite (220 tests, 0 failures) and :runners:spark:4:test (196 tests, 0 failures) locally on JDK 17, plus spotless, checkstyle and a live ErrorProne compile.
The end to end evidence that this seam carries a working streaming runner is in draft #39576. Remaining slices, in order: Kryo registrations, the DataSourceV2 unbounded source, the state and timer bridge on transformWithState, and the translators with the end to end tests.
R: @Abacn