[Spark][#36841] Translate stateless streaming pipelines on the Spark 4 runner by tkaymak · Pull Request #40090 · apache/beam

tkaymak · 2026-09-10T11:10:58Z

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 [Spark][#36841] Add the DataSourceV2 unbounded source for the Spark 4 streaming runner #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

Makes the DataSourceV2 unbounded source from apache#39971 reachable. The Spark 4 module overrides PipelineTranslatorFactory and dispatches streaming pipelines to PipelineTranslatorStreaming, which translates unbounded reads and reuses the batch translators for stateless single output ParDo, Window.Assign, Flatten and Reshuffle. GroupByKey, Combine.perKey, stateful ParDo, ParDo with side inputs or additional outputs, Impulse and bounded reads fail at translation, the batch translators for them persist or collect the Dataset, which Spark rejects on a streaming plan. StreamingEvaluationContext runs one noop sink query per leaf, checkpoints under checkpointDir/<leaf index>, stops siblings when a query fails and stops a query after streamingStopAfterIdleBatches triggers without input. The test source of BeamMicroBatchSourceTest moves to TestUnboundedSource so the translator tests share it.

github-actions · 2026-09-10T12:46:57Z

Assigning reviewers:

R: @chamikaramj added as fallback since no labels match configuration

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

tkaymak · 2026-09-10T12:49:49Z

Waiting for #40093 to be merged, then the Spark Precommit can run here

Abacn · 2026-09-10T13:43:51Z

Thanks. Close and reopen PR to triggger tests

Abacn

Thanks, had a few comments.

Abacn · 2026-09-10T20:47:20Z

+    return super.getTransformTranslator(transform);
+  }
+
+  private static UnsupportedOperationException unsupported(String what) {


This is a thin wrapper. In convention exceptions should be created at the exact moment an error occurs.

Abacn · 2026-09-10T20:47:56Z

+
+  /** Returns a {@link TransformTranslator} for the given {@link PTransform} if known. */
+  @Override
+  @SuppressWarnings({"rawtypes", "unchecked"})


Please clean up SuppressWarnings

Abacn · 2026-09-10T20:49:44Z

+      LOG.warn(
+          "Error while stopping streaming query {}: {}",
+          query.id(),
+          String.valueOf(e.getMessage()));


redundant String.valueOf(String)

Abacn · 2026-09-10T20:51:40Z

+    assertEquals(PipelineResult.State.CANCELLED, cancelledState);
+    assertEquals(PipelineResult.State.CANCELLED, result.getState());
+
+    long deadline = System.currentTimeMillis() + QUERY_START_TIMEOUT_MILLIS;


This almost duplicates awaitQueryStarted() except the loop condition. Consider generalize awaitQueryStarted -> awaitQuery(...)

Abacn · 2026-09-10T20:51:56Z

+    assertEquals(PipelineResult.State.FAILED, result.getState());
+
+    long deadline = System.currentTimeMillis() + QUERY_START_TIMEOUT_MILLIS;
+    while (SESSION.getSession().streams().active().length > 0) {


same here (awaitQueryStarted)

Abacn · 2026-09-10T20:53:16Z

+  public void tearDown() {
+    TestUnboundedSource.forget("lifecycle-done");
+    TestUnboundedSource.forget("lifecycle-cancel");
+    TestUnboundedSource.forget("lifecycle-healthy");


This is unusual. Generic teardown that runs after every test seems to clear some test specific resources.

Abacn · 2026-09-10T20:58:03Z

+
+  /** Returns a snapshot of everything collected so far under {@code collectorId}. */
+  @SuppressWarnings("unchecked")
+  public static <T> List<T> getCollected(String collectorId) {


getCollected defensively new an ArrayList, while it's callers also redundantly new an ArrayList / Set.

Abacn · 2026-09-10T21:02:11Z

+ * for them persist or collect the Dataset, which Spark rejects for streaming plans.
+ */
+@Internal
+public class PipelineTranslatorStreaming extends PipelineTranslatorBatch {


While it works and resuses some code, future change in PipelineTranslatorBatch may result in surprises on streaming path. For example, new translations meant only for batch now silently port to streaming. On the other hand, refactoring it sounds an risky choice either. At minimum can we rename "PipelineTranslatorBatch" to "PipelineTranslatorCommon" and create a thin subclass "PipelineTranslatorBatch" to it?

Abacn · 2026-09-10T21:06:15Z

+        toAwait = new ArrayList<>(queries);
+      }
+      awaitTermination(toAwait);
+    } finally {


Is there a risk of leak query on exception thrown in the try block? should we call stopQuery for all remaining queries in a catch?

Abacn · 2026-09-10T21:08:10Z

+          .writeStream()
+          .format("noop")
+          .outputMode("append")
+          .option("checkpointLocation", checkpointBaseDir + "/" + leafIndex)


consider use builtin path join method to handle trailing paths in checkpointBaseDir

github-actions Bot added runners spark labels Sep 10, 2026

github-actions Bot added the Next Action: Reviewers label Sep 10, 2026

Abacn closed this Sep 10, 2026

Abacn reopened this Sep 10, 2026

tkaymak closed this Sep 10, 2026

tkaymak reopened this Sep 10, 2026

Abacn reviewed Sep 10, 2026

View reviewed changes

Uh oh!

Conversation

tkaymak commented Sep 10, 2026

Uh oh!

github-actions Bot commented Sep 10, 2026

Uh oh!

tkaymak commented Sep 10, 2026

Uh oh!

Abacn commented Sep 10, 2026

Uh oh!

Abacn left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants