You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "Eliaaazzz (via GitHub)" <gi...@apache.org> on 2026/09/15 12:38:24 UTC
[PR] [Spark][#36841] Add an opt-in Dataset-based backend for portable pipelines [beam]
Eliaaazzz opened a new pull request, #40129:
URL: https://github.com/apache/beam/pull/40129
Proof of concept for the Dataset-based portable Spark 4 backend discussed in the "[DISCUSS] Spark 4 portable runner direction" thread on dev@. Opening this as a draft so the discussion has concrete code to work from. Addresses #36841.
### What this adds
An opt-in `--useStructuredStreaming` flag on the portable Spark runner. With the flag set, `SparkPipelineRunner` translates the fused Runner API pipeline into Spark Datasets through a new `SparkDatasetPortablePipelineTranslator`:
- `Impulse`, `Flatten`, `Reshuffle` and `GroupByKey` become Dataset operations. `GroupByKey` groups on the encoded key and runs `SparkGroupAlsoByWindowViaOutputBufferFn` per key. That is the general `GroupAlsoByWindow` the RDD batch translator falls back to when a windowing strategy is not eligible for the `GroupNonMergingWindowsFunctions` fast paths. The Dataset path always takes the general branch, so it gives up the non-merging memory optimization and keeps the `ReduceFnRunner` semantics.
- Executable stages run inside `Dataset.mapPartitions` through the existing Fn API bridge, `SparkExecutableStageFunction`, which is unchanged. Outputs are demultiplexed with `EncoderHelpers.oneOfEncoder`, following `ParDoTranslatorBatch`. Side inputs are collected during translation and broadcast.
- Fully bounded pipelines run as batch Datasets. Unbounded input, user state and timers are rejected at translation with a message pointing at the tracking issue, so nothing silently runs on a path that cannot support it yet.
The translation context attaches to the `SparkContext` that `SparkContextFactory` already created, so the job server code path, the metrics accumulator and result handling are unchanged. The translator lives in the shared `runners/spark` base and compiles for both the Spark 3 and Spark 4 modules.
The flag name matches the terminology in the dev@ thread and is provisional. Today it only selects the Dataset translator. No Structured Streaming query is built yet.
### The exit gate
Following Yi's suggestion on the thread, `validatesPortableRunnerStructuredStreaming` runs the streaming `PortableValidatesRunner` suite with the flag on, keeping exactly the same categories and test filters as `validatesPortableRunnerStreaming`, and a `PostCommit Java PVR Spark4 StructuredStreaming` workflow runs it. That suite is the exit gate for flipping the default for portable streaming pipelines.
What that gate currently measures is narrower than its name suggests. `--streaming` does not by itself mark any PCollection unbounded, and the streaming exclusion list already removes the categories that would produce one. `ValidatesRunner` extends `NeedsRunner`, so tests annotated with `NeedsRunner` and `UsesUnboundedPCollections` are outside the suite, and the remaining `ValidatesRunner` tests carrying `UsesUnboundedPCollections` are each excluded by `UsesTimersInParDo`, `UsesTestStream`, `UsesStrictTimerOrdering` or `UsesOnWindowExpiration`. The first run therefore exercises bounded Dataset execution under the streaming exclusion list. The risk it actually probes is windowing, triggers and GroupByKey behaviour on this backend. This matches the point on the thread that the suite is mostly batch pipelines after the exclusions.
The new task is deliberately not part of the aggregate `validatesPortableRunner` task while the backend is a preview.
### Out of scope, tracked separately
- Unbounded splittable DoFn residuals across micro-batches, #19468.
- Stateful stages, #20396.
- Timers, #20397.
- The Structured Streaming query itself for pipelines with unbounded input, #36841.
This does not modify the in-process Structured Streaming runner, including the work merged in #39906 and #39939 and the open #39576. It does reuse two of its classes, `EncoderHelpers` and `EvaluationContext`, so the portable path also picks up the Spark 4 `EncoderHelpers` override that came with the Spark 4 runner in #38255.
### Validation
- `SparkDatasetPortableExecutionTest` runs Impulse, ParDo, Flatten, GroupByKey and PAssert end to end on the Dataset backend with the embedded SDK harness, and asserts that unbounded input and a stateful ParDo are rejected at translation with the expected message. Three tests, green on `:runners:spark:4:test` under JDK 17 and on `:runners:spark:3:test` under JDK 11.
- Known coverage gaps in this PR, all of which the PVR run is meant to cover: multi-output stages, so the `oneOfEncoder` demux is only ever built with one alternative; `Reshuffle`; the `Flatten` re-encode branch for mismatched coders; side inputs, which the streaming suite also excludes via `UsesSideInputs`; and the timer half of the state and timer rejection.
- Only the global window is covered by a test here. Sessions, sliding windows and non-default timestamp combiners are unverified on this backend.
- I have not run the full PVR suite locally, so the first run of the new workflow is the real signal.
- The Spark 3 compile runs with the nullness checker enabled. The two new files suppress `nullness` at class level, matching the surrounding translator code.
### Notes for review
Three things I would particularly like a second opinion on:
- Setting `pipelineOptions.setStreaming(false)` in the new branch. `MetricsAccumulator.init` reads the option directly, not the mode the runner resolved, so without this it takes its checkpoint branch and seeds the accumulator from a checkpoint file this path never writes back.
- Persisted Datasets are never unpersisted and the `SparkSession` is never cleared. That is fine when `PortableBatchMode.stop()` stops the context, and it accumulates when the context is provided or reused.
- Side inputs are collected eagerly during translation, which makes translation itself trigger Spark actions.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [x] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes. Holding this until the flag is more than a preview.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Spark][#36841] Add an opt-in Dataset-based backend for portable pipelines [beam]
Posted by "Eliaaazzz (via GitHub)" <gi...@apache.org>.
Eliaaazzz commented on code in PR #40129:
URL: https://github.com/apache/beam/pull/40129#discussion_r4052155697
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java:
##########
@@ -82,7 +83,14 @@ public SparkPipelineRunner(SparkPipelineOptions pipelineOptions) {
public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo) {
SparkPortablePipelineTranslator translator;
boolean isStreaming = pipelineOptions.isStreaming() || hasUnboundedPCollections(pipeline);
- if (isStreaming) {
+ if (pipelineOptions.getUseStructuredStreaming()) {
+ // The Dataset backend evaluates its own leaves. It never starts a DStream context, and it
Review Comment:
Done in f4a2d39cfcf. The runner no longer sets the streaming option. It selects the Dataset translator by useStructuredStreaming, and with that flag set it never selects the DStream translator and never starts a streaming context, whatever the streaming option says.
For the metrics accumulator I added an init overload with an explicit useCheckpoint argument. The portable runner passes the streaming option restricted to the DStream path, so the Dataset backend never opens the DStream metrics checkpoint and the DStream path keeps its previous condition. The two argument init keeps reading the streaming option, so the classic runner is unchanged.
I also reworded the runner comment, the option description, the translator javadoc, the rejection message and the README so the unbounded rejection reads as the current status.
##########
runners/spark/src/test/java/org/apache/beam/runners/spark/SparkDatasetPortableExecutionTest.java:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.hasItem;
+import static org.junit.Assert.assertEquals;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.beam.model.jobmanagement.v1.JobApi.JobState;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.runners.jobsubmission.JobInvocation;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.BigEndianLongCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.options.PortablePipelineOptions;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.CrashingRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.Flatten;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.Impulse;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.WithKeys;
+import org.apache.beam.sdk.util.construction.Environments;
+import org.apache.beam.sdk.util.construction.PipelineTranslation;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionList;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.ListeningExecutorService;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.MoreExecutors;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Runs portable pipelines end to end on the Dataset-based backend: job invocation, executable stage
+ * translation, and execution with the embedded SDK harness.
+ */
+@RunWith(JUnit4.class)
+public class SparkDatasetPortableExecutionTest implements Serializable {
+ private static ListeningExecutorService executor;
+
+ @BeforeClass
+ public static void setUp() {
+ executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(1));
+ }
+
+ @AfterClass
+ public static void tearDown() throws InterruptedException {
+ executor.shutdown();
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+ executor = null;
+ }
+
+ @Test(timeout = 180_000)
+ public void boundedPipelineRunsOnDatasets() throws Exception {
+ SparkPipelineOptions options = options();
+ Pipeline p = Pipeline.create(options);
+ PCollection<String> words =
+ p.apply("impulse", Impulse.create())
Review Comment:
Done in f4a2d39cfcf. SparkDatasetPortableExecutionTest keeps one end to end run through the job invoker with --streaming and --useStructuredStreaming, as the gate task sets them. The rest moved to component tests in the translation package. The end to end rejection checks went with them, so a translation failure surfacing as a FAILED job is no longer covered here.
SparkDatasetPortablePipelineTranslatorTest covers one piece at a time. Flatten with mismatched input coders, GroupByKey with fixed windows and Reshuffle are translated on injected Datasets. Impulse, a multi-output stage and a side input go through the translator on a small pipeline and run in the embedded harness. The unbounded, state and timer rejections are asserted on the translator directly. SparkDatasetTranslationContextTest covers the persist decisions and that only unconsumed Datasets are evaluated. The per transform translators are package private with @VisibleForTesting for that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Spark][#36841] Add an opt-in Dataset-based backend for portable pipelines [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #40129:
URL: https://github.com/apache/beam/pull/40129#issuecomment-5691248344
Assigning reviewers:
R: @Abacn for label build.
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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Spark][#36841] Add an opt-in Dataset-based backend for portable pipelines [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn merged PR #40129:
URL: https://github.com/apache/beam/pull/40129
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Spark][#36841] Add an opt-in Dataset-based backend for portable pipelines [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #40129:
URL: https://github.com/apache/beam/pull/40129#discussion_r4049956534
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java:
##########
@@ -82,7 +83,14 @@ public SparkPipelineRunner(SparkPipelineOptions pipelineOptions) {
public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo) {
SparkPortablePipelineTranslator translator;
boolean isStreaming = pipelineOptions.isStreaming() || hasUnboundedPCollections(pipeline);
- if (isStreaming) {
+ if (pipelineOptions.getUseStructuredStreaming()) {
+ // The Dataset backend evaluates its own leaves. It never starts a DStream context, and it
Review Comment:
We shouldn't overwrite `pipelineOptions.setStreaming` here as it doesn't sounds semantically correct. We should prevent the logic ends up detecting "isStreaming" to avoid expansion involving DStream. The correct flag is "useStructuredStreaming"
"The Dataset backend rejects unbounded input at translation" is current supporting status and isn't a spec
##########
runners/spark/src/test/java/org/apache/beam/runners/spark/SparkDatasetPortableExecutionTest.java:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.hasItem;
+import static org.junit.Assert.assertEquals;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.beam.model.jobmanagement.v1.JobApi.JobState;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.runners.jobsubmission.JobInvocation;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.BigEndianLongCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.options.PortablePipelineOptions;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.CrashingRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.Flatten;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.Impulse;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.WithKeys;
+import org.apache.beam.sdk.util.construction.Environments;
+import org.apache.beam.sdk.util.construction.PipelineTranslation;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionList;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.ListeningExecutorService;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.MoreExecutors;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Runs portable pipelines end to end on the Dataset-based backend: job invocation, executable stage
+ * translation, and execution with the embedded SDK harness.
+ */
+@RunWith(JUnit4.class)
+public class SparkDatasetPortableExecutionTest implements Serializable {
+ private static ListeningExecutorService executor;
+
+ @BeforeClass
+ public static void setUp() {
+ executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(1));
+ }
+
+ @AfterClass
+ public static void tearDown() throws InterruptedException {
+ executor.shutdown();
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+ executor = null;
+ }
+
+ @Test(timeout = 180_000)
+ public void boundedPipelineRunsOnDatasets() throws Exception {
+ SparkPipelineOptions options = options();
+ Pipeline p = Pipeline.create(options);
+ PCollection<String> words =
+ p.apply("impulse", Impulse.create())
Review Comment:
We can have a few full pipeline test but the majority of runner unit tests are targeted, testing component but not a pipeline. The latter is covered by validate runner tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Spark][#36841] Add an opt-in Dataset-based backend for portable pipelines [beam]
Posted by "Eliaaazzz (via GitHub)" <gi...@apache.org>.
Eliaaazzz commented on code in PR #40129:
URL: https://github.com/apache/beam/pull/40129#discussion_r4052155697
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java:
##########
@@ -82,7 +83,14 @@ public SparkPipelineRunner(SparkPipelineOptions pipelineOptions) {
public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo) {
SparkPortablePipelineTranslator translator;
boolean isStreaming = pipelineOptions.isStreaming() || hasUnboundedPCollections(pipeline);
- if (isStreaming) {
+ if (pipelineOptions.getUseStructuredStreaming()) {
+ // The Dataset backend evaluates its own leaves. It never starts a DStream context, and it
Review Comment:
Fixed in f4a2d39cfcf. The runner no longer touches the streaming option. With useStructuredStreaming set it picks the Dataset translator and skips the DStream translator and streaming context. MetricsAccumulator.init got an explicit useCheckpoint argument, so the Dataset path never opens the DStream metrics checkpoint and the DStream path keeps its old condition. I also reworded the comments, option description and README to describe the unbounded rejection as current status.
##########
runners/spark/src/test/java/org/apache/beam/runners/spark/SparkDatasetPortableExecutionTest.java:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.hasItem;
+import static org.junit.Assert.assertEquals;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.beam.model.jobmanagement.v1.JobApi.JobState;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.runners.jobsubmission.JobInvocation;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.BigEndianLongCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.options.PortablePipelineOptions;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.CrashingRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.Flatten;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.Impulse;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.WithKeys;
+import org.apache.beam.sdk.util.construction.Environments;
+import org.apache.beam.sdk.util.construction.PipelineTranslation;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionList;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.ListeningExecutorService;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.MoreExecutors;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Runs portable pipelines end to end on the Dataset-based backend: job invocation, executable stage
+ * translation, and execution with the embedded SDK harness.
+ */
+@RunWith(JUnit4.class)
+public class SparkDatasetPortableExecutionTest implements Serializable {
+ private static ListeningExecutorService executor;
+
+ @BeforeClass
+ public static void setUp() {
+ executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(1));
+ }
+
+ @AfterClass
+ public static void tearDown() throws InterruptedException {
+ executor.shutdown();
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+ executor = null;
+ }
+
+ @Test(timeout = 180_000)
+ public void boundedPipelineRunsOnDatasets() throws Exception {
+ SparkPipelineOptions options = options();
+ Pipeline p = Pipeline.create(options);
+ PCollection<String> words =
+ p.apply("impulse", Impulse.create())
Review Comment:
Fixed in f4a2d39cfcf. I kept one end to end test through the job invoker, with --streaming and --useStructuredStreaming like the gate task, and moved the rest into component tests. SparkDatasetPortablePipelineTranslatorTest translates single transforms on injected Datasets and runs stages in the embedded harness, SparkDatasetTranslationContextTest checks the persist and evaluation logic, and the rejection cases are asserted on the translator directly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org