You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "Amar3tto (via GitHub)" <gi...@apache.org> on 2025/03/11 19:00:56 UTC
[PR] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Amar3tto opened a new pull request, #34253:
URL: https://github.com/apache/beam/pull/34253
Fixes #34207
Successful run:
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] 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.
- [ ] 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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #34253:
URL: https://github.com/apache/beam/pull/34253#discussion_r1991949228
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java:
##########
@@ -322,13 +342,21 @@ private static <T> void translateFlatten(
dStreams.add(unboundedDataset.getDStream());
} else {
// create a single RDD stream.
- 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());
-
+ JavaDStream<WindowedValue<T>> dStream;
+ if (context.getSparkContext().version().startsWith("3")) {
+ 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
Review Comment:
Thanks, let me merge this and reopen #20426
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto commented on code in PR #34253:
URL: https://github.com/apache/beam/pull/34253#discussion_r1991780035
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java:
##########
@@ -322,13 +342,21 @@ private static <T> void translateFlatten(
dStreams.add(unboundedDataset.getDStream());
} else {
// create a single RDD stream.
- 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());
-
+ JavaDStream<WindowedValue<T>> dStream;
+ if (context.getSparkContext().version().startsWith("3")) {
+ 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
Review Comment:
I just saw that Spark 3 actually is a primary version. So I will just rollback changes related to the PortableRunner (without checking Spark version).
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn merged PR #34253:
URL: https://github.com/apache/beam/pull/34253
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on code in PR #34253:
URL: https://github.com/apache/beam/pull/34253#discussion_r1991760239
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java:
##########
@@ -322,13 +342,21 @@ private static <T> void translateFlatten(
dStreams.add(unboundedDataset.getDStream());
} else {
// create a single RDD stream.
- 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());
-
+ JavaDStream<WindowedValue<T>> dStream;
+ if (context.getSparkContext().version().startsWith("3")) {
+ 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
Review Comment:
I should have been more thorough when working on the PR, and I sincerely apologize for not doing so. I would also like to express my gratitude to @Amar3tto for their support and assistance.
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on code in PR #34253:
URL: https://github.com/apache/beam/pull/34253#discussion_r1991756755
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java:
##########
@@ -322,13 +342,21 @@ private static <T> void translateFlatten(
dStreams.add(unboundedDataset.getDStream());
} else {
// create a single RDD stream.
- 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());
-
+ JavaDStream<WindowedValue<T>> dStream;
+ if (context.getSparkContext().version().startsWith("3")) {
+ 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
Review Comment:
Hello. I am currently reviewing the issue.
At the time of submitting the PR, the following trigger files were used, but it seems that `PortableRunner` was not included, which is likely why the issue was not detected:
- `beam_PostCommit_Java_ValidatesRunner_Spark.json`
- `beam_PostCommit_Java_ValidatesRunner_Spark_Java11.json`
- `beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming.json`
I agree with @Amar3tto's suggestion. For now, it seems we need to roll back the changes related to the `PortableRunner` and revisit the task afterward.
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto commented on PR #34253:
URL: https://github.com/apache/beam/pull/34253#issuecomment-2716438609
Run Java PreCommit
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto commented on PR #34253:
URL: https://github.com/apache/beam/pull/34253#issuecomment-2715483468
Run Java PreCommit
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #34253:
URL: https://github.com/apache/beam/pull/34253#discussion_r1991697412
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java:
##########
@@ -322,13 +342,21 @@ private static <T> void translateFlatten(
dStreams.add(unboundedDataset.getDStream());
} else {
// create a single RDD stream.
- 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());
-
+ JavaDStream<WindowedValue<T>> dStream;
+ if (context.getSparkContext().version().startsWith("3")) {
+ 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
Review Comment:
This adds back a previously closed issue #20426 that #34080 intended to solve. To what extent #34080 still functioning, or shall we revert it as a whole?
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #34253:
URL: https://github.com/apache/beam/pull/34253#issuecomment-2717475205
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto commented on PR #34253:
URL: https://github.com/apache/beam/pull/34253#issuecomment-2717472172
R: @Abacn
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on code in PR #34253:
URL: https://github.com/apache/beam/pull/34253#discussion_r1991785088
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java:
##########
@@ -322,13 +342,21 @@ private static <T> void translateFlatten(
dStreams.add(unboundedDataset.getDStream());
} else {
// create a single RDD stream.
- 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());
-
+ JavaDStream<WindowedValue<T>> dStream;
+ if (context.getSparkContext().version().startsWith("3")) {
+ 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
Review Comment:
Thank you.
I will make it a priority to revisit and proceed with the PortableRunner-related tasks associated with #20426 as soon as possible.
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #34253:
URL: https://github.com/apache/beam/pull/34253#issuecomment-2718173084
CC: @twosom
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto commented on code in PR #34253:
URL: https://github.com/apache/beam/pull/34253#discussion_r1991746547
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java:
##########
@@ -322,13 +342,21 @@ private static <T> void translateFlatten(
dStreams.add(unboundedDataset.getDStream());
} else {
// create a single RDD stream.
- 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());
-
+ JavaDStream<WindowedValue<T>> dStream;
+ if (context.getSparkContext().version().startsWith("3")) {
+ 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
Review Comment:
There might be a better fix for Spark 3, that also solves https://github.com/apache/beam/issues/20426 issue.
I didn't dig too much, just wanted to fix permared test.
@twosom What do you think?
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto commented on code in PR #34253:
URL: https://github.com/apache/beam/pull/34253#discussion_r1991738026
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java:
##########
@@ -322,13 +342,21 @@ private static <T> void translateFlatten(
dStreams.add(unboundedDataset.getDStream());
} else {
// create a single RDD stream.
- 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());
-
+ JavaDStream<WindowedValue<T>> dStream;
+ if (context.getSparkContext().version().startsWith("3")) {
+ 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
Review Comment:
#34080 should still work for Spark 2. I guess that PR was made and tested only for Spark 2.
Spark 3 test was failing, so my PR should fix it.
--
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] Fix PostCommit Java PVR Spark3 Streaming job [beam]
Posted by "Amar3tto (via GitHub)" <gi...@apache.org>.
Amar3tto commented on code in PR #34253:
URL: https://github.com/apache/beam/pull/34253#discussion_r1991738026
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkStreamingPortablePipelineTranslator.java:
##########
@@ -322,13 +342,21 @@ private static <T> void translateFlatten(
dStreams.add(unboundedDataset.getDStream());
} else {
// create a single RDD stream.
- 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());
-
+ JavaDStream<WindowedValue<T>> dStream;
+ if (context.getSparkContext().version().startsWith("3")) {
+ 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
Review Comment:
#34080 should still work for Spark 2 / Non-portable runner. I guess that PR was made and tested only for Spark 2.
Spark 3 test was failing, so my PR should fix it.
--
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