f-loris opened a new pull request, #39793: URL: https://github.com/apache/beam/pull/39793
Adds support for stateful ParDo — state, timers and `@RequiresTimeSortedInput` — to the Spark Structured Streaming (Dataset-based) runner in batch mode, including additional (tagged) outputs. Fixes #39779 ### Approach Stateful, timer-using and `@RequiresTimeSortedInput` `DoFn`s are dispatched to a new `StatefulParDoTranslatorBatch`, which groups by key via `KeyValueGroupedDataset#flatMapSortedGroups` and orders each group by event time. `StatefulDoFnGroupFunction` then runs the `DoFn` per key with `InMemoryStateInternals`/`InMemoryTimerInternals` from `runners-core`. Because batch has no state store to bridge onto — state is a heap object scoped to a single key — every Beam state type is already implemented by `InMemoryStateInternals`, so the state categories are enabled together rather than incrementally. Memory is bounded by the state one key holds rather than by that key's element count: groups stream off Spark's spillable sort instead of being materialized, unlike the classic runner's RDD `groupByKey`. Sorting uses the `timestamp` column of the `WindowedValue` encoder, which is `LongType` epoch millis, with `asc_nulls_last`: a null timestamp encodes `END_OF_WINDOW`, which no concrete timestamp of the same window can exceed. The `DoFn` is set up once per task and torn down from a `TaskCompletionListener`, with state, timers and a bundle per key. ### ValidatesRunner Enabled for `validatesStructuredStreamingRunnerBatch`: `UsesStatefulParDo`, `UsesKeyInParDo`, `UsesTimersInParDo`, `UsesMapState`, `UsesMultimapState`, `UsesSetState`, `UsesOrderedListState`, `UsesTimerMap`. Results on this branch: | Suite | Result | |---|---| | `:runners:spark:3:validatesStructuredStreamingRunnerBatch` | 304 tests, 0 failures | | `:runners:spark:4:validatesStructuredStreamingRunnerBatch` | 303 tests, 0 failures | Also added: unit tests for translator dispatch and preconditions, and execution tests covering per-key state, state isolation across many keys in one partition, `@RequiresTimeSortedInput` ordering, event-time timers, looping timers, `@FinishBundle` ordering relative to timers, and tagged outputs. ### Notes for reviewers - **No CI covers this suite on Spark 4.** `beam_PostCommit_Java_ValidatesRunner_Spark4` runs `:runners:spark:4:validatesRunner` (the classic runner), and `beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming` is pinned to `:runners:spark:3:...`. This is pre-existing — the suite has never run on Spark 4 — so the 303/0 above was verified locally. Happy to add a `beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming4` workflow here or as a follow-up if you'd like that coverage; it would be a near-copy of the existing one with `java-version: '17'`. - **Spark 3.4+ requirement.** `flatMapSortedGroups` was added in Spark 3.4, so stateful pipelines are rejected at translation time on 3.1–3.3 with a clear message rather than failing with `NoSuchMethodError` mid-job. The stateful tests skip on those versions so `sparkVersion31Test`/`sparkVersion33Test` stay green. - **`PerKeyOrderingTest#testMultipleStatefulOrdering*` are excluded by name**, not by category. They build on `PeriodicImpulse` and so are unbounded, but are not categorized as `UsesUnboundedPCollections`; adding that category upstream would also stop Dataflow and Flink running them, so the exclusion is kept local to this runner. - **The `ParDoTest` OrderedListState range tests gain `UsesOnWindowExpiration`**, which their `DoFn`s declare but which they were not categorized for. This is the one change outside `runners/spark`. It is a no-op for other runners: every runner excluding `UsesOnWindowExpiration` also excludes `UsesOrderedListState`, so those tests did not run there. - **Deliberate mirroring of the stateless path.** `createEncoders`, `createSideInputReader`, the output receivers and the persist/split-by-tag block intentionally follow `ParDoTranslatorBatch`/`DoFnPartitionIteratorFactory`. Happy to consolidate into shared helpers if preferred. - **`@OnWindowExpiration` remains unsupported** (#22524) and is rejected explicitly; `UsesOnWindowExpiration` stays excluded. Processing-time timers re-armed from `@OnTimer` do not re-fire in batch, matching the classic runner; this is documented on `fireNextTimer`. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
