You are viewing a plain text version of this content. The canonical link for it is here.

Posted to github@beam.apache.org by "tkaymak (via GitHub)" <gi...@apache.org> on 2026/08/01 05:27:20 UTC

[PR] [Spark 4] Streaming runner proof of concept, unbounded sources, windowed GBK and stateful ParDo on transformWithState [beam]

tkaymak opened a new pull request, #39576:
URL: https://github.com/apache/beam/pull/39576

   **Draft, opened for design discussion rather than for merging as one lump.** See "How I would like to split this" at the bottom.
   
   Addresses #36841.
   
   ## What this is
   
   The Spark 4 runner merged in #38255 is batch only. `SparkStructuredStreamingRunner` rejects streaming outright. This branch is a proof of concept that Spark 4's `transformWithState` can host the Beam streaming model, and it turns out it can, including several chained stateful operators inside a single Structured Streaming query with a watermark that propagates correctly across them.
   
   I would like to agree on the approach before splitting this into reviewable pieces.
   
   ## Approach
   
   **Dispatch seam.** The shared base gains `translation/PipelineTranslatorFactory.create(boolean streaming)`, which throws for streaming and names the Spark 4 module. `runners/spark/4/src` shadows that one file through the existing source override copy and returns a streaming translator instead. Every piece of `transformWithState`, `StreamingQuery` and DataSourceV2 streaming code lives only under `runners/spark/4/src`, because the shared base still compiles against Spark 3.5 where `transformWithState` does not exist. All shared base edits are behaviour preserving for Spark 3.
   
   **Source.** A DataSourceV2 `TableProvider` wrapping any `UnboundedSource`. Offsets are opaque epoch counters, the driver never reads them. Executor side `PartitionReader`s hold `UnboundedReader`s in a static cache and resume from a locally cached `CheckpointMark`, in the style of the legacy `MicrobatchSource`. The row schema is `payload BINARY` plus `eventTimestamp TIMESTAMP`, so no new 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 `EventTimeWatermark` plan node survives the projection and the typed map, which was the main thing I was unsure about going in. There is a test that asserts this at both the plan level and at runtime.
   
   **Stateful execution.** One generic `transformWithState` operator, `StatefulProcessor<byte[], byte[], byte[]>` with `Encoders.BINARY()` throughout, hosting any `DoFn` through `DoFnRunners`. Stateful ParDo goes through `simpleRunner` plus `defaultStatefulDoFnRunner`, GBK through `GroupAlsoByWindowViaWindowSetNewDoFn` plus `SystemReduceFn.buffering` and `KeyedWorkItems`, which follows the Flink `WindowDoFnOperator` recipe. State is a port of the legacy `SparkStateInternals` onto a single `MapState<String, byte[]>` keyed by namespace plus tag. That layout is the only one I found that can host `ReduceFnRunner`'s dynamically created system tags. Per `@StateId` column families would be faster and are an obvious follow up.
   
   **Lifecycle.** `StreamingEvaluationContext` starts one query per leaf against the `noop` sink and blocks until all are terminal. `cancel()` reaches it through an `AtomicReference` set by the async translate task. Tests terminate through an idle stop listener that gracefully stops a query after N consecutive empty micro batches.
   
   ## Results
   
   | Suite | Tests | Failures | Skipped |
   |---|---|---|---|
   | `:runners:spark:3:test` | 200 | 0 | 7 |
   | `:runners:spark:4:test` | 226 | 0 | 5 |
   
   121 of the Spark 4 tests are structured streaming. The streaming suite was run four separate times with identical results.
   
   Asserted end to end: stateless ParDo, fixed and sliding window GBK, late data dropping, stateful dedup with an event time timer, chained stateful then windowed sum, and pipeline lifecycle through RUNNING, DONE and CANCELLED.
   
   The chained case is the one that matters, and it is asserted against Spark's own `StreamingQueryProgress` rather than only against the computed values: one query id across every batch, exactly two `transformWithStateExec` operators per batch, and four strictly increasing watermarks. That rules out two separate queries that happen to sum correctly, and it rules out everything landing in one batch where all data is trivially on time.
   
   ## One finding worth flagging on its own
   
   Spark expires a `transformWithState` wake up at `expiry <= batchWatermark`. Beam fires an event time timer only once the watermark is **strictly** past it, and `AfterWatermark.pastEndOfWindow` is strict as well. Handing Beam a timer one millisecond early is therefore not merely early, it 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 exactly `window.maxTimestamp()`, so this hits most windows.
   
   Fixed by bounding the fire set at `min(firedExpiry, watermark - 1)` and re-arming the withheld timer at `firedExpiry + 1`. Anyone else bridging Beam timers onto `transformWithState` will hit this.
   
   ## Rejected at translation time, with a named message
   
   Merging and session windows, custom triggers including `AfterWatermark` with early or late firings, accumulating panes, processing time timers, `@OnWindowExpiration`, `@RequiresTimeSortedInput`, side inputs on a stateful ParDo, non deterministic key coders.
   
   ## Known limitations
   
   1. **No `PAssert` on the streaming path.** No source emits a final positive infinity watermark, so panes never finalise. Streaming tests assert against a static collector after `waitUntilFinish`. I would like input on what reviewers expect here, since this is the biggest departure from how the rest of Beam tests runners.
   2. **Weak checkpoint recovery.** Readers resume from a per JVM static cache and the source id is a fresh UUID per translation, so restart from a previous run's checkpoint is not supported. Accepted POC scope, but it is the largest gap between this and something shippable.
   3. **One micro batch of timer latency.** The watermark inside a stateful operator is the batch start watermark, so an end of window timer fires one micro batch after the data crossed the window end. A latency floor, not a correctness issue.
   4. **`stop()` does not drain.** An in flight micro batch finishes, anything not yet pulled is left unprocessed.
   5. Out of scope by design: session windows, custom triggers, accumulating panes, streaming side inputs, Kafka, continuous mode, portability, streaming `Combine.PerKey`.
   
   ## Two smaller things in here
   
   `SparkSessionFactory` now registers `StateSchemaMetadata` and `MemoryWriterCommitMessage` with Kryo, by name since the shared base also compiles against Spark 3. Spark 4 broadcasts a `StateSchemaMetadata` for every `transformWithState` query, so without this any stateful streaming pipeline dies on its first micro batch under `spark.kryo.registrationRequired=true`.
   
   `runners/spark/4/build.gradle` narrows `exclude "**/translation/streaming/**"` to `exclude "**/runners/spark/translation/streaming/**"`. The old glob also matched `structuredstreaming/translation/streaming`, so it would have silently dropped every new test in this branch from the Spark 4 module.
   
   ## Not done yet, deliberately
   
   - No CHANGES.md entry, waiting on the outcome of this discussion.
   - Four tests in the shared base still carry `@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. Happy to do that, it just needs a decision on where they should live.
   
   ## How I would like to split this
   
   Assuming the approach is acceptable, five PRs, each independently green:
   
   1. Dispatch seam, options, lifecycle hooks, build glob fix. Shared base only, no behaviour change for Spark 3.
   2. Kryo registrations for Spark's streaming internals. Independent and self justifying.
   3. DataSourceV2 micro batch source plus tests.
   4. State and timer bridge plus unit tests. This one carries the timer fix described above and deserves its own attention.
   5. Translators, streaming evaluation context, end to end tests.
   
   Slices 4 and 5 would benefit from a reviewer who knows Spark's Structured Streaming internals, not only Beam.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org