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 2020/11/16 14:56:28 UTC
[GitHub] [beam] dmvk opened a new pull request #13353: [BEAM-11267] Remove unecessary reshuffle for stateful ParDo after key…
dmvk opened a new pull request #13353:
URL: https://github.com/apache/beam/pull/13353
https://issues.apache.org/jira/browse/BEAM-11267
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**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_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] dmvk commented on pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
dmvk commented on pull request #13353:
URL: https://github.com/apache/beam/pull/13353#issuecomment-746142267
Run Flink ValidatesRunner
----------------------------------------------------------------
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] dmvk commented on a change in pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
dmvk commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r526640194
##########
File path: runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslatorTest.java
##########
@@ -120,4 +138,95 @@ public void testAutoBalanceShardKeyCacheMaxSize() throws Exception {
assertThat(
fn.getCache().size(), equalTo(FlinkAutoBalancedShardKeyShardingFunction.CACHE_MAX_SIZE));
}
+
+ @Test
+ public void testStatefulParDoAfterCombineChaining() {
+ final JobGraph stablePartitioning = getStatefulParDoAfterCombineChainingJobGraph(true);
+ final JobGraph unstablePartitioning = getStatefulParDoAfterCombineChainingJobGraph(false);
+ // We expect an extra shuffle stage for unstable partitioning.
+ Assert.assertEquals(
+ 1,
+ Iterables.size(unstablePartitioning.getVertices())
+ - Iterables.size(stablePartitioning.getVertices()));
+ }
+
+ private JobGraph getStatefulParDoAfterCombineChainingJobGraph(boolean stablePartitioning) {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ final FlinkStreamingPipelineTranslator translator =
+ new FlinkStreamingPipelineTranslator(env, PipelineOptionsFactory.create());
+ final PipelineOptions pipelineOptions = PipelineOptionsFactory.create();
+ pipelineOptions.setRunner(FlinkRunner.class);
Review comment:
unfortunately yes, otherwise pipeline.create() would fail
----------------------------------------------------------------
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] je-ik commented on a change in pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r524597532
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WorkItemKeySelector.java
##########
@@ -49,6 +52,6 @@ public ByteBuffer getKey(WindowedValue<SingletonKeyedWorkItem<K, V>> value) thro
@Override
public TypeInformation<ByteBuffer> getProducedType() {
- return new GenericTypeInfo<>(ByteBuffer.class);
+ return new CoderTypeInformation<>(FlinkKeyUtils.ByteBufferCoder.of(), pipelineOptions.get());
Review comment:
I would think that the partitioning is ensured by the `reinterpretAsKeyedStream` and will therefore be preserved from the previous shuffle phase. Is this not enough?
----------------------------------------------------------------
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] mxm commented on a change in pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r524634584
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
##########
@@ -971,7 +987,9 @@ public void translateNode(
.transform(fullName, outputTypeInfo, (OneInputStreamOperator) doFnOperator)
.uid(fullName);
- context.setOutputDataStream(context.getOutput(transform), outDataStream);
+ final PCollection<KV<K, Iterable<InputT>>> output = context.getOutput(transform);
+ context.setOutputDataStream(output, outDataStream);
+ context.setProducer(output, transform);
Review comment:
It would be nice if these explicit calls wouldn't be required. I believe `context.setOutputDataStream` internally has the current transform available. So we could perform it internally in the context.
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
##########
@@ -1127,7 +1148,9 @@ public void translateNode(
keyedWorkItemStream.getExecutionEnvironment().addOperator(rawFlinkTransform);
- context.setOutputDataStream(context.getOutput(transform), outDataStream);
+ final PCollection<KV<K, OutputT>> output = context.getOutput(transform);
+ context.setOutputDataStream(output, outDataStream);
+ context.setProducer(output, transform);
Review comment:
Same here.
----------------------------------------------------------------
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] dmvk commented on a change in pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
dmvk commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r526382315
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
##########
@@ -971,7 +987,9 @@ public void translateNode(
.transform(fullName, outputTypeInfo, (OneInputStreamOperator) doFnOperator)
.uid(fullName);
- context.setOutputDataStream(context.getOutput(transform), outDataStream);
+ final PCollection<KV<K, Iterable<InputT>>> output = context.getOutput(transform);
+ context.setOutputDataStream(output, outDataStream);
+ context.setProducer(output, transform);
Review comment:
nice catch 👍
----------------------------------------------------------------
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] dmvk commented on a change in pull request #13353: [BEAM-11267] Remove unecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
dmvk commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r524545316
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WorkItemKeySelector.java
##########
@@ -49,6 +52,6 @@ public ByteBuffer getKey(WindowedValue<SingletonKeyedWorkItem<K, V>> value) thro
@Override
public TypeInformation<ByteBuffer> getProducedType() {
- return new GenericTypeInfo<>(ByteBuffer.class);
+ return new CoderTypeInformation<>(FlinkKeyUtils.ByteBufferCoder.of(), pipelineOptions.get());
Review comment:
Not sure if this is necessary. I wanted to ensure that the new "reinterpreted partitioning" is compatible with the one used by GBK / Combine.
The idea was if partitioning is not compatible, it may result in some state partitioning related glitches (eg. you wouldn't have local state for a key-group you need).
Second thoughts, flink selects target partition (key group) based on "pojo hash code" (not based on binary representation), so the previous version was probably compatible enough 🤔
@mxm WDYT?
----------------------------------------------------------------
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] dmvk commented on a change in pull request #13353: [BEAM-11267] Remove unecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
dmvk commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r524545989
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTranslationContext.java
##########
@@ -84,6 +85,17 @@ public void setOutputDataStream(PValue value, DataStream<?> set) {
}
}
+ <T extends PValue> void setProducer(T value, PTransform<?, T> producer) {
+ if (!producers.containsKey(value)) {
Review comment:
👍 makes sense
----------------------------------------------------------------
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] dmvk commented on a change in pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
dmvk commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r526383542
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WorkItemKeySelector.java
##########
@@ -49,6 +52,6 @@ public ByteBuffer getKey(WindowedValue<SingletonKeyedWorkItem<K, V>> value) thro
@Override
public TypeInformation<ByteBuffer> getProducedType() {
- return new GenericTypeInfo<>(ByteBuffer.class);
+ return new CoderTypeInformation<>(FlinkKeyUtils.ByteBufferCoder.of(), pipelineOptions.get());
Review comment:
I think that `reinterpretAsKeyedStream` just trusts the user and there are no safety nets in place.
----------------------------------------------------------------
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] je-ik commented on pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
je-ik commented on pull request #13353:
URL: https://github.com/apache/beam/pull/13353#issuecomment-728337299
Do we have a test (in flink runner) for the GBK -> stateful pardo pair? Not sure if there is one in ValidatesRunner suite.
----------------------------------------------------------------
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] dmvk commented on a change in pull request #13353: [BEAM-11267] Remove unecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
dmvk commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r524545316
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WorkItemKeySelector.java
##########
@@ -49,6 +52,6 @@ public ByteBuffer getKey(WindowedValue<SingletonKeyedWorkItem<K, V>> value) thro
@Override
public TypeInformation<ByteBuffer> getProducedType() {
- return new GenericTypeInfo<>(ByteBuffer.class);
+ return new CoderTypeInformation<>(FlinkKeyUtils.ByteBufferCoder.of(), pipelineOptions.get());
Review comment:
Not sure if this is necessary. I wanted to ensure that the new "reinterpreted partitioning" is compatible with the one used by GBK / Combine.
The idea was if partitioning is not compatible, it may result in some state partitioning related glitches (eg. you wouldn't have local state for a key-group you need).
Second thoughts, flink selects target partition based on "pojo hash code" (not based on binary representation), so the previous version was probably compatible enough 🤔
@mxm WDYT?
----------------------------------------------------------------
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] dmvk commented on pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
dmvk commented on pull request #13353:
URL: https://github.com/apache/beam/pull/13353#issuecomment-729924707
Run Flink ValidatesRunner
----------------------------------------------------------------
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] je-ik commented on a change in pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r526473302
##########
File path: runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslatorTest.java
##########
@@ -120,4 +138,95 @@ public void testAutoBalanceShardKeyCacheMaxSize() throws Exception {
assertThat(
fn.getCache().size(), equalTo(FlinkAutoBalancedShardKeyShardingFunction.CACHE_MAX_SIZE));
}
+
+ @Test
+ public void testStatefulParDoAfterCombineChaining() {
+ final JobGraph stablePartitioning = getStatefulParDoAfterCombineChainingJobGraph(true);
+ final JobGraph unstablePartitioning = getStatefulParDoAfterCombineChainingJobGraph(false);
+ // We expect an extra shuffle stage for unstable partitioning.
+ Assert.assertEquals(
+ 1,
+ Iterables.size(unstablePartitioning.getVertices())
+ - Iterables.size(stablePartitioning.getVertices()));
+ }
+
+ private JobGraph getStatefulParDoAfterCombineChainingJobGraph(boolean stablePartitioning) {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ final FlinkStreamingPipelineTranslator translator =
+ new FlinkStreamingPipelineTranslator(env, PipelineOptionsFactory.create());
+ final PipelineOptions pipelineOptions = PipelineOptionsFactory.create();
+ pipelineOptions.setRunner(FlinkRunner.class);
Review comment:
Do we need to set runner if we enforce the translator?
##########
File path: runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslatorTest.java
##########
@@ -120,4 +138,95 @@ public void testAutoBalanceShardKeyCacheMaxSize() throws Exception {
assertThat(
fn.getCache().size(), equalTo(FlinkAutoBalancedShardKeyShardingFunction.CACHE_MAX_SIZE));
}
+
+ @Test
+ public void testStatefulParDoAfterCombineChaining() {
+ final JobGraph stablePartitioning = getStatefulParDoAfterCombineChainingJobGraph(true);
+ final JobGraph unstablePartitioning = getStatefulParDoAfterCombineChainingJobGraph(false);
+ // We expect an extra shuffle stage for unstable partitioning.
+ Assert.assertEquals(
+ 1,
+ Iterables.size(unstablePartitioning.getVertices())
+ - Iterables.size(stablePartitioning.getVertices()));
+ }
+
+ private JobGraph getStatefulParDoAfterCombineChainingJobGraph(boolean stablePartitioning) {
Review comment:
I would call this parameter `forcesShuffle` or something similar.
##########
File path: runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslatorTest.java
##########
@@ -120,4 +138,95 @@ public void testAutoBalanceShardKeyCacheMaxSize() throws Exception {
assertThat(
fn.getCache().size(), equalTo(FlinkAutoBalancedShardKeyShardingFunction.CACHE_MAX_SIZE));
}
+
+ @Test
+ public void testStatefulParDoAfterCombineChaining() {
+ final JobGraph stablePartitioning = getStatefulParDoAfterCombineChainingJobGraph(true);
+ final JobGraph unstablePartitioning = getStatefulParDoAfterCombineChainingJobGraph(false);
+ // We expect an extra shuffle stage for unstable partitioning.
+ Assert.assertEquals(
+ 1,
+ Iterables.size(unstablePartitioning.getVertices())
+ - Iterables.size(stablePartitioning.getVertices()));
+ }
+
+ private JobGraph getStatefulParDoAfterCombineChainingJobGraph(boolean stablePartitioning) {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ final FlinkStreamingPipelineTranslator translator =
+ new FlinkStreamingPipelineTranslator(env, PipelineOptionsFactory.create());
+ final PipelineOptions pipelineOptions = PipelineOptionsFactory.create();
+ pipelineOptions.setRunner(FlinkRunner.class);
+ final Pipeline pipeline = Pipeline.create(pipelineOptions);
+ PCollection<KV<String, Long>> aggregate =
+ pipeline
+ .apply(Create.of("foo", "bar").withCoder(StringUtf8Coder.of()))
+ .apply(Count.perElement());
+ if (!stablePartitioning) {
+ // When we insert any element-wise "map" operation between aggregation and stateful ParDo, we
+ // can no longer assume that partitioning did not change, therefore we need an extra shuffle
+ aggregate = aggregate.apply(ParDo.of(new StatelessIdentityDoFn<>()));
Review comment:
Instead of the `StatelessIdentityDoFn` we could use `MapElements.into(...).via(e -> KV.of("", e.getValue())`, which would enforce shuffle semantically. That might improve readability a bit.
##########
File path: runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslatorTest.java
##########
@@ -120,4 +138,95 @@ public void testAutoBalanceShardKeyCacheMaxSize() throws Exception {
assertThat(
fn.getCache().size(), equalTo(FlinkAutoBalancedShardKeyShardingFunction.CACHE_MAX_SIZE));
}
+
+ @Test
+ public void testStatefulParDoAfterCombineChaining() {
+ final JobGraph stablePartitioning = getStatefulParDoAfterCombineChainingJobGraph(true);
+ final JobGraph unstablePartitioning = getStatefulParDoAfterCombineChainingJobGraph(false);
+ // We expect an extra shuffle stage for unstable partitioning.
+ Assert.assertEquals(
+ 1,
+ Iterables.size(unstablePartitioning.getVertices())
+ - Iterables.size(stablePartitioning.getVertices()));
+ }
+
+ private JobGraph getStatefulParDoAfterCombineChainingJobGraph(boolean stablePartitioning) {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ final FlinkStreamingPipelineTranslator translator =
+ new FlinkStreamingPipelineTranslator(env, PipelineOptionsFactory.create());
+ final PipelineOptions pipelineOptions = PipelineOptionsFactory.create();
+ pipelineOptions.setRunner(FlinkRunner.class);
+ final Pipeline pipeline = Pipeline.create(pipelineOptions);
+ PCollection<KV<String, Long>> aggregate =
+ pipeline
+ .apply(Create.of("foo", "bar").withCoder(StringUtf8Coder.of()))
+ .apply(Count.perElement());
+ if (!stablePartitioning) {
+ // When we insert any element-wise "map" operation between aggregation and stateful ParDo, we
+ // can no longer assume that partitioning did not change, therefore we need an extra shuffle
+ aggregate = aggregate.apply(ParDo.of(new StatelessIdentityDoFn<>()));
+ }
+ aggregate.apply(ParDo.of(new StatefulNoopDoFn<>()));
+ translator.translate(pipeline);
+ return env.getStreamGraph().getJobGraph();
+ }
+
+ @Test
+ public void testStatefulParDoAfterGroupByKeyChaining() {
+ final JobGraph stablePartitioning = getStatefulParDoAfterGroupByKeyChainingJobGraph(true);
+ final JobGraph unstablePartitioning = getStatefulParDoAfterGroupByKeyChainingJobGraph(false);
+ // We expect an extra shuffle stage for unstable partitioning.
+ Assert.assertEquals(
+ 1,
+ Iterables.size(unstablePartitioning.getVertices())
+ - Iterables.size(stablePartitioning.getVertices()));
+ }
+
+ private JobGraph getStatefulParDoAfterGroupByKeyChainingJobGraph(boolean stablePartitioning) {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ final FlinkStreamingPipelineTranslator translator =
+ new FlinkStreamingPipelineTranslator(env, PipelineOptionsFactory.create());
+ final PipelineOptions pipelineOptions = PipelineOptionsFactory.create();
+ pipelineOptions.setRunner(FlinkRunner.class);
+ final Pipeline pipeline = Pipeline.create(pipelineOptions);
+ PCollection<KV<String, Iterable<Long>>> aggregate =
+ pipeline
+ .apply(
+ Create.of(KV.of("foo", 1L), KV.of("bar", 1L))
+ .withCoder(KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of())))
+ .apply(GroupByKey.create());
Review comment:
Can we merge the two methods, that seem to differ only by this PTransform applied here?
----------------------------------------------------------------
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] je-ik commented on a change in pull request #13353: [BEAM-11267] Remove unecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r524333570
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTranslationContext.java
##########
@@ -84,6 +85,17 @@ public void setOutputDataStream(PValue value, DataStream<?> set) {
}
}
+ <T extends PValue> void setProducer(T value, PTransform<?, T> producer) {
+ if (!producers.containsKey(value)) {
Review comment:
Should this be turned to `Precondition.checkState`?
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WorkItemKeySelector.java
##########
@@ -49,6 +52,6 @@ public ByteBuffer getKey(WindowedValue<SingletonKeyedWorkItem<K, V>> value) thro
@Override
public TypeInformation<ByteBuffer> getProducedType() {
- return new GenericTypeInfo<>(ByteBuffer.class);
+ return new CoderTypeInformation<>(FlinkKeyUtils.ByteBufferCoder.of(), pipelineOptions.get());
Review comment:
This looks unrelated, can you just explain this modification? I suppose it is correct, just wonder why it was `GenericTypeInfo` before.
----------------------------------------------------------------
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] dmvk commented on pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
dmvk commented on pull request #13353:
URL: https://github.com/apache/beam/pull/13353#issuecomment-729949617
@je-ik @mxm thanks for the review ;) I've incorporated your suggestions and it's ready for another pass
----------------------------------------------------------------
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] mxm commented on a change in pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r524634584
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
##########
@@ -971,7 +987,9 @@ public void translateNode(
.transform(fullName, outputTypeInfo, (OneInputStreamOperator) doFnOperator)
.uid(fullName);
- context.setOutputDataStream(context.getOutput(transform), outDataStream);
+ final PCollection<KV<K, Iterable<InputT>>> output = context.getOutput(transform);
+ context.setOutputDataStream(output, outDataStream);
+ context.setProducer(output, transform);
Review comment:
It would be nice if these explicit calls wouldn't be required. I believe `context.setOutputDataStream` internally has the current transform available. So we could update the producer internally in the context.
----------------------------------------------------------------
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] dmvk commented on a change in pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
dmvk commented on a change in pull request #13353:
URL: https://github.com/apache/beam/pull/13353#discussion_r526641055
##########
File path: runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslatorTest.java
##########
@@ -120,4 +138,95 @@ public void testAutoBalanceShardKeyCacheMaxSize() throws Exception {
assertThat(
fn.getCache().size(), equalTo(FlinkAutoBalancedShardKeyShardingFunction.CACHE_MAX_SIZE));
}
+
+ @Test
+ public void testStatefulParDoAfterCombineChaining() {
+ final JobGraph stablePartitioning = getStatefulParDoAfterCombineChainingJobGraph(true);
+ final JobGraph unstablePartitioning = getStatefulParDoAfterCombineChainingJobGraph(false);
+ // We expect an extra shuffle stage for unstable partitioning.
+ Assert.assertEquals(
+ 1,
+ Iterables.size(unstablePartitioning.getVertices())
+ - Iterables.size(stablePartitioning.getVertices()));
+ }
+
+ private JobGraph getStatefulParDoAfterCombineChainingJobGraph(boolean stablePartitioning) {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ final FlinkStreamingPipelineTranslator translator =
+ new FlinkStreamingPipelineTranslator(env, PipelineOptionsFactory.create());
+ final PipelineOptions pipelineOptions = PipelineOptionsFactory.create();
+ pipelineOptions.setRunner(FlinkRunner.class);
+ final Pipeline pipeline = Pipeline.create(pipelineOptions);
+ PCollection<KV<String, Long>> aggregate =
+ pipeline
+ .apply(Create.of("foo", "bar").withCoder(StringUtf8Coder.of()))
+ .apply(Count.perElement());
+ if (!stablePartitioning) {
+ // When we insert any element-wise "map" operation between aggregation and stateful ParDo, we
+ // can no longer assume that partitioning did not change, therefore we need an extra shuffle
+ aggregate = aggregate.apply(ParDo.of(new StatelessIdentityDoFn<>()));
+ }
+ aggregate.apply(ParDo.of(new StatefulNoopDoFn<>()));
+ translator.translate(pipeline);
+ return env.getStreamGraph().getJobGraph();
+ }
+
+ @Test
+ public void testStatefulParDoAfterGroupByKeyChaining() {
+ final JobGraph stablePartitioning = getStatefulParDoAfterGroupByKeyChainingJobGraph(true);
+ final JobGraph unstablePartitioning = getStatefulParDoAfterGroupByKeyChainingJobGraph(false);
+ // We expect an extra shuffle stage for unstable partitioning.
+ Assert.assertEquals(
+ 1,
+ Iterables.size(unstablePartitioning.getVertices())
+ - Iterables.size(stablePartitioning.getVertices()));
+ }
+
+ private JobGraph getStatefulParDoAfterGroupByKeyChainingJobGraph(boolean stablePartitioning) {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ final FlinkStreamingPipelineTranslator translator =
+ new FlinkStreamingPipelineTranslator(env, PipelineOptionsFactory.create());
+ final PipelineOptions pipelineOptions = PipelineOptionsFactory.create();
+ pipelineOptions.setRunner(FlinkRunner.class);
+ final Pipeline pipeline = Pipeline.create(pipelineOptions);
+ PCollection<KV<String, Iterable<Long>>> aggregate =
+ pipeline
+ .apply(
+ Create.of(KV.of("foo", 1L), KV.of("bar", 1L))
+ .withCoder(KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of())))
+ .apply(GroupByKey.create());
Review comment:
I thought about it, but the methods are really small so it's worth decreasing readability here
----------------------------------------------------------------
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] dmvk merged pull request #13353: [BEAM-11267] Remove unnecessary reshuffle for stateful ParDo after key…
Posted by GitBox <gi...@apache.org>.
dmvk merged pull request #13353:
URL: https://github.com/apache/beam/pull/13353
----------------------------------------------------------------
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