[Spark 4] Streaming runner proof of concept, unbounded sources, windowed GBK and stateful ParDo on transformWithState - #39576
Conversation
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
|
Hi @tkaymak, I found this PR while looking into Spark streaming work. I had been planning to work on Spark portable streaming state and side inputs via #20396 and I’m reading through this POC now so I do not duplicate work or build against the wrong runner My initial interest was state/side-input support, but I see this POC already covers stateful |
|
Hi @Eliaaazzz, happy to share where this is heading. The thinking behind the POC: Spark 4 added transformWithState, and it turned out flexible enough to host the Beam streaming model directly. The POC proves the core claim, two chained Beam stateful operators run inside a single Structured Streaming query with a watermark that propagates correctly between them. So yes, this is the intended direction, and I would not build new work on the DStream path. Once there is agreement on the approach, the plan is to split this branch into five smaller PRs: the dispatch seam and options, the Kryo registrations, the DataSourceV2 unbounded source, the state and timer bridge, and the translators with the end to end tests. Concrete places where help would be useful right now:
Streaming side inputs, your original interest, are deliberately out of scope for the POC, but they are a natural follow up phase once this foundation is agreed. Session windows and full trigger support are planned as later phases too. If you want to start somewhere, I would suggest number 1, it is contained and it teaches you the test setup you would need for anything bigger. |
|
R: @Abacn for the runner and build mechanics, you reviewed the Spark 4 batch runner PRs this builds on. R: @kennknowles for the Beam model side, watermark propagation, trigger semantics and the state and timer bridge on transformWithState. The most interesting finding for you is probably the timer off by one described in the PR, Spark fires a wake up at expiry equal to the watermark while the Beam model requires the watermark to be strictly past the timer, which silently dropped the on time pane until we bounded the fire set. (This is a draft to get agreement on the direction first, the plan is to split it into five smaller PRs afterwards.) |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
|
Update for reviewers, two rounds of changes since the review request. First, the four remarks from my own review pass are addressed: the Kryo registrations for Spark's streaming internals moved to the end of the registrator so auto assigned ids stay identical across Spark 3 and 4, the fallback temporary checkpoint directory is cleaned up after evaluation, the throwing checks in the stateful streaming translator now live in a named rejectUnsupported method with the contract documented, and awaitTermination polls the leaf queries round robin so a failing query surfaces immediately and stops its siblings. Second, the biggest known gap is closed: checkpoint recovery is now durable. The source id is derived deterministically from the read transform's full name, the first run pins its split list under the checkpoint location (Beam sources do not guarantee deterministic splitting), each partition reader persists its checkpoint mark per batch epoch atomically through the Hadoop filesystem, and a restarted run resumes from the newest mark at or before its batch start epoch. Semantics are at least once, a mark is written when a batch finishes reading rather than transactionally with Spark's commit, that caveat is documented on BeamReaderCache. A new end to end test restarts a pipeline against the same checkpoint location with a cleared reader cache and proves it resumes instead of re reading. |
aa1c70e to
be7e3ce
Compare
|
PAssert update. Streaming PAssert now works in non global windows too, commit e991c5d. The earlier commit made PAssert resolve in the global window by emitting an end of stream sentinel that advances the watermark to TIMESTAMP_MAX_VALUE. Non global windows still failed with an empty assertion iterable. The cause was not the fixed windows themselves, PAssert rewindows everything into the global window and chains two GroupByKeys, and the late data filter in front of the second one judged the global window expired because LateDataUtils truncates garbage collection times to the end of the global window while the sentinel watermark sits one day past it. The pane flushed by the first GroupByKey was dropped on arrival at the second. The fix clamps the watermark used for arrival side expiry decisions to the end of the global window, through a delegating TimerInternals view in the step context. Timer firing and the ReduceFnRunner triggers keep the real watermark. Late data semantics for finite windows are unchanged, the clamp only takes effect once the watermark is already past the end of the global window, which only the sentinel can cause. Tests: the fixed windows PAssert case is restored, and both window modes now have negative tests proving a wrong expectation genuinely fails the pipeline. Known remaining gap: chained groupings in finite windows that are flushed only by the final sentinel would still drop their in flight panes. PAssert never builds that shape, it always rewindows to global first. The clean general solution I would choose is per operator output watermark holds, which is on the roadmap for the productionization phase. |
|
Run Java PreCommit |
|
The runner now covers allowed lateness, late firings via AfterPane.elementCountAtLeast(1), accumulating panes, durable checkpoint restart and PAssert in global and non global windows, each guarded by tests that prove the unsupported shapes still fail loudly. Next step is the split announced in the description. The first slice, the streaming dispatch seam, changes nothing for Spark 3 and is coming as a regular PR shortly. I will link it here. This draft stays open as the end to end evidence for the discussion. |
|
The first slice is out: #39906, the streaming dispatch seam. No behavior change for either Spark version, full local gates green. The remaining slices follow in the order listed there once it lands. |
Exposes any Beam UnboundedSource as a Spark 4 DataSourceV2 streaming table with a fixed two column schema, encoded payload plus event timestamp. Offsets are opaque, strictly increasing epoch counters, so Spark keeps scheduling micro-batches and termination stays with the lifecycle owner. Recovery is durable under the query's checkpoint location: the source id derives deterministically from the read transform's full name, the first run pins its split list (Beam sources do not guarantee deterministic splitting), and every split persists its CheckpointMark per epoch with a retention of two, written atomically via temp file and rename. Executors cache live readers between micro-batches and fall back to the newest durable mark at or before the replayed epoch after a restart. Semantics are at least once, a crash between finishing a read and Spark's commit replays the last micro-batch. The batch cutoff honors maxRecordsPerBatch, values below 1, including the default, mean no limit and the batch ends on the duration deadline.
Marks are no longer finalized when a partition reader closes. A reader finalizes the mark taken at its start offset when the next micro-batch for that split is scheduled, because Spark only starts a batch at the initial offset or at the end offset of a batch already in its commit log. A reader whose position does not match the start offset, or that moved without completing its batch (task retry, killed attempt, executor change, restart), is dropped without finalizing and recreated from the durable mark at that offset. Per source state lives under the checkpoint location Spark hands to toMicroBatchStream, written through CheckpointFileManager with the session Hadoop configuration broadcast to executors. Marks are coded with the source's checkpoint mark coder. commit(end) purges marks below end on a background thread, nothing is retained by a fixed count. The dataset is built like BoundedDatasetFactory, a Table holding the real objects wrapped in StreamingRelationV2, no string options, no Base64. Splits travel as objects in the InputPartition, options and Hadoop configuration as broadcasts. maxRecordsPerBatch is divided across splits like the legacy MicrobatchSource, defaultParallelism decides the split count, idle readers back off with FluentBackoff, offsets serialize as the bare epoch like LongOffset with the base class equality. A new option readerIdleTimeoutMillis bounds how long an executor keeps an idle reader. Tests drive the reader cache protocol directly and prove restart recovery, finalization only after commit, and mark purging against Spark's real offsets and commits logs. The JUnit per test timeout is removed from the streaming test, its throwaway thread group poisoned Spark's static pools for later batch tests in the same JVM.
…atch source Brings the final file states of the spark4-streaming-poc branch onto the head of the slice 3 rework (apache#39971) as one commit: the streaming pipeline translator and evaluation context, the Read, Impulse, GroupByKey and stateful ParDo translators, the transformWithState state and timer bridge, and the end to end streaming tests. The io/streaming package of the rework is kept as is, the POC's own version of it is dropped. The end of stream sentinel the POC had added to the old source is re-applied on the reworked BeamPartitionReader and BeamReaderCache: a batch that holds data ends at the first empty poll so its watermark is declared first, and an exhausted reader whose watermark reached the end of the global window emits one empty payload row at the maximum timestamp once per cached reader. The translators filter that row. Callers of the removed int maxRecordsPerMicroBatch option now use the long maxRecordsPerBatch option of master, whose per batch quota is split across the splits with at least one record per split, which is what the tests relied on. StreamingCheckpointRestartTest asserts the reworked checkpoint layout, splits and marks under the per source location Spark hands the stream, instead of the old beam-source-<id> directory found by a recursive search. The JUnit method timeouts are removed from every streaming test. Its timeout thread group leaks into Spark's static pools and breaks later tests in the same JVM. StreamingTestUtils gains run and waitUntilFinish helpers with a five minute deadline that cancel the pipeline and fail the test instead. SparkSessionFactory, build.gradle, the pipeline options, result, runner, evaluation context and pipeline translator of the shared base need no change, the merged slices already carry the POC's deltas including the RocksDB state store default and the Kryo registrations.
c30d670 to
800c9bd
Compare
2 participants
Draft, opened for design discussion rather than for merging as one lump. See "How this is being split" at the bottom. Rebased 2026-09-03 onto the reworked micro-batch source of #39971, the branch is now one commit on top of that PR.
Addresses #36841.
What this is
The Spark 4 runner merged in #38255 is batch only.
SparkStructuredStreamingRunnerrejects streaming outright. This branch is the proof that Spark 4'stransformWithStatecan host the Beam streaming model, including several chained stateful operators inside a single Structured Streaming query with a watermark that propagates correctly across them. It is the end to end evidence behind the slice PRs.Approach
Dispatch seam. Merged in #39906. The shared base gains
translation/PipelineTranslatorFactory.create(boolean streaming), which throws for streaming and names the Spark 4 module.runners/spark/4/srcshadows that one file and returns a streaming translator instead. Every piece oftransformWithState,StreamingQueryand DataSourceV2 streaming code lives only underrunners/spark/4/src.Source. In review as #39971. A DataSourceV2 micro-batch source over any
UnboundedSource, built likeBoundedDatasetFactorywith the real objects in aTablewrapped inStreamingRelationV2. Offsets are opaque epoch counters. Per source state lives under the location Spark passes totoMicroBatchStream, written throughCheckpointFileManager: a pinned split list and, per split, the checkpoint mark coded withgetCheckpointMarkCoder()at the end of every micro-batch. Executors cache live readers and reuse one only when it is positioned at the batch's start offset, and finalize the mark taken there at that moment, since Spark only starts a batch at an offset already in its commit log. Every other case restores the reader from the durable mark at the start offset.commit(end)purges marks belowend. Delivery is at least once. The row schema ispayload BINARYpluseventTimestamp TIMESTAMP, no Catalyst encoder work was needed.Watermark. Declared exactly once, in the read translator, on the raw rows before the typed decode. Spark forbids re-declaring it. The
EventTimeWatermarkplan node survives the projection and the typed map, asserted at plan level and at runtime.End of stream. When a reader's watermark reaches the end of the global window the source emits one sentinel row with an empty payload and
TIMESTAMP_MAX_VALUE, in a micro-batch of its own after the last data batch. The translators filter it before user code. It pushes Spark's watermark to the end so every window finalizes, which is what makesPAssertwork on the streaming path.Stateful execution. One generic
transformWithStateoperator,StatefulProcessor<byte[], byte[], byte[]>withEncoders.BINARY()throughout, hosting anyDoFnthroughDoFnRunners. Stateful ParDo goes throughsimpleRunnerplusdefaultStatefulDoFnRunner, GBK throughGroupAlsoByWindowViaWindowSetNewDoFnplusSystemReduceFn.bufferingandKeyedWorkItems, the FlinkWindowDoFnOperatorrecipe. State is a port of the legacySparkStateInternalsonto a singleMapState<String, byte[]>keyed by namespace plus tag, the only layout that can hostReduceFnRunner's dynamically created system tags. Per@StateIdcolumn families are an obvious follow up. Allowed lateness, late firings after the end of window and accumulating panes are supported.Lifecycle.
StreamingEvaluationContextstarts one query per leaf against thenoopsink and blocks until all are terminal.cancel()reaches it through anAtomicReferenceset by the async translate task. Tests terminate through an idle stop listener that gracefully stops a query after N consecutive empty micro-batches.Results
:runners:spark:3:test:runners:spark:4:test84 of the Spark 4 tests are structured streaming, 25 of them the source protocol tests from #39971.
Asserted end to end: stateless ParDo, fixed and sliding window GBK, late data dropping and late firings, accumulating panes, stateful dedup with an event time timer, chained stateful then windowed sum,
PAssertin global and non global windows, restart from a previous run's checkpoint, and pipeline lifecycle through RUNNING, DONE and CANCELLED.The chained case is the one that matters, and it is asserted against Spark's own
StreamingQueryProgressrather than only against the computed values: one query id across every batch, exactly twotransformWithStateExecoperators per batch, and four strictly increasing watermarks.One finding worth flagging on its own
Spark expires a
transformWithStatewake up atexpiry <= batchWatermark. Beam fires an event time timer only once the watermark is strictly past it, andAfterWatermark.pastEndOfWindowis strict as well. Handing Beam a timer one millisecond early is destructive: the trigger declines, the runner is entitled to assume the timer will not be redelivered, and the on time pane is silently lost with no error anywhere. The end of window timer of a fixed window sits at exactlywindow.maxTimestamp(), so this hits most windows.Fixed by bounding the fire set at
min(firedExpiry, watermark - 1)and re-arming the withheld timer atfiredExpiry + 1. Anyone else bridging Beam timers ontotransformWithStatewill hit this.Rejected at translation time, with a named message
Merging windows, triggers other than the default,
AfterWatermark.pastEndOfWindowwith an optional late firing andNever, processing time timers,@OnWindowExpiration,@RequiresTimeSortedInput, side inputs on a stateful ParDo, non deterministic key coders.Known limitations
stop()does not drain. An in flight micro-batch finishes, anything not yet pulled is left unprocessed.spark.speculationis not supported for sources with non deterministic reads.Combine.PerKey.Not done yet, deliberately
@Ignore("TODO: Reactivate with streaming."). They cannot be un-ignored in place, since the shared base test tree is compiled by both modules and Spark 3 still correctly throws. Reactivating them means moving them into the Spark 4 override tree, planned with slice 5.How this is being split
maxRecordsPerBatchoption reuse merged as [Spark][#36841] Reuse the legacy maxRecordsPerBatch option in the Structured Streaming runner #39952.Slices 4 and 5 would benefit from a reviewer who knows Spark's Structured Streaming internals, not only Beam.