(beam) branch master updated: Spark Runner : Fix invalid translateImpulse (#34272)

[yh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

yhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new bdb26939350 Spark Runner : Fix invalid translateImpulse (#34272)
bdb26939350 is described below

commit bdb26939350a1a56618bac9596eb072411c556b9
Author: twosom <72...@users.noreply.github.com>
AuthorDate: Fri Mar 14 02:26:24 2025 +0900

    Spark Runner : Fix invalid translateImpulse (#34272)
    
    * fix : fix invalid flatten translate
    
    * trigger PostCommit Java PVR Spark3 Streaming test
---
 .../beam_PostCommit_Java_PVR_Spark3_Streaming.json |  2 +-
 .../SparkStreamingPortablePipelineTranslator.java  | 38 ++++++++++++----------
 2 files changed, 21 insertions(+), 19 deletions(-)

diff --git a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark3_Streaming.json b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark3_Streaming.json
index c537844dc84..e0266d62f2e 100644
--- a/.github/trigger_files/beam_PostCommit_Java_PVR_Spark3_Streaming.json
+++ b/.github/trigger_files/beam_PostCommit_Java_PVR_Spark3_Streaming.json
@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 3
+  "modification": 4
 }
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java
index 54c102ee2b3..a252cc044fc 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java
@@ -30,9 +30,7 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Queue;
 import java.util.Set;
-import java.util.concurrent.LinkedBlockingQueue;
 import org.apache.beam.model.pipeline.v1.RunnerApi;
 import org.apache.beam.runners.fnexecution.provisioning.JobInfo;
 import org.apache.beam.runners.spark.SparkPipelineOptions;
@@ -61,10 +59,10 @@ import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.BiMap;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.JavaSparkContext$;
 import org.apache.spark.broadcast.Broadcast;
 import org.apache.spark.storage.StorageLevel;
 import org.apache.spark.streaming.api.java.JavaDStream;
-import org.apache.spark.streaming.api.java.JavaInputDStream;
 import scala.Tuple2;
 import scala.collection.JavaConverters;
 
@@ -158,14 +156,14 @@ public class SparkStreamingPortablePipelineTranslator
             .parallelize(CoderHelpers.toByteArrays(windowedValues, windowCoder))
             .map(CoderHelpers.fromByteFunction(windowCoder));
 
-    Queue<JavaRDD<WindowedValue<byte[]>>> rddQueue = new LinkedBlockingQueue<>();
-    rddQueue.offer(emptyByteArrayRDD);
-    JavaInputDStream<WindowedValue<byte[]>> emptyByteArrayStream =
-        context.getStreamingContext().queueStream(rddQueue, true /* oneAtATime */);
+    final SingleEmitInputDStream<WindowedValue<byte[]>> inputDStream =
+        new SingleEmitInputDStream<>(context.getStreamingContext().ssc(), emptyByteArrayRDD.rdd());
+
+    final JavaDStream<WindowedValue<byte[]>> stream =
+        JavaDStream.fromDStream(inputDStream, JavaSparkContext$.MODULE$.fakeClassTag());
+
     UnboundedDataset<byte[]> output =
-        new UnboundedDataset<>(
-            emptyByteArrayStream,
-            Collections.singletonList(emptyByteArrayStream.inputDStream().id()));
+        new UnboundedDataset<>(stream, Collections.singletonList(inputDStream.id()));
 
     // Add watermark to holder and advance to infinity to ensure future watermarks can be updated
     GlobalWatermarkHolder.SparkWatermarks sparkWatermark =
@@ -305,9 +303,11 @@ public class SparkStreamingPortablePipelineTranslator
     List<Integer> streamSources = new ArrayList<>();
 
     if (inputsMap.isEmpty()) {
-      Queue<JavaRDD<WindowedValue<T>>> q = new LinkedBlockingQueue<>();
-      q.offer(context.getSparkContext().emptyRDD());
-      unifiedStreams = context.getStreamingContext().queueStream(q);
+      final JavaRDD<WindowedValue<T>> emptyRDD = context.getSparkContext().emptyRDD();
+      final SingleEmitInputDStream<WindowedValue<T>> singleEmitInputDStream =
+          new SingleEmitInputDStream<>(context.getStreamingContext().ssc(), emptyRDD.rdd());
+      unifiedStreams =
+          JavaDStream.fromDStream(singleEmitInputDStream, JavaSparkContext$.MODULE$.fakeClassTag());
     } else {
       List<JavaDStream<WindowedValue<T>>> dStreams = new ArrayList<>();
       for (String inputId : inputsMap.values()) {
@@ -318,11 +318,13 @@ public class SparkStreamingPortablePipelineTranslator
           dStreams.add(unboundedDataset.getDStream());
         } else {
           // create a single RDD stream.
-          Queue<JavaRDD<WindowedValue<T>>> q = new LinkedBlockingQueue<>();
-          q.offer(((BoundedDataset) dataset).getRDD());
-          // TODO (https://github.com/apache/beam/issues/20426): this is not recoverable from
-          // checkpoint!
-          JavaDStream<WindowedValue<T>> dStream = context.getStreamingContext().queueStream(q);
+          final SingleEmitInputDStream<WindowedValue<T>> singleEmitInputDStream =
+              new SingleEmitInputDStream<WindowedValue<T>>(
+                  context.getStreamingContext().ssc(), ((BoundedDataset) dataset).getRDD().rdd());
+          final JavaDStream<WindowedValue<T>> dStream =
+              JavaDStream.fromDStream(
+                  singleEmitInputDStream, JavaSparkContext$.MODULE$.fakeClassTag());
+
           dStreams.add(dStream);
         }
       }