tkaymak opened a new pull request, #40090: URL: https://github.com/apache/beam/pull/40090
Part of #36841, follows #39971 (DataSourceV2 unbounded source). This makes that source reachable: the Spark 4 runner now translates and runs stateless streaming pipelines. **Scope** Supported in streaming mode: `Read.from(UnboundedSource)`, stateless single output `ParDo` without side inputs, `Window.Assign`, `Flatten`, `Reshuffle`. The last four reuse the batch translators unchanged. Everything else fails at translation with an `UnsupportedOperationException` pointing at #36841: `GroupByKey`, `Combine.perKey`, stateful `ParDo`, `ParDo` with side inputs or additional outputs, `Impulse` and bounded reads (so `Create` and `PAssert`). The rejections are explicit because the batch translators for those primitives persist or collect the Dataset, which Spark refuses on a streaming plan with a raw `AnalysisException`. `GroupByKey` and stateful `ParDo` arrive with the `transformWithState` bridge in the next PR. **Main code, four files under `runners/spark/4`** - `translation/PipelineTranslatorFactory.java` shadows the shared base file that throws for streaming today. The Spark 4 module compiles the override tree with later wins, so only Spark 4 gets the streaming dispatch. - `translation/PipelineTranslatorStreaming.java` routes `PrimitiveUnboundedRead` to the new translator, rejects the unsupported primitives, and falls back to the batch registry for the rest. Without the rejections `GroupByKey` would silently run the batch translator against a streaming Dataset. - `translation/StreamingEvaluationContext.java` starts one `noop` sink query per leaf with a processing time trigger of `maxBatchDurationMillis`, checkpoints under `checkpointDir/<leaf index>`, blocks until every query terminates, stops siblings when one fails, and stops a query after `streamingStopAfterIdleBatches` triggers without input when that option is set. Idle triggers arrive as zero row progress events while the source offset moves and as `QueryIdleEvent` otherwise, the listener counts both. `checkpointDir` must be set, the shared default is `/tmp/<jobName>`. - `translation/streaming/ReadUnboundedTranslator.java` builds the Dataset through `UnboundedSourceDataset.of` and decodes the payload column with the full windowed value coder. **Tests, all live `StreamingQuery` runs** - `TestUnboundedSource` is the source that `BeamMicroBatchSourceTest` used as a nested class in #39971, extracted so the translator tests share it. No second synthetic source. - `StatelessParDoStreamingTest`: pass through, and `Flatten` of two unbounded reads. - `StreamingPipelineLifecycleTest`: RUNNING to DONE on idle, cancel, a failing leaf fails the pipeline and stops its healthy sibling. - `StreamingCheckpointRestartTest`: two runs against one checkpoint location with the reader cache wiped in between, the second run recreates readers from the durable marks and re-emits nothing the first run committed. The file layout itself is covered by `BeamMicroBatchSourceTest`, here only the wiring of the checkpoint location is asserted. - `PipelineTranslatorStreamingTest`: the rejections surface from `run()` with the Beam message, not a Spark one. Delivery is at least once, as documented on `BeamReaderCache` in #39971. No `CHANGES.md` entry yet, that comes when the runner can execute a windowed `GroupByKey`. R: @Abacn -- 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]
