diff --git a/runners/spark/4/build.gradle b/runners/spark/4/build.gradle
index ec1af8df38a4..a3da1d8922fd 100644
--- a/runners/spark/4/build.gradle
+++ b/runners/spark/4/build.gradle
@@ -53,11 +53,13 @@ tasks.validatesStructuredStreamingRunnerBatch {
tasks.validatesRunner.dependsOn(validatesStructuredStreamingRunnerBatch)
-// Exclude DStream-based streaming tests from the shared-base copy: the Spark 4 module
-// supports only structured streaming (batch) and does not include legacy DStream support.
+// Exclude legacy DStream-based streaming tests (under runners/spark/translation/streaming)
+// from the shared-base copy: the Spark 4 module does not include legacy DStream support.
// Streaming test utilities also depend on kafka.server.KafkaServerStartable which was
// removed in Kafka 2.8.0 (the first Kafka version with a _2.13 artifact).
+// Note: structured streaming tests (**/structuredstreaming/translation/streaming/**) are
+// intentionally NOT excluded.
tasks.named("copyTestSourceOverrides") {
- exclude "**/translation/streaming/**"
+ exclude "**/runners/spark/translation/streaming/**"
}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamInputPartition.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamInputPartition.java
new file mode 100644
index 000000000000..389ed3fec91e
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamInputPartition.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.io.streaming;
+
+import org.apache.spark.sql.connector.read.InputPartition;
+
+/**
+ * One split of a Beam unbounded source for one micro-batch.
+ *
+ *
Everything the executor needs travels as base64 of the Java serialized object, so the
+ * partition works across JVMs and is not limited to Spark local mode.
+ */
+public class BeamInputPartition implements InputPartition {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String sourceB64;
+ private final String coderB64;
+ private final String pipelineOptionsB64;
+ private final String sourceId;
+ private final int splitId;
+ private final String checkpointLocation;
+ private final int maxRecordsPerMicroBatch;
+ private final long maxBatchDurationMillis;
+
+ BeamInputPartition(
+ String sourceB64,
+ String coderB64,
+ String pipelineOptionsB64,
+ String sourceId,
+ int splitId,
+ String checkpointLocation,
+ int maxRecordsPerMicroBatch,
+ long maxBatchDurationMillis) {
+ this.sourceB64 = sourceB64;
+ this.coderB64 = coderB64;
+ this.pipelineOptionsB64 = pipelineOptionsB64;
+ this.sourceId = sourceId;
+ this.splitId = splitId;
+ this.checkpointLocation = checkpointLocation;
+ this.maxRecordsPerMicroBatch = maxRecordsPerMicroBatch;
+ this.maxBatchDurationMillis = maxBatchDurationMillis;
+ }
+
+ String sourceB64() {
+ return sourceB64;
+ }
+
+ String coderB64() {
+ return coderB64;
+ }
+
+ String pipelineOptionsB64() {
+ return pipelineOptionsB64;
+ }
+
+ String sourceId() {
+ return sourceId;
+ }
+
+ int splitId() {
+ return splitId;
+ }
+
+ String checkpointLocation() {
+ return checkpointLocation;
+ }
+
+ int maxRecordsPerMicroBatch() {
+ return maxRecordsPerMicroBatch;
+ }
+
+ long maxBatchDurationMillis() {
+ return maxBatchDurationMillis;
+ }
+
+ @Override
+ public String toString() {
+ return "BeamInputPartition{source=" + sourceId + ", split=" + splitId + "}";
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamMicroBatchStream.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamMicroBatchStream.java
new file mode 100644
index 000000000000..33d9198412d6
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamMicroBatchStream.java
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.io.streaming;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.runners.core.construction.SerializablePipelineOptions;
+import org.apache.beam.sdk.io.UnboundedSource;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.spark.sql.connector.read.InputPartition;
+import org.apache.spark.sql.connector.read.PartitionReaderFactory;
+import org.apache.spark.sql.connector.read.streaming.MicroBatchStream;
+import org.apache.spark.sql.connector.read.streaming.Offset;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A {@link MicroBatchStream} over a Beam {@link UnboundedSource}.
+ *
+ *
Offsets are opaque epoch counters, see {@link BeamOffset}. {@link #latestOffset()} always
+ * reports a value greater than the previous one so Spark keeps scheduling micro-batches, even ones
+ * that turn out to be empty. Termination of a streaming pipeline is therefore never driven by the
+ * offsets, it is driven by the lifecycle owner (the idle batch listener of the evaluation context,
+ * or an explicit {@code StreamingQuery.stop()}).
+ *
+ *
The wrapped source is split exactly once, on the driver, and the resulting sub sources are
+ * cached so every micro-batch plans the same, stable set of partitions. Splits must be stable
+ * across micro-batches because the executor side reader cache is keyed by split index.
+ */
+public class BeamMicroBatchStream implements MicroBatchStream {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BeamMicroBatchStream.class);
+
+ private final String sourceB64;
+ private final String coderB64;
+ private final String pipelineOptionsB64;
+ private final String sourceId;
+ private final String checkpointLocation;
+ private final int desiredNumSplits;
+ private final int maxRecordsPerMicroBatch;
+ private final long maxBatchDurationMillis;
+
+ private long epoch;
+ private @Nullable List splitsB64;
+
+ BeamMicroBatchStream(CaseInsensitiveStringMap options, String checkpointLocation) {
+ this.sourceB64 = BeamStreamingSource.required(options, BeamStreamingSource.OPT_SOURCE);
+ this.coderB64 = BeamStreamingSource.required(options, BeamStreamingSource.OPT_CODER);
+ this.pipelineOptionsB64 =
+ BeamStreamingSource.required(options, BeamStreamingSource.OPT_PIPELINE_OPTIONS);
+ this.sourceId = BeamStreamingSource.required(options, BeamStreamingSource.OPT_SOURCE_ID);
+ this.checkpointLocation = checkpointLocation;
+ this.desiredNumSplits = Math.max(1, options.getInt(BeamStreamingSource.OPT_NUM_SPLITS, 1));
+ this.maxRecordsPerMicroBatch =
+ Math.max(1, options.getInt(BeamStreamingSource.OPT_MAX_RECORDS, 1000));
+ this.maxBatchDurationMillis =
+ Math.max(1L, options.getLong(BeamStreamingSource.OPT_MAX_BATCH_DURATION_MILLIS, 500L));
+ }
+
+ @Override
+ public Offset initialOffset() {
+ return BeamOffset.ZERO;
+ }
+
+ @Override
+ public synchronized Offset latestOffset() {
+ return new BeamOffset(++epoch);
+ }
+
+ @Override
+ public Offset deserializeOffset(String json) {
+ return BeamOffset.fromJson(json);
+ }
+
+ @Override
+ public void commit(Offset end) {
+ LOG.debug("Committed epoch offset {} of Beam source {}.", end, sourceId);
+ }
+
+ @Override
+ public void stop() {
+ LOG.info("Stopping Beam micro-batch stream for source {}.", sourceId);
+ }
+
+ @Override
+ public InputPartition[] planInputPartitions(Offset start, Offset end) {
+ List splits = splits();
+ InputPartition[] partitions = new InputPartition[splits.size()];
+ for (int i = 0; i < splits.size(); i++) {
+ partitions[i] =
+ new BeamInputPartition(
+ splits.get(i),
+ coderB64,
+ pipelineOptionsB64,
+ sourceId,
+ i,
+ checkpointLocation,
+ maxRecordsPerMicroBatch,
+ maxBatchDurationMillis);
+ }
+ return partitions;
+ }
+
+ @Override
+ public PartitionReaderFactory createReaderFactory() {
+ return new BeamPartitionReaderFactory();
+ }
+
+ /** Splits the wrapped source once and memoizes the serialized sub sources. */
+ private synchronized List splits() {
+ if (splitsB64 != null) {
+ return splitsB64;
+ }
+ UnboundedSource, ?> source = BeamStreamingSource.decode(sourceB64, "UnboundedSource");
+ SerializablePipelineOptions serializableOptions =
+ BeamStreamingSource.decode(pipelineOptionsB64, "PipelineOptions");
+ PipelineOptions options = serializableOptions.get();
+ List extends UnboundedSource, ?>> split;
+ try {
+ split = source.split(desiredNumSplits, options);
+ } catch (Exception e) {
+ throw new IllegalStateException(
+ "Failed to split UnboundedSource " + source.getClass().getCanonicalName(), e);
+ }
+ if (split.isEmpty()) {
+ split = Collections.singletonList(source);
+ }
+ List encoded = new ArrayList<>(split.size());
+ for (UnboundedSource, ?> s : split) {
+ encoded.add(BeamStreamingSource.encode(s));
+ }
+ LOG.info(
+ "Split Beam source {} into {} partition(s) (desired {}).",
+ sourceId,
+ encoded.size(),
+ desiredNumSplits);
+ splitsB64 = encoded;
+ return encoded;
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamOffset.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamOffset.java
new file mode 100644
index 000000000000..f7e5e0e56027
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamOffset.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.io.streaming;
+
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.apache.spark.sql.connector.read.streaming.Offset;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * An opaque epoch counter used as the Spark streaming {@link Offset} of a Beam unbounded source.
+ *
+ *
The offset carries no information about the position inside the wrapped Beam source. The
+ * driver never reads from the source and never inspects its progress, it only needs a monotonically
+ * increasing value so that Spark keeps planning micro-batches. The actual read position lives in
+ * the executor side {@link BeamReaderCache} as a Beam {@code CheckpointMark}.
+ */
+public class BeamOffset extends Offset {
+
+ /** The offset every Beam unbounded stream starts at. */
+ public static final BeamOffset ZERO = new BeamOffset(0L);
+
+ private static final Pattern EPOCH_PATTERN = Pattern.compile("-?\\d+");
+
+ private final long epoch;
+
+ public BeamOffset(long epoch) {
+ this.epoch = epoch;
+ }
+
+ /** The epoch counter value. */
+ public long epoch() {
+ return epoch;
+ }
+
+ @Override
+ public String json() {
+ return "{\"epoch\":" + epoch + "}";
+ }
+
+ /** Parses the form produced by {@link #json()}, a bare number is also accepted. */
+ public static BeamOffset fromJson(String json) {
+ Matcher matcher = EPOCH_PATTERN.matcher(json);
+ if (!matcher.find()) {
+ throw new IllegalArgumentException("Not a valid BeamOffset: " + json);
+ }
+ return new BeamOffset(Long.parseLong(matcher.group()));
+ }
+
+ @Override
+ public boolean equals(@Nullable Object o) {
+ return o instanceof BeamOffset && ((BeamOffset) o).epoch == epoch;
+ }
+
+ @Override
+ public int hashCode() {
+ return Long.hashCode(epoch);
+ }
+
+ @Override
+ public String toString() {
+ return json();
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamPartitionReader.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamPartitionReader.java
new file mode 100644
index 000000000000..ff6c319f01c8
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamPartitionReader.java
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.io.streaming;
+
+import java.io.IOException;
+import org.apache.beam.runners.core.construction.SerializablePipelineOptions;
+import org.apache.beam.runners.spark.structuredstreaming.io.streaming.BeamReaderCache.CachedReader;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.io.UnboundedSource;
+import org.apache.beam.sdk.util.CoderUtils;
+import org.apache.beam.sdk.values.WindowedValue;
+import org.apache.beam.sdk.values.WindowedValues;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.Uninterruptibles;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.connector.read.PartitionReader;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Reads one split of a Beam {@link UnboundedSource} for the duration of one Spark micro-batch.
+ *
+ *
The batch ends as soon as either {@code maxRecordsPerMicroBatch} elements were emitted or
+ * {@code maxBatchDurationMillis} of wall clock time elapsed, whichever comes first. When the source
+ * has no data available the reader polls with a short sleep until the deadline, so an idle source
+ * produces an empty micro-batch rather than blocking the query.
+ *
+ *
The underlying Beam reader is not closed at the end of the batch, it stays in {@link
+ * BeamReaderCache} and the next micro-batch continues from the same position. See that class for
+ * the failure recovery caveats.
+ *
+ * @param the element type of the wrapped source
+ */
+@SuppressWarnings({
+ "nullness" // the current row is only read between a true next() and the following one
+})
+public class BeamPartitionReader implements PartitionReader {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BeamPartitionReader.class);
+
+ /** Sleep between two unsuccessful advance attempts while the batch deadline has not passed. */
+ private static final long POLL_INTERVAL_MILLIS = 10L;
+
+ private final String cacheKey;
+ private final CachedReader cached;
+ private final Coder> windowedValueCoder;
+ private final int maxRecordsPerMicroBatch;
+ private final long maxBatchDurationMillis;
+
+ private long recordsRead;
+ private long deadlineMillis = -1L;
+ private @Nullable InternalRow current;
+
+ BeamPartitionReader(BeamInputPartition partition) {
+ UnboundedSource source =
+ BeamStreamingSource.decode(partition.sourceB64(), "UnboundedSource split");
+ this.windowedValueCoder =
+ BeamStreamingSource.decode(partition.coderB64(), "WindowedValue coder");
+ SerializablePipelineOptions options =
+ BeamStreamingSource.decode(partition.pipelineOptionsB64(), "PipelineOptions");
+ this.maxRecordsPerMicroBatch = partition.maxRecordsPerMicroBatch();
+ this.maxBatchDurationMillis = partition.maxBatchDurationMillis();
+ this.cacheKey =
+ BeamReaderCache.key(
+ partition.checkpointLocation(), partition.sourceId(), partition.splitId());
+ this.cached = BeamReaderCache.getOrCreate(cacheKey, source, options.get());
+ }
+
+ @Override
+ public boolean next() throws IOException {
+ if (deadlineMillis < 0) {
+ deadlineMillis = System.currentTimeMillis() + maxBatchDurationMillis;
+ }
+ while (true) {
+ if (recordsRead >= maxRecordsPerMicroBatch) {
+ current = null;
+ return false;
+ }
+ long remaining = deadlineMillis - System.currentTimeMillis();
+ if (remaining <= 0) {
+ current = null;
+ return false;
+ }
+ if (cached.startOrAdvance()) {
+ recordsRead++;
+ current = toRow();
+ return true;
+ }
+ Uninterruptibles.sleepUninterruptibly(
+ Math.min(remaining, POLL_INTERVAL_MILLIS), java.util.concurrent.TimeUnit.MILLISECONDS);
+ }
+ }
+
+ @Override
+ public InternalRow get() {
+ if (current == null) {
+ throw new IllegalStateException("No current row, next() did not return true.");
+ }
+ return current;
+ }
+
+ /**
+ * Ends the micro-batch. The Beam reader deliberately stays open in {@link BeamReaderCache}, only
+ * its checkpoint mark is remembered and finalized.
+ */
+ @Override
+ public void close() {
+ current = null;
+ try {
+ UnboundedSource.CheckpointMark mark = cached.reader().getCheckpointMark();
+ BeamReaderCache.rememberCheckpointMark(cacheKey, mark);
+ mark.finalizeCheckpoint();
+ } catch (Exception e) {
+ LOG.warn("Failed to finalize the checkpoint mark of Beam reader {}.", cacheKey, e);
+ }
+ LOG.debug("Beam reader {} emitted {} record(s) in this micro-batch.", cacheKey, recordsRead);
+ }
+
+ private InternalRow toRow() {
+ Instant timestamp = cached.reader().getCurrentTimestamp();
+ WindowedValue windowedValue =
+ WindowedValues.timestampedValueInGlobalWindow(cached.reader().getCurrent(), timestamp);
+ byte[] payload;
+ try {
+ payload = CoderUtils.encodeToByteArray(windowedValueCoder, windowedValue);
+ } catch (CoderException e) {
+ throw new IllegalStateException("Failed to encode element read from a Beam source.", e);
+ }
+ // Spark stores TimestampType as microseconds since the epoch.
+ return new GenericInternalRow(new Object[] {payload, timestamp.getMillis() * 1000L});
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamPartitionReaderFactory.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamPartitionReaderFactory.java
new file mode 100644
index 000000000000..db2573eff315
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamPartitionReaderFactory.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.io.streaming;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.read.InputPartition;
+import org.apache.spark.sql.connector.read.PartitionReader;
+import org.apache.spark.sql.connector.read.PartitionReaderFactory;
+
+/** Creates a {@link BeamPartitionReader} for a {@link BeamInputPartition} on the executor. */
+public class BeamPartitionReaderFactory implements PartitionReaderFactory {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public PartitionReader createReader(InputPartition partition) {
+ return new BeamPartitionReader<>((BeamInputPartition) partition);
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamReaderCache.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamReaderCache.java
new file mode 100644
index 000000000000..9ccfc3388bd3
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamReaderCache.java
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.io.streaming;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+import org.apache.beam.sdk.io.UnboundedSource;
+import org.apache.beam.sdk.io.UnboundedSource.CheckpointMark;
+import org.apache.beam.sdk.io.UnboundedSource.UnboundedReader;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.Cache;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.CacheBuilder;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.RemovalListener;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Executor side cache of live Beam {@link UnboundedReader}s, keyed by (checkpoint location, source
+ * id, split id).
+ *
+ *
A Spark micro-batch creates a fresh {@link BeamPartitionReader} every batch, but a Beam
+ * unbounded reader is expensive to create and, more importantly, holds the read position. Keeping
+ * the reader alive between micro-batches lets the next batch simply continue where the previous one
+ * stopped. This mirrors what the legacy Spark runner does in {@code
+ * org.apache.beam.runners.spark.io.MicrobatchSource}.
+ *
+ *
POC scope, weak failure recovery. The {@code CheckpointMark} of every split is cached
+ * in executor memory only, it is never written to the Spark checkpoint location and never restored
+ * from it. If an executor JVM dies, or the cache entry expires, the reader is recreated from the
+ * last mark that happens to still be in memory, and from scratch if there is none. Data loss or
+ * duplication across executor failures is therefore possible. Making recovery durable means
+ * committing marks through the Spark offset log and is deliberately out of scope for the POC.
+ */
+public final class BeamReaderCache {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BeamReaderCache.class);
+
+ /** Readers idle for longer than this are closed, releasing the underlying source connections. */
+ private static final long READER_CACHE_INTERVAL_MILLIS = 10 * 60 * 1000L;
+
+ private static final RemovalListener> CLOSE_ON_REMOVAL =
+ notification -> {
+ CachedReader> reader = notification.getValue();
+ String key = String.valueOf(notification.getKey());
+ if (reader != null) {
+ LOG.info("Evicting cached Beam reader {}.", key);
+ try {
+ reader.close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close evicted Beam reader {}.", key, e);
+ }
+ }
+ };
+
+ private static final Cache> READERS =
+ CacheBuilder.newBuilder()
+ .expireAfterAccess(READER_CACHE_INTERVAL_MILLIS, TimeUnit.MILLISECONDS)
+ .removalListener(CLOSE_ON_REMOVAL)
+ .build();
+
+ /** Last known checkpoint mark per key, used when a reader has to be recreated. */
+ private static final ConcurrentMap MARKS = new ConcurrentHashMap<>();
+
+ private BeamReaderCache() {}
+
+ /** Builds the cache key of one split of one source of one streaming query. */
+ public static String key(String checkpointLocation, String sourceId, int splitId) {
+ return checkpointLocation + '|' + sourceId + '|' + splitId;
+ }
+
+ /**
+ * Returns the cached reader for {@code key}, creating it from the last cached checkpoint mark if
+ * there is none.
+ */
+ @SuppressWarnings({"unchecked", "nullness"}) // the mark type always matches the source
+ public static CachedReader getOrCreate(
+ String key, UnboundedSource source, PipelineOptions options) {
+ try {
+ return (CachedReader)
+ READERS.get(
+ key,
+ () -> {
+ CheckpointMarkT mark = (CheckpointMarkT) MARKS.get(key);
+ LOG.info(
+ "No cached Beam reader for {}, creating one at checkpoint mark {}.", key, mark);
+ return new CachedReader<>(source.createReader(options, mark));
+ });
+ } catch (Exception e) {
+ throw new IllegalStateException("Failed to get or create Beam unbounded reader " + key, e);
+ }
+ }
+
+ /** Remembers the checkpoint mark of {@code key} so a recreated reader can resume from it. */
+ public static void rememberCheckpointMark(String key, @Nullable CheckpointMark mark) {
+ if (mark != null) {
+ MARKS.put(key, mark);
+ }
+ }
+
+ /** Closes and forgets every cached reader, intended for tests and for query shutdown. */
+ @VisibleForTesting
+ public static void invalidateAll() {
+ READERS.invalidateAll();
+ READERS.cleanUp();
+ MARKS.clear();
+ }
+
+ /** A cached {@link UnboundedReader} that remembers whether it has been started already. */
+ public static final class CachedReader implements Closeable {
+ private final UnboundedReader reader;
+ private boolean started;
+
+ CachedReader(UnboundedReader reader) {
+ this.reader = reader;
+ }
+
+ /** The wrapped Beam reader. */
+ public UnboundedReader reader() {
+ return reader;
+ }
+
+ /**
+ * Starts the reader on first use and advances it afterwards, returning {@code true} if an
+ * element is available.
+ */
+ public synchronized boolean startOrAdvance() throws IOException {
+ if (!started) {
+ started = true;
+ return reader.start();
+ }
+ return reader.advance();
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamStreamingSource.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamStreamingSource.java
new file mode 100644
index 000000000000..b9a1100f11e4
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamStreamingSource.java
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.io.streaming;
+
+import java.io.Serializable;
+import java.util.Base64;
+import java.util.Map;
+import org.apache.beam.sdk.util.SerializableUtils;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.TableProvider;
+import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.sources.DataSourceRegister;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+/**
+ * Spark DataSourceV2 {@link TableProvider} exposing an arbitrary Beam {@link
+ * org.apache.beam.sdk.io.UnboundedSource} as a micro-batch streaming source.
+ *
+ *
The produced rows always have exactly two columns:
+ *
+ *
+ *
{@code payload} of type {@code BINARY}, holding the element encoded with the Beam {@code
+ * WindowedValues.FullWindowedValueCoder} supplied by the translator.
+ *
{@code eventTimestamp} of type {@code TIMESTAMP}, holding the event timestamp reported by
+ * the Beam reader for that element.
+ *
+ *
+ *
Deliberately no Catalyst encoder is generated for Beam types, everything stays opaque bytes
+ * until a downstream translator decodes it.
+ *
+ *
Translators should not reference this class directly, use {@link UnboundedSourceDataset#of}
+ * instead.
+ *
+ *
Note on the format name: this class implements {@link DataSourceRegister} and reports the
+ * short name {@value #SHORT_NAME}, but no {@code META-INF/services} entry is shipped, so the short
+ * name is not resolvable through the {@code ServiceLoader}. Use {@link #FORMAT}, the fully
+ * qualified class name, as the argument of {@code DataStreamReader.format(String)}.
+ */
+public class BeamStreamingSource implements TableProvider, DataSourceRegister {
+
+ /** Short name of this source, see the class level note about {@code META-INF/services}. */
+ public static final String SHORT_NAME = "beam-unbounded";
+
+ /** Format string to pass to {@code DataStreamReader.format(String)}. */
+ public static final String FORMAT =
+ "org.apache.beam.runners.spark.structuredstreaming.io.streaming.BeamStreamingSource";
+
+ /**
+ * Base64 of the Java serialized {@link org.apache.beam.sdk.io.UnboundedSource}.
+ *
+ *
All option keys are lower case on purpose, Spark funnels DataSourceV2 options through {@link
+ * CaseInsensitiveStringMap}.
+ */
+ public static final String OPT_SOURCE = "beam.source";
+
+ /** Base64 of the Java serialized {@code Coder>}. */
+ public static final String OPT_CODER = "beam.coder";
+
+ /**
+ * Base64 of the Java serialized {@link
+ * org.apache.beam.runners.core.construction.SerializablePipelineOptions}.
+ */
+ public static final String OPT_PIPELINE_OPTIONS = "beam.pipelineoptions";
+
+ /** Identifier making the reader cache key unique per source instance. */
+ public static final String OPT_SOURCE_ID = "beam.sourceid";
+
+ /** Desired number of splits handed to {@code UnboundedSource.split}. */
+ public static final String OPT_NUM_SPLITS = "beam.numsplits";
+
+ /** Maximum number of records read per split per micro-batch. */
+ public static final String OPT_MAX_RECORDS = "beam.maxrecords";
+
+ /** Maximum wall clock duration of a single micro-batch read, in milliseconds. */
+ public static final String OPT_MAX_BATCH_DURATION_MILLIS = "beam.maxbatchdurationmillis";
+
+ /** The fixed two column schema of this source. */
+ public static final StructType SCHEMA =
+ new StructType()
+ .add(UnboundedSourceDataset.COL_PAYLOAD, DataTypes.BinaryType, false)
+ .add(UnboundedSourceDataset.COL_EVENT_TS, DataTypes.TimestampType, false);
+
+ /** Required public no-arg constructor, Spark instantiates this provider reflectively. */
+ public BeamStreamingSource() {}
+
+ @Override
+ public String shortName() {
+ return SHORT_NAME;
+ }
+
+ @Override
+ public StructType inferSchema(CaseInsensitiveStringMap options) {
+ return SCHEMA;
+ }
+
+ @Override
+ public Table getTable(
+ StructType schema, Transform[] partitioning, Map properties) {
+ return new BeamStreamingTable(new CaseInsensitiveStringMap(properties));
+ }
+
+ @Override
+ public boolean supportsExternalMetadata() {
+ return false;
+ }
+
+ /** Base64 encodes the Java serialized form of {@code value}. */
+ static String encode(Serializable value) {
+ return Base64.getEncoder().encodeToString(SerializableUtils.serializeToByteArray(value));
+ }
+
+ /** Inverse of {@link #encode}, {@code description} is only used in error messages. */
+ @SuppressWarnings("unchecked")
+ static T decode(String encoded, String description) {
+ return (T)
+ SerializableUtils.deserializeFromByteArray(
+ Base64.getDecoder().decode(encoded), description);
+ }
+
+ /** Reads a required option, failing loudly rather than silently defaulting. */
+ static String required(CaseInsensitiveStringMap options, String key) {
+ String value = options.get(key);
+ if (value == null) {
+ throw new IllegalArgumentException(
+ "Missing required option '" + key + "' for " + SHORT_NAME + " source.");
+ }
+ return value;
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamStreamingTable.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamStreamingTable.java
new file mode 100644
index 000000000000..4ba6c24a5cef
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/BeamStreamingTable.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.io.streaming;
+
+import java.util.Set;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
+import org.apache.spark.sql.connector.catalog.SupportsRead;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.TableCapability;
+import org.apache.spark.sql.connector.read.Scan;
+import org.apache.spark.sql.connector.read.ScanBuilder;
+import org.apache.spark.sql.connector.read.streaming.MicroBatchStream;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+/**
+ * The {@link Table} returned by {@link BeamStreamingSource}. It only ever declares {@link
+ * TableCapability#MICRO_BATCH_READ}, batch reads of an unbounded Beam source are handled elsewhere.
+ */
+public class BeamStreamingTable implements Table, SupportsRead {
+
+ private final CaseInsensitiveStringMap options;
+
+ BeamStreamingTable(CaseInsensitiveStringMap options) {
+ this.options = options;
+ }
+
+ @Override
+ public String name() {
+ return "BeamUnboundedSource["
+ + options.getOrDefault(BeamStreamingSource.OPT_SOURCE_ID, "?")
+ + "]";
+ }
+
+ @Override
+ public StructType schema() {
+ return BeamStreamingSource.SCHEMA;
+ }
+
+ @Override
+ public Set capabilities() {
+ return ImmutableSet.of(TableCapability.MICRO_BATCH_READ);
+ }
+
+ @Override
+ public ScanBuilder newScanBuilder(CaseInsensitiveStringMap scanOptions) {
+ // Spark hands the full DataSourceV2 option map to both getTable and newScanBuilder. Prefer the
+ // scan options and fall back to the table properties for anything missing.
+ CaseInsensitiveStringMap merged = merge(options, scanOptions);
+ return () -> new BeamScan(merged);
+ }
+
+ private static CaseInsensitiveStringMap merge(
+ CaseInsensitiveStringMap base, CaseInsensitiveStringMap override) {
+ java.util.Map map = new java.util.HashMap<>(base.asCaseSensitiveMap());
+ map.putAll(override.asCaseSensitiveMap());
+ return new CaseInsensitiveStringMap(map);
+ }
+
+ /** The {@link Scan} of a Beam unbounded source, micro-batch only. */
+ private static class BeamScan implements Scan {
+ private final CaseInsensitiveStringMap options;
+
+ BeamScan(CaseInsensitiveStringMap options) {
+ this.options = options;
+ }
+
+ @Override
+ public StructType readSchema() {
+ return BeamStreamingSource.SCHEMA;
+ }
+
+ @Override
+ public String description() {
+ return "BeamUnboundedSource["
+ + options.getOrDefault(BeamStreamingSource.OPT_SOURCE_ID, "?")
+ + "]";
+ }
+
+ @Override
+ public MicroBatchStream toMicroBatchStream(String checkpointLocation) {
+ return new BeamMicroBatchStream(options, checkpointLocation);
+ }
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/UnboundedSourceDataset.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/UnboundedSourceDataset.java
new file mode 100644
index 000000000000..7e4af6687cfc
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/io/streaming/UnboundedSourceDataset.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.io.streaming;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.beam.runners.core.construction.SerializablePipelineOptions;
+import org.apache.beam.runners.spark.structuredstreaming.SparkStructuredStreamingPipelineOptions;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.io.UnboundedSource;
+import org.apache.beam.sdk.values.WindowedValue;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+
+/**
+ * Translator facing entry point turning a Beam {@link UnboundedSource} into a streaming Spark
+ * {@link Dataset} of rows.
+ *
+ *
The returned dataset has exactly two columns, {@value #COL_PAYLOAD} of type {@code BINARY}
+ * carrying the element encoded with the supplied {@code WindowedValue} coder, and {@value
+ * #COL_EVENT_TS} of type {@code TIMESTAMP} carrying the event timestamp of that element.
+ *
+ *
The watermark is declared here and only here. Spark 4 rejects a second {@code
+ * withWatermark} declaration further down the plan, so this is the single declaration point for a
+ * whole Beam pipeline. Downstream translators must never call {@code withWatermark} again, they
+ * simply keep transforming the dataset. Both columns are still present in the returned dataset, the
+ * event timestamp column has to survive at least until the first stateful operator for the
+ * watermark to be meaningful.
+ */
+public final class UnboundedSourceDataset {
+
+ /** Name of the binary column holding the encoded {@code WindowedValue}. */
+ public static final String COL_PAYLOAD = "payload";
+
+ /** Name of the timestamp column holding the Beam event timestamp. */
+ public static final String COL_EVENT_TS = "eventTimestamp";
+
+ /** Upper bound on the number of splits requested from a source, keeps the POC predictable. */
+ private static final int MAX_DESIRED_SPLITS = 8;
+
+ private UnboundedSourceDataset() {}
+
+ /**
+ * Builds the streaming {@link Dataset} for {@code source}, with the event time watermark already
+ * applied.
+ *
+ * @param session the active Spark session
+ * @param source the Beam unbounded source to read
+ * @param windowedValueCoder the coder used to encode the {@value #COL_PAYLOAD} column, normally a
+ * {@code WindowedValues.FullWindowedValueCoder}
+ * @param options the pipeline options, supplying the watermark delay and the micro-batch limits
+ * @param the element type of the source
+ * @param the checkpoint mark type of the source
+ */
+ public static Dataset of(
+ SparkSession session,
+ UnboundedSource source,
+ Coder> windowedValueCoder,
+ SparkStructuredStreamingPipelineOptions options) {
+
+ Map readerOptions = new HashMap<>();
+ readerOptions.put(BeamStreamingSource.OPT_SOURCE, BeamStreamingSource.encode(source));
+ readerOptions.put(
+ BeamStreamingSource.OPT_CODER, BeamStreamingSource.encode(windowedValueCoder));
+ readerOptions.put(
+ BeamStreamingSource.OPT_PIPELINE_OPTIONS,
+ BeamStreamingSource.encode(new SerializablePipelineOptions(options)));
+ readerOptions.put(BeamStreamingSource.OPT_SOURCE_ID, UUID.randomUUID().toString());
+ readerOptions.put(
+ BeamStreamingSource.OPT_NUM_SPLITS, Integer.toString(desiredNumSplits(session)));
+ readerOptions.put(
+ BeamStreamingSource.OPT_MAX_RECORDS,
+ Integer.toString(options.getMaxRecordsPerMicroBatch()));
+ readerOptions.put(
+ BeamStreamingSource.OPT_MAX_BATCH_DURATION_MILLIS,
+ Long.toString(options.getMaxBatchDurationMillis()));
+
+ Dataset rows =
+ session.readStream().format(BeamStreamingSource.FORMAT).options(readerOptions).load();
+
+ // Exactly one watermark declaration per pipeline, see the class javadoc.
+ return rows.withWatermark(COL_EVENT_TS, options.getWatermarkDelayMillis() + " milliseconds");
+ }
+
+ private static int desiredNumSplits(SparkSession session) {
+ int parallelism = session.sparkContext().defaultParallelism();
+ return Math.max(1, Math.min(MAX_DESIRED_SPLITS, parallelism));
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/PipelineTranslatorFactory.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/PipelineTranslatorFactory.java
new file mode 100644
index 000000000000..ac7f6aa5d6c1
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/PipelineTranslatorFactory.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.translation;
+
+import org.apache.beam.runners.spark.structuredstreaming.translation.batch.PipelineTranslatorBatch;
+import org.apache.beam.sdk.annotations.Internal;
+
+/**
+ * Factory to create the {@link PipelineTranslator} matching the execution mode of the pipeline.
+ *
+ *
This file shadows the shared base version of the same name found under {@code
+ * runners/spark/src}. The Spark 4 module compiles a merged source tree of the shared base plus
+ * {@code runners/spark/4/src}, with a later-wins duplicate strategy, so this copy silently replaces
+ * the base one for the Spark 4 module only. The base version keeps throwing for streaming, this one
+ * dispatches to the real streaming translator.
+ */
+@Internal
+public final class PipelineTranslatorFactory {
+ private PipelineTranslatorFactory() {}
+
+ /** Creates a {@link PipelineTranslator} for the given execution mode. */
+ public static PipelineTranslator create(boolean streaming) {
+ return streaming ? new PipelineTranslatorStreaming() : new PipelineTranslatorBatch();
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/PipelineTranslatorStreaming.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/PipelineTranslatorStreaming.java
new file mode 100644
index 000000000000..7901fb9f4a31
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/PipelineTranslatorStreaming.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.translation;
+
+import java.util.Collection;
+import org.apache.beam.runners.spark.SparkCommonPipelineOptions;
+import org.apache.beam.runners.spark.structuredstreaming.translation.batch.PipelineTranslatorBatch;
+import org.apache.beam.runners.spark.structuredstreaming.translation.streaming.GroupByKeyStreamingTranslator;
+import org.apache.beam.runners.spark.structuredstreaming.translation.streaming.ReadUnboundedTranslator;
+import org.apache.beam.runners.spark.structuredstreaming.translation.streaming.StatefulParDoStreamingTranslator;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignature;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignatures;
+import org.apache.beam.sdk.util.construction.SplittableParDo;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.spark.sql.SparkSession;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * {@link PipelineTranslator} for executing a streaming {@link org.apache.beam.sdk.Pipeline} on
+ * Spark 4.
+ *
+ *
This extends {@link PipelineTranslatorBatch} purely for reuse: {@link
+ * PipelineTranslatorBatch#getTransformTranslator} and its private registry are the only way to
+ * reach the (package-private) batch translators for {@code Impulse}, {@code Window.Assign}, {@code
+ * Flatten}, {@code Reshuffle}, the bounded read, and stateless {@code ParDo}, all of which are
+ * reused completely unchanged for streaming. This class only intercepts the handful of transforms
+ * that need genuinely different, streaming-aware handling before falling back to {@code super}.
+ */
+@Internal
+public class PipelineTranslatorStreaming extends PipelineTranslatorBatch {
+
+ /** Returns a {@link TransformTranslator} for the given {@link PTransform} if known. */
+ @Override
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ @Nullable
+ protected >
+ TransformTranslator getTransformTranslator(TransformT transform) {
+
+ if (transform instanceof SplittableParDo.PrimitiveUnboundedRead) {
+ return (TransformTranslator) new ReadUnboundedTranslator<>();
+ }
+
+ if (transform instanceof GroupByKey) {
+ return (TransformTranslator) new GroupByKeyStreamingTranslator<>();
+ }
+
+ // Deliberately never registered: leaving Combine.PerKey unhandled here makes Beam auto-expand
+ // it into GroupByKey + ParDo, so the streaming translations above take over the expanded
+ // primitives instead of the batch CombinePerKeyTranslatorBatch, which has no streaming
+ // support.
+ if (transform instanceof Combine.PerKey) {
+ return null;
+ }
+
+ if (transform instanceof ParDo.MultiOutput) {
+ DoFnSignature signature =
+ DoFnSignatures.signatureForDoFn(((ParDo.MultiOutput, ?>) transform).getFn());
+ if (signature.usesState() || signature.usesTimers()) {
+ return (TransformTranslator) new StatefulParDoStreamingTranslator<>();
+ }
+ // Stateless ParDo falls through to super, reusing ParDoTranslatorBatch unchanged.
+ }
+
+ // Impulse, Window.Assign, Flatten, Reshuffle, the bounded read, and stateless ParDo: reused
+ // unchanged from the batch registry.
+ return super.getTransformTranslator(transform);
+ }
+
+ @Override
+ protected EvaluationContext createEvaluationContext(
+ Collection extends EvaluationContext.NamedDataset>> leaves,
+ SparkSession session,
+ SparkCommonPipelineOptions options) {
+ return new StreamingEvaluationContext(leaves, session, options);
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/StreamingEvaluationContext.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/StreamingEvaluationContext.java
new file mode 100644
index 000000000000..c45ec0de30ab
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/StreamingEvaluationContext.java
@@ -0,0 +1,306 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.translation;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.beam.runners.spark.SparkCommonPipelineOptions;
+import org.apache.beam.runners.spark.structuredstreaming.SparkStructuredStreamingPipelineOptions;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.streaming.StreamingQuery;
+import org.apache.spark.sql.streaming.StreamingQueryException;
+import org.apache.spark.sql.streaming.StreamingQueryListener;
+import org.apache.spark.sql.streaming.Trigger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The streaming counterpart of {@link EvaluationContext}: instead of forcing a one-shot batch
+ * evaluation of every leaf dataset, it starts one Spark Structured Streaming query per leaf and
+ * blocks until all of them reach a terminal state.
+ *
+ *
Sink choice
+ *
+ *
Every query uses the {@code noop} sink, never {@code memory}. A Beam pipeline emits through
+ * its own sinks inside the leaf {@code DoFn}s, so a leaf dataset's rows have already served their
+ * purpose by the time the Spark sink sees them; the {@code memory} sink would accumulate every one
+ * of them in driver memory for nobody to read. Note that {@code MemoryWriterCommitMessage} is
+ * registered by {@code SparkSessionFactory.SparkKryoRegistrator} anyway, so switching a query to
+ * the {@code memory} sink for debugging no longer trips {@code
+ * spark.kryo.registrationRequired=true}.
+ *
+ *
Termination
+ *
+ *
Queries read from sources with opaque epoch offsets that never settle (see {@code
+ * UnboundedSourceDataset}), so {@code StreamingQueryManager#awaitAnyTermination} without outside
+ * help would hang forever. Two independent knobs exist to terminate a query:
+ *
+ *
+ *
{@link #stop()}, invoked by {@code SparkStructuredStreamingPipelineResult#cancel()}.
+ *
The idle-stop listener registered in {@link #evaluate()} when {@code
+ * SparkStructuredStreamingPipelineOptions#getStreamingStopAfterIdleBatches()} is {@code >=
+ * 0}: it counts consecutive micro-batches with zero input rows per query and gracefully stops
+ * that one query once the threshold is reached. This is how streaming tests in this module
+ * terminate on their own.
+ *
+ *
+ *
Draining on {@link #stop()}
+ *
+ *
{@link #stop()} does not attempt a full {@code Trigger.AvailableNow()} drain pass that
+ * processes every already-buffered offset before halting: doing so would require stopping the query
+ * and restarting it with a different trigger against the same checkpoint, which is more machinery
+ * than this POC's lifecycle warrants and risks checkpoint-compatibility bugs of its own. Instead,
+ * {@link #stop()} relies on {@link StreamingQuery#stop()}'s own graceful behaviour, which lets a
+ * micro-batch that is already in flight finish normally instead of interrupting it, and only then
+ * halts the query. Data that was not yet pulled into an in-flight micro-batch at the moment {@link
+ * #stop()} is called is simply left unprocessed. This is a documented limitation of the POC, not an
+ * oversight.
+ */
+@Internal
+public class StreamingEvaluationContext extends EvaluationContext {
+ private static final Logger LOG = LoggerFactory.getLogger(StreamingEvaluationContext.class);
+
+ private final SparkStructuredStreamingPipelineOptions options;
+
+ // Guards both `queries` and `stopped` so evaluate() (which appends to `queries` as it starts
+ // queries) and stop() (which may run concurrently on another thread, see the class javadoc on
+ // thread-safety below) never race on which queries have been started or already stopped.
+ private final Object lock = new Object();
+ private final List queries = new ArrayList<>();
+ private boolean stopped = false;
+
+ StreamingEvaluationContext(
+ Collection extends NamedDataset>> leaves,
+ SparkSession session,
+ SparkCommonPipelineOptions options) {
+ super(leaves, session);
+ this.options = options.as(SparkStructuredStreamingPipelineOptions.class);
+ }
+
+ /**
+ * Starts one streaming query per leaf dataset and blocks until every one of them has reached a
+ * terminal state, either because {@link #stop()} was called (typically via {@code cancel()}) or
+ * because the idle-stop listener stopped it after enough consecutive empty micro-batches.
+ */
+ @Override
+ public void evaluate() {
+ String checkpointBaseDir = checkpointBaseDir(options);
+ int idleStopThreshold = options.getStreamingStopAfterIdleBatches();
+
+ StreamingQueryListener idleStopListener = null;
+ if (idleStopThreshold >= 0) {
+ idleStopListener = new IdleStopListener(idleStopThreshold);
+ getSparkSession().streams().addListener(idleStopListener);
+ }
+
+ try {
+ int leafIndex = 0;
+ for (NamedDataset> ds : leaves()) {
+ Dataset> dataset = ds.dataset();
+ if (dataset == null) {
+ continue;
+ }
+ synchronized (lock) {
+ if (stopped) {
+ // stop() already ran (e.g. an immediate cancel()); do not start further queries.
+ break;
+ }
+ }
+ if (!dataset.isStreaming()) {
+ // Defensive fallback: a leaf that turns out not to be streaming (e.g. a bounded side
+ // collection) is simply evaluated the batch way instead of starting a query for it.
+ EvaluationContext.evaluate(ds.name(), dataset);
+ continue;
+ }
+
+ StreamingQuery query = startQuery(dataset, checkpointBaseDir, leafIndex++, options);
+ boolean alreadyStopped;
+ synchronized (lock) {
+ queries.add(query);
+ alreadyStopped = stopped;
+ }
+ if (alreadyStopped) {
+ // stop() ran in the window between the check above and this query actually starting.
+ stopQuery(query);
+ }
+ }
+
+ List toAwait;
+ synchronized (lock) {
+ toAwait = new ArrayList<>(queries);
+ }
+ for (StreamingQuery query : toAwait) {
+ awaitTermination(query);
+ }
+ } finally {
+ if (idleStopListener != null) {
+ getSparkSession().streams().removeListener(idleStopListener);
+ }
+ }
+ }
+
+ /**
+ * Stops all queries started by {@link #evaluate()}.
+ *
+ *
Idempotent and safe to call from a thread other than the one running {@link #evaluate()}:
+ * {@code cancel()} calls this from the main thread while {@code evaluate()} is blocked awaiting
+ * termination on the pipeline execution thread. See the class javadoc for the drain limitation.
+ */
+ @Override
+ public void stop() {
+ List toStop;
+ synchronized (lock) {
+ if (stopped) {
+ return;
+ }
+ stopped = true;
+ toStop = new ArrayList<>(queries);
+ }
+ for (StreamingQuery query : toStop) {
+ stopQuery(query);
+ }
+ }
+
+ private StreamingQuery startQuery(
+ Dataset> dataset,
+ String checkpointBaseDir,
+ int leafIndex,
+ SparkStructuredStreamingPipelineOptions options) {
+ try {
+ return dataset
+ .writeStream()
+ .format("noop")
+ .outputMode("append")
+ .option("checkpointLocation", checkpointBaseDir + "/" + leafIndex)
+ .trigger(Trigger.ProcessingTime(options.getMaxBatchDurationMillis()))
+ .start();
+ } catch (TimeoutException e) {
+ throw new RuntimeException(
+ "Failed to start streaming query for leaf dataset index " + leafIndex, e);
+ }
+ }
+
+ private void awaitTermination(StreamingQuery query) {
+ try {
+ query.awaitTermination();
+ } catch (StreamingQueryException e) {
+ LOG.error("Streaming query {} terminated with an exception.", query.id(), e);
+ // Make sure sibling queries do not keep running once one of them has failed.
+ stop();
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Best-effort, idempotent stop of a single query, see the class javadoc for what "best-effort"
+ * means here.
+ */
+ private void stopQuery(StreamingQuery query) {
+ try {
+ if (query.isActive()) {
+ query.stop();
+ }
+ } catch (TimeoutException | RuntimeException e) {
+ LOG.warn(
+ "Error while stopping streaming query {}: {}",
+ query.id(),
+ String.valueOf(e.getMessage()));
+ }
+ }
+
+ private void stopQueryById(UUID id) {
+ StreamingQuery match = null;
+ synchronized (lock) {
+ for (StreamingQuery query : queries) {
+ if (query.id().equals(id)) {
+ match = query;
+ break;
+ }
+ }
+ }
+ if (match != null) {
+ stopQuery(match);
+ }
+ }
+
+ private static String checkpointBaseDir(SparkCommonPipelineOptions options) {
+ String dir = options.getCheckpointDir();
+ if (dir == null || dir.isEmpty()) {
+ try {
+ dir = Files.createTempDirectory("beam-spark4-streaming-checkpoint").toString();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ LOG.warn("No checkpoint directory configured, falling back to temporary directory {}.", dir);
+ }
+ return dir;
+ }
+
+ /**
+ * Counts, per query, the number of consecutive micro-batches with zero input rows, and gracefully
+ * stops a query once its count reaches {@code threshold}. The count for a query resets to zero as
+ * soon as one of its micro-batches has rows.
+ *
+ *
Stopping happens on a dedicated thread rather than inline in {@link #onQueryProgress}:
+ * {@link StreamingQuery#stop()} blocks until the query's execution thread has shut down, which
+ * should not happen on the listener bus thread that dispatches these callbacks.
+ */
+ private final class IdleStopListener extends StreamingQueryListener {
+ private final int threshold;
+ private final Map idleCounts = new ConcurrentHashMap<>();
+
+ IdleStopListener(int threshold) {
+ this.threshold = threshold;
+ }
+
+ @Override
+ public void onQueryStarted(QueryStartedEvent event) {}
+
+ @Override
+ public void onQueryProgress(QueryProgressEvent event) {
+ UUID id = event.progress().id();
+ if (event.progress().numInputRows() == 0) {
+ int count = idleCounts.computeIfAbsent(id, unused -> new AtomicInteger()).incrementAndGet();
+ if (count >= threshold) {
+ idleCounts.remove(id);
+ Thread stopThread = new Thread(() -> stopQueryById(id), "beam-idle-stop-" + id);
+ stopThread.setDaemon(true);
+ stopThread.start();
+ }
+ } else {
+ idleCounts.remove(id);
+ }
+ }
+
+ @Override
+ public void onQueryTerminated(QueryTerminatedEvent event) {
+ idleCounts.remove(event.id());
+ }
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/GroupByKeyStreamingTranslator.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/GroupByKeyStreamingTranslator.java
new file mode 100644
index 000000000000..d166c663321d
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/GroupByKeyStreamingTranslator.java
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.translation.streaming;
+
+import java.util.Collections;
+import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator;
+import org.apache.beam.runners.spark.structuredstreaming.translation.streaming.state.BeamStatefulProcessorConfig;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.IterableCoder;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.WindowedValue;
+import org.apache.beam.sdk.values.WindowedValues;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoder;
+import org.apache.spark.sql.Encoders;
+
+/**
+ * Streaming translator for {@link GroupByKey}, implemented as a group-also-by-window hosted by the
+ * generic {@code transformWithState} operator.
+ *
+ *
Unlike the batch translator, which may ignore triggering and simply collect every value of a
+ * key, a streaming {@code GroupByKey} has to respect the watermark: a window's single on-time pane
+ * is emitted only once the watermark has passed the end of that window, and values arriving after
+ * that are dropped as late. Both are the responsibility of the Beam {@code ReduceFnRunner} that
+ * {@code BeamStatefulProcessorConfig.Mode#GROUP_ALSO_BY_WINDOW} sets up inside the operator, so all
+ * this translator does is put the data into and take it back out of the operator's byte[] row
+ * layout, see {@link TwsTransformFactory}.
+ *
+ *
{@code Combine.PerKey} is deliberately not registered for streaming, so combines reach this
+ * translator already expanded into {@code GroupByKey} plus a plain {@code ParDo}.
+ */
+public class GroupByKeyStreamingTranslator
+ extends TransformTranslator<
+ PCollection>, PCollection>>, GroupByKey> {
+
+ /** Output tag of the group-also-by-window {@code DoFn}, index 0, its only output. */
+ private static final String MAIN_OUTPUT_TAG_ID = "gbk-main-output";
+
+ public GroupByKeyStreamingTranslator() {
+ super(0.2f);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected void translate(GroupByKey transform, Context cxt) {
+ PCollection> input = cxt.getInput();
+ String stepName = cxt.getCurrentTransform().getFullName();
+
+ WindowingStrategy, ?> windowing = input.getWindowingStrategy();
+ StreamingTranslationHelpers.checkSupportedWindowing(windowing, stepName);
+
+ if (!(input.getCoder() instanceof KvCoder)) {
+ throw StreamingTranslationHelpers.unsupported(
+ stepName, "the non KV input coder " + input.getCoder());
+ }
+ KvCoder inputCoder = (KvCoder) input.getCoder();
+ Coder keyCoder = inputCoder.getKeyCoder();
+ Coder valueCoder = inputCoder.getValueCoder();
+ StreamingTranslationHelpers.checkDeterministicKeyCoder(keyCoder, stepName);
+
+ Coder extends BoundedWindow> windowCoder = windowing.getWindowFn().windowCoder();
+ KvCoder> outputCoder = KvCoder.of(keyCoder, IterableCoder.of(valueCoder));
+ TupleTag>> mainOutputTag = new TupleTag<>(MAIN_OUTPUT_TAG_ID);
+
+ Dataset keyedRows =
+ cxt.getDataset(input)
+ .map(
+ new StreamingTranslationHelpers.EncodeKeyedRow<>(
+ keyCoder, WindowedValues.getFullCoder(valueCoder, windowCoder)),
+ Encoders.BINARY());
+
+ BeamStatefulProcessorConfig config =
+ BeamStatefulProcessorConfig.builder()
+ .setMode(BeamStatefulProcessorConfig.Mode.GROUP_ALSO_BY_WINDOW)
+ .setKeyCoder(keyCoder)
+ .setValueCoder(valueCoder)
+ .setWindowingStrategy(windowing)
+ .setMainOutputTag(mainOutputTag)
+ .setOutputCoders(
+ Collections., Coder>>singletonMap(mainOutputTag, outputCoder))
+ .setOptionsSupplier(
+ StreamingTranslationHelpers.optionsSupplier(cxt.getOptionsSupplier()))
+ .setStepName(stepName)
+ .build();
+
+ // GROUP_ALSO_BY_WINDOW has a single output tag, so every row carries index 0 and no filtering
+ // by tag is needed on the way out.
+ Dataset outputRows = TwsTransformFactory.transform(keyedRows, config);
+
+ Encoder>>> encoder = cxt.windowedEncoder(outputCoder);
+ Dataset>>> result =
+ outputRows.map(
+ new StreamingTranslationHelpers.DecodeTaggedOutput<>(
+ WindowedValues.getFullCoder(outputCoder, windowCoder)),
+ encoder);
+
+ cxt.putDataset(cxt.getOutput(), result);
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/ReadUnboundedTranslator.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/ReadUnboundedTranslator.java
new file mode 100644
index 000000000000..36cf305e3376
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/ReadUnboundedTranslator.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.translation.streaming;
+
+import static org.apache.spark.sql.functions.col;
+
+import org.apache.beam.runners.spark.structuredstreaming.SparkStructuredStreamingPipelineOptions;
+import org.apache.beam.runners.spark.structuredstreaming.io.streaming.UnboundedSourceDataset;
+import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator;
+import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.io.UnboundedSource;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.util.construction.SplittableParDo;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.WindowedValue;
+import org.apache.beam.sdk.values.WindowedValues;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoder;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.Row;
+
+/**
+ * Translator for {@link SplittableParDo.PrimitiveUnboundedRead}, the streaming counterpart of the
+ * batch {@code ReadSourceTranslatorBatch}.
+ *
+ *
The heavy lifting is done by {@link UnboundedSourceDataset}, which wraps the Beam {@link
+ * UnboundedSource} in a DataSourceV2 micro-batch stream and returns a two column {@code
+ * Dataset}: the element encoded as a {@code WindowedValue}, plus its event timestamp. All this
+ * translator adds is the typed decode back into the {@code Dataset>} shape every
+ * other translator consumes.
+ *
+ *
Two things about the result are worth spelling out.
+ *
+ *
+ *
The watermark is already declared by {@link UnboundedSourceDataset} and must never
+ * be re-declared, Spark rejects a second {@code withWatermark} in the same plan. The {@code
+ * EventTimeWatermark} plan node sits below the projection and the typed map applied here and
+ * survives both, and a {@code transformWithState} operator downstream reads the query wide
+ * watermark rather than a column, so dropping the timestamp column here is safe.
+ *
Elements arrive in the global window, timestamped with the reader's record
+ * timestamp. A windowed pipeline therefore still needs its {@code Window.Assign}, which is
+ * translated by the reused batch translator.
+ *
+ */
+public class ReadUnboundedTranslator
+ extends TransformTranslator, SplittableParDo.PrimitiveUnboundedRead> {
+
+ public ReadUnboundedTranslator() {
+ super(0.05f);
+ }
+
+ @Override
+ protected void translate(SplittableParDo.PrimitiveUnboundedRead transform, Context cxt) {
+ PCollection output = cxt.getOutput();
+ UnboundedSource source = transform.getSource();
+ Coder elementCoder = output.getCoder();
+
+ // Matches what the partition readers emit: a value in the global window, timestamped with the
+ // record's own event timestamp.
+ WindowedValues.FullWindowedValueCoder payloadCoder =
+ WindowedValues.getFullCoder(elementCoder, GlobalWindow.Coder.INSTANCE);
+
+ SparkStructuredStreamingPipelineOptions options =
+ cxt.getOptions().as(SparkStructuredStreamingPipelineOptions.class);
+
+ Dataset rows =
+ UnboundedSourceDataset.of(cxt.getSparkSession(), source, payloadCoder, options);
+
+ Encoder> encoder =
+ cxt.windowedEncoder(elementCoder, GlobalWindow.Coder.INSTANCE);
+
+ Dataset> dataset =
+ rows.select(col(UnboundedSourceDataset.COL_PAYLOAD))
+ .as(Encoders.BINARY())
+ .map(new DecodePayload<>(payloadCoder), encoder);
+
+ cxt.putDataset(output, dataset);
+ }
+
+ /** Decodes the binary payload column back into a Beam {@code WindowedValue}. */
+ private static final class DecodePayload implements MapFunction> {
+ private final Coder> coder;
+
+ DecodePayload(Coder> coder) {
+ this.coder = coder;
+ }
+
+ @Override
+ public WindowedValue call(byte[] payload) {
+ return CoderHelpers.fromByteArray(payload, coder);
+ }
+ }
+}
diff --git a/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java
new file mode 100644
index 000000000000..01f2dd63c557
--- /dev/null
+++ b/runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTranslator.java
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.translation.streaming;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator;
+import org.apache.beam.runners.spark.structuredstreaming.translation.streaming.state.BeamStatefulProcessorConfig;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignature;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignatures;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.util.construction.ParDoTranslation;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.WindowedValue;
+import org.apache.beam.sdk.values.WindowedValues;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoder;
+import org.apache.spark.sql.Encoders;
+
+/**
+ * Streaming translator for a stateful {@link ParDo.MultiOutput}, that is a {@code ParDo} whose
+ * {@code DoFn} declares {@code @StateId} state or {@code @TimerId} timers. Stateless {@code ParDo}s
+ * keep using the batch translator unchanged, the streaming registry only routes stateful ones here.
+ *
+ *
The user's {@code DoFn} is hosted verbatim by the generic {@code transformWithState} operator
+ * in {@code BeamStatefulProcessorConfig.Mode#STATEFUL_PARDO}, which runs it through {@code
+ * DoFnRunners.simpleRunner} wrapped in {@code DoFnRunners.defaultStatefulDoFnRunner}, so Beam's own
+ * state, timer and window garbage collection semantics apply. This translator's whole job is the
+ * byte[] row plumbing documented on {@link TwsTransformFactory}: encode {@code WindowedValue>} into keyed input rows, and split the tagged output rows back into one {@code
+ * Dataset>} per {@link TupleTag}.
+ *
+ *
Note on multiple outputs: each additional tag adds one {@code filter} plus one {@code map} on
+ * top of the same operator, which Spark plans as a separate branch. Only outputs that are actually
+ * consumed downstream, plus the main output, get a dataset at all.
+ */
+public class StatefulParDoStreamingTranslator
+ extends TransformTranslator<
+ PCollection extends KV>, PCollectionTuple, ParDo.MultiOutput, OutputT>> {
+
+ public StatefulParDoStreamingTranslator() {
+ super(0.2f);
+ }
+
+ @Override
+ protected boolean canTranslate(ParDo.MultiOutput, OutputT> transform) {
+ DoFn, OutputT> doFn = transform.getFn();
+ String stepName = doFn.getClass().getName();
+ DoFnSignature signature = DoFnSignatures.signatureForDoFn(doFn);
+
+ checkState(
+ !signature.processElement().isSplittable(),
+ "Not expected to directly translate splittable DoFn, should have been overridden: %s",
+ doFn);
+
+ StreamingTranslationHelpers.checkNoProcessingTimeTimers(doFn, signature, stepName);
+
+ if (signature.onWindowExpiration() != null) {
+ throw StreamingTranslationHelpers.unsupported(stepName, "@OnWindowExpiration");
+ }
+ if (signature.processElement().requiresTimeSortedInput()) {
+ throw StreamingTranslationHelpers.unsupported(stepName, "@RequiresTimeSortedInput");
+ }
+ if (!transform.getSideInputs().isEmpty()) {
+ throw StreamingTranslationHelpers.unsupported(
+ stepName,
+ "side inputs on a stateful ParDo. Broadcasting a side input requires collecting its "
+ + "PCollection, which is not possible while the pipeline is streaming");
+ }
+ return true;
+ }
+
+ @Override
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ protected void translate(ParDo.MultiOutput, OutputT> transform, Context cxt) {
+ PCollection> input = (PCollection>) cxt.getInput();
+ String stepName = cxt.getCurrentTransform().getFullName();
+
+ WindowingStrategy, ?> windowing = input.getWindowingStrategy();
+ StreamingTranslationHelpers.checkSupportedWindowing(windowing, stepName);
+
+ // Beam guarantees a stateful DoFn is applied to a keyed PCollection, but say so clearly rather
+ // than failing with a ClassCastException deep in the operator.
+ if (!(input.getCoder() instanceof KvCoder)) {
+ throw StreamingTranslationHelpers.unsupported(
+ stepName,
+ "state or timers on the non KV input coder "
+ + input.getCoder()
+ + ". A stateful ParDo must be applied to a PCollection of KVs");
+ }
+ KvCoder inputCoder = (KvCoder) input.getCoder();
+ Coder keyCoder = inputCoder.getKeyCoder();
+ Coder valueCoder = inputCoder.getValueCoder();
+ StreamingTranslationHelpers.checkDeterministicKeyCoder(keyCoder, stepName);
+
+ Coder extends BoundedWindow> windowCoder = windowing.getWindowFn().windowCoder();
+
+ TupleTag mainOutputTag = transform.getMainOutputTag();
+ List> additionalOutputTags =
+ new ArrayList<>(transform.getAdditionalOutputTags().getAll());
+
+ // One coder per tag the DoFn may emit to, taken from the PCollection behind that tag.
+ Map, Coder>> outputCoders = new LinkedHashMap<>();
+ outputCoders.put(mainOutputTag, cxt.getOutput(mainOutputTag).getCoder());
+ for (TupleTag> tag : additionalOutputTags) {
+ outputCoders.put(tag, cxt.getOutput((TupleTag) tag).getCoder());
+ }
+
+ Dataset keyedRows =
+ cxt.getDataset(input)
+ .map(
+ new StreamingTranslationHelpers.EncodeKeyedRow<>(
+ keyCoder, WindowedValues.getFullCoder(valueCoder, windowCoder)),
+ Encoders.BINARY());
+
+ BeamStatefulProcessorConfig config =
+ BeamStatefulProcessorConfig.builder()
+ .setMode(BeamStatefulProcessorConfig.Mode.STATEFUL_PARDO)
+ .setDoFn(transform.getFn())
+ .setKeyCoder(keyCoder)
+ .setValueCoder(valueCoder)
+ .setWindowingStrategy(windowing)
+ .setMainOutputTag(mainOutputTag)
+ .setAdditionalOutputTags(additionalOutputTags)
+ .setOutputCoders(outputCoders)
+ .setDoFnSchemaInformation(
+ ParDoTranslation.getSchemaInformation(cxt.getCurrentTransform()))
+ .setOptionsSupplier(
+ StreamingTranslationHelpers.optionsSupplier(cxt.getOptionsSupplier()))
+ .setStepName(stepName)
+ .build();
+
+ Dataset outputRows = TwsTransformFactory.transform(keyedRows, config);
+
+ List> allTags = config.outputTags();
+ boolean singleTag = allTags.size() == 1;
+ for (int tagIndex = 0; tagIndex < allTags.size(); tagIndex++) {
+ TupleTag