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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
