You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/01/25 18:50:50 UTC
[GitHub] [beam] nehsyc opened a new pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
nehsyc opened a new pull request #13805:
URL: https://github.com/apache/beam/pull/13805
Add a transform translator for `GroupIntoBatches` in Java to populate the URN and payload in the runner API.
Also updated `GroupIntoBatches` implementation to wrap batching parameters in a class so that can be reused by both variation of the transform (with or without sharded keys).
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [x] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] 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://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam
.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | --- | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
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)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #13805:
URL: https://github.com/apache/beam/pull/13805#discussion_r568869083
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,27 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
+ if (!options.isEnableStreamingEngine()) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for "
+ + "non-Streaming Engine jobs. Default sharding will be applied.");
return;
}
- if (!options.isEnableStreamingEngine()) {
+ if (!hasExperiment(options, "enable_streaming_auto_sharding")) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
+ "Runner determined sharding not enabled in Dataflow for GroupIntoBatches for Streaming "
+ + "Engine jobs: --enable_streaming_auto_sharding=false. Default sharding will be "
+ + "applied.");
return;
}
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
+ if (hasExperiment(options, "beam_fn_api") && !hasExperiment(options, "use_runner_v2")) {
+ LOG.warn(
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
+ + "jobs not using runner v2: --beam_fn_api=true, --use_runner_v2=false. Default "
+ + "sharding will be applied.");
+ return;
Review comment:
I would prefer rejecting the pipeline since it's also bad when customer expects autosharding but it's not.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz merged pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
boyuanzz merged pull request #13805:
URL: https://github.com/apache/beam/pull/13805
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #13805:
URL: https://github.com/apache/beam/pull/13805#discussion_r568093837
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,19 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
+ if (!options.isEnableStreamingEngine()) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for "
+ + "non-Streaming Engine jobs. Default sharding will be applied.");
return;
}
- if (!options.isEnableStreamingEngine()) {
+ if (!hasExperiment(options, "enable_streaming_auto_sharding")) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
+ "Runner determined sharding not enabled in Dataflow for GroupIntoBatches for Streaming "
+ + "Engine jobs. Default sharding will be applied.");
return;
}
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
- }
+ pcollectionsRequiringAutoSharding.add(pcol);
Review comment:
I think we also want to want to check the case `hasExperiment(options, "beam_fn_api") && !hasExperiment(options, "use_runner_v2")`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nehsyc commented on pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13805:
URL: https://github.com/apache/beam/pull/13805#issuecomment-767034330
R: @boyuanzz
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #13805:
URL: https://github.com/apache/beam/pull/13805#issuecomment-771160961
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nehsyc commented on pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13805:
URL: https://github.com/apache/beam/pull/13805#issuecomment-767034330
R: @boyuanzz
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nehsyc commented on a change in pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13805:
URL: https://github.com/apache/beam/pull/13805#discussion_r568325487
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,19 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
+ if (!options.isEnableStreamingEngine()) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for "
+ + "non-Streaming Engine jobs. Default sharding will be applied.");
return;
}
- if (!options.isEnableStreamingEngine()) {
+ if (!hasExperiment(options, "enable_streaming_auto_sharding")) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
+ "Runner determined sharding not enabled in Dataflow for GroupIntoBatches for Streaming "
+ + "Engine jobs. Default sharding will be applied.");
return;
}
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
- }
+ pcollectionsRequiringAutoSharding.add(pcol);
Review comment:
Good catch. Added the check.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nehsyc commented on a change in pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13805:
URL: https://github.com/apache/beam/pull/13805#discussion_r568325487
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,19 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
+ if (!options.isEnableStreamingEngine()) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for "
+ + "non-Streaming Engine jobs. Default sharding will be applied.");
return;
}
- if (!options.isEnableStreamingEngine()) {
+ if (!hasExperiment(options, "enable_streaming_auto_sharding")) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
+ "Runner determined sharding not enabled in Dataflow for GroupIntoBatches for Streaming "
+ + "Engine jobs. Default sharding will be applied.");
return;
}
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
- }
+ pcollectionsRequiringAutoSharding.add(pcol);
Review comment:
Good catch. Added the check.
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,27 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
+ if (!options.isEnableStreamingEngine()) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for "
+ + "non-Streaming Engine jobs. Default sharding will be applied.");
return;
}
- if (!options.isEnableStreamingEngine()) {
+ if (!hasExperiment(options, "enable_streaming_auto_sharding")) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
+ "Runner determined sharding not enabled in Dataflow for GroupIntoBatches for Streaming "
+ + "Engine jobs: --enable_streaming_auto_sharding=false. Default sharding will be "
+ + "applied.");
return;
}
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
+ if (hasExperiment(options, "beam_fn_api") && !hasExperiment(options, "use_runner_v2")) {
+ LOG.warn(
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
+ + "jobs not using runner v2: --beam_fn_api=true, --use_runner_v2=false. Default "
+ + "sharding will be applied.");
+ return;
Review comment:
Actually on a second thought, do we have to reject JRH? I guess it can work (although I haven't verified). Would it be a better user experience if we fail the job when autosharding falls back to the default implementation?
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,27 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
+ if (!options.isEnableStreamingEngine()) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for "
+ + "non-Streaming Engine jobs. Default sharding will be applied.");
return;
}
- if (!options.isEnableStreamingEngine()) {
+ if (!hasExperiment(options, "enable_streaming_auto_sharding")) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
+ "Runner determined sharding not enabled in Dataflow for GroupIntoBatches for Streaming "
+ + "Engine jobs: --enable_streaming_auto_sharding=false. Default sharding will be "
+ + "applied.");
return;
}
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
+ if (hasExperiment(options, "beam_fn_api") && !hasExperiment(options, "use_runner_v2")) {
+ LOG.warn(
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
+ + "jobs not using runner v2: --beam_fn_api=true, --use_runner_v2=false. Default "
+ + "sharding will be applied.");
+ return;
Review comment:
Thanks. Replaced the warnings to checkArguments, and added tests accordingly. PTAL.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #13805:
URL: https://github.com/apache/beam/pull/13805#discussion_r569007893
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,22 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
- LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
- return;
- }
- if (!options.isEnableStreamingEngine()) {
- LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
- return;
- }
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
- }
+ // Auto-sharding is only supported in Streaming Engine.
+ checkArgument(
+ options.isEnableStreamingEngine(),
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for"
Review comment:
The error messages here and below sounds confusing to me. If we find that pipeline options is not valid, how about we give the correct pipeline options in the message? We can have this improvement in a separated PR.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nehsyc commented on a change in pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13805:
URL: https://github.com/apache/beam/pull/13805#discussion_r568357423
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,27 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
+ if (!options.isEnableStreamingEngine()) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for "
+ + "non-Streaming Engine jobs. Default sharding will be applied.");
return;
}
- if (!options.isEnableStreamingEngine()) {
+ if (!hasExperiment(options, "enable_streaming_auto_sharding")) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
+ "Runner determined sharding not enabled in Dataflow for GroupIntoBatches for Streaming "
+ + "Engine jobs: --enable_streaming_auto_sharding=false. Default sharding will be "
+ + "applied.");
return;
}
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
+ if (hasExperiment(options, "beam_fn_api") && !hasExperiment(options, "use_runner_v2")) {
+ LOG.warn(
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
+ + "jobs not using runner v2: --beam_fn_api=true, --use_runner_v2=false. Default "
+ + "sharding will be applied.");
+ return;
Review comment:
Actually on a second thought, do we have to reject JRH? I guess it can work (although I haven't verified). Would it be a better user experience if we fail the job when autosharding falls back to the default implementation?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #13805:
URL: https://github.com/apache/beam/pull/13805#discussion_r568346093
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,27 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
+ if (!options.isEnableStreamingEngine()) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for "
+ + "non-Streaming Engine jobs. Default sharding will be applied.");
return;
}
- if (!options.isEnableStreamingEngine()) {
+ if (!hasExperiment(options, "enable_streaming_auto_sharding")) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
+ "Runner determined sharding not enabled in Dataflow for GroupIntoBatches for Streaming "
+ + "Engine jobs: --enable_streaming_auto_sharding=false. Default sharding will be "
+ + "applied.");
return;
}
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
+ if (hasExperiment(options, "beam_fn_api") && !hasExperiment(options, "use_runner_v2")) {
+ LOG.warn(
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
+ + "jobs not using runner v2: --beam_fn_api=true, --use_runner_v2=false. Default "
+ + "sharding will be applied.");
+ return;
Review comment:
Will a warning message be enough here? Should we reject the job? (I should comment on this early. Sorry that I forgot to do so : ( )
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #13805:
URL: https://github.com/apache/beam/pull/13805#issuecomment-771095694
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #13805:
URL: https://github.com/apache/beam/pull/13805#issuecomment-772131397
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #13805:
URL: https://github.com/apache/beam/pull/13805#discussion_r568346093
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,27 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
+ if (!options.isEnableStreamingEngine()) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for "
+ + "non-Streaming Engine jobs. Default sharding will be applied.");
return;
}
- if (!options.isEnableStreamingEngine()) {
+ if (!hasExperiment(options, "enable_streaming_auto_sharding")) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
+ "Runner determined sharding not enabled in Dataflow for GroupIntoBatches for Streaming "
+ + "Engine jobs: --enable_streaming_auto_sharding=false. Default sharding will be "
+ + "applied.");
return;
}
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
+ if (hasExperiment(options, "beam_fn_api") && !hasExperiment(options, "use_runner_v2")) {
+ LOG.warn(
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
+ + "jobs not using runner v2: --beam_fn_api=true, --use_runner_v2=false. Default "
+ + "sharding will be applied.");
+ return;
Review comment:
Will a warning message be enough here? Should we reject the job? (I should comment on this early. Sorry that I forgot to do so : ( )
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,27 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
+ if (!options.isEnableStreamingEngine()) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for "
+ + "non-Streaming Engine jobs. Default sharding will be applied.");
return;
}
- if (!options.isEnableStreamingEngine()) {
+ if (!hasExperiment(options, "enable_streaming_auto_sharding")) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
+ "Runner determined sharding not enabled in Dataflow for GroupIntoBatches for Streaming "
+ + "Engine jobs: --enable_streaming_auto_sharding=false. Default sharding will be "
+ + "applied.");
return;
}
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
+ if (hasExperiment(options, "beam_fn_api") && !hasExperiment(options, "use_runner_v2")) {
+ LOG.warn(
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
+ + "jobs not using runner v2: --beam_fn_api=true, --use_runner_v2=false. Default "
+ + "sharding will be applied.");
+ return;
Review comment:
I would prefer rejecting the pipeline since it's also bad when customer expects autosharding but it's not.
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,22 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
- LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
- return;
- }
- if (!options.isEnableStreamingEngine()) {
- LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
- return;
- }
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
- }
+ // Auto-sharding is only supported in Streaming Engine.
+ checkArgument(
+ options.isEnableStreamingEngine(),
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for"
Review comment:
The error messages here and below sounds confusing to me. If we find that pipeline options is not valid, how about we give the correct pipeline options in the message? We can have this improvement in a separated PR.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nehsyc commented on a change in pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13805:
URL: https://github.com/apache/beam/pull/13805#discussion_r568977188
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,27 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
+ if (!options.isEnableStreamingEngine()) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for "
+ + "non-Streaming Engine jobs. Default sharding will be applied.");
return;
}
- if (!options.isEnableStreamingEngine()) {
+ if (!hasExperiment(options, "enable_streaming_auto_sharding")) {
LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
+ "Runner determined sharding not enabled in Dataflow for GroupIntoBatches for Streaming "
+ + "Engine jobs: --enable_streaming_auto_sharding=false. Default sharding will be "
+ + "applied.");
return;
}
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
+ if (hasExperiment(options, "beam_fn_api") && !hasExperiment(options, "use_runner_v2")) {
+ LOG.warn(
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
+ + "jobs not using runner v2: --beam_fn_api=true, --use_runner_v2=false. Default "
+ + "sharding will be applied.");
+ return;
Review comment:
Thanks. Replaced the warnings to checkArguments, and added tests accordingly. PTAL.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #13805: [BEAM-10475] Add transform translator for GroupIntoBatches in Java.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #13805:
URL: https://github.com/apache/beam/pull/13805#discussion_r569671565
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
##########
@@ -1365,21 +1365,22 @@ void addPCollectionRequiringIndexedFormat(PCollection<?> pcol) {
}
void maybeRecordPCollectionWithAutoSharding(PCollection<?> pcol) {
- if (hasExperiment(options, "beam_fn_api")) {
- LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for portable "
- + "jobs. Default sharding will be applied.");
- return;
- }
- if (!options.isEnableStreamingEngine()) {
- LOG.warn(
- "Runner determined sharding not available in Dataflow for GroupIntoBatches for Streaming "
- + "Appliance jobs. Default sharding will be applied.");
- return;
- }
- if (hasExperiment(options, "enable_streaming_auto_sharding")) {
- pcollectionsRequiringAutoSharding.add(pcol);
- }
+ // Auto-sharding is only supported in Streaming Engine.
+ checkArgument(
+ options.isEnableStreamingEngine(),
+ "Runner determined sharding not available in Dataflow for GroupIntoBatches for"
+ + " non-Streaming-Engine jobs.");
Review comment:
I'm thinking maybe that we give the full instructions will be more helpful. For example, `In order to use runenr determined sharding, please use --experiments=enable_streaming_auto_sharding, --experiments=use_runner_v2, --streaming, --enable_streaming_engine`. But it really depends on you.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org