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/06/02 15:18:39 UTC
[GitHub] [beam] dmvk opened a new pull request #11889: [BEAM-10164] Flink Batch Runner: Memory efficient combine implementat…
dmvk opened a new pull request #11889:
URL: https://github.com/apache/beam/pull/11889
https://issues.apache.org/jira/browse/BEAM-10164
------------------------
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 | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/)
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/)
XLang | --- | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/)
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://builds.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://builds.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.
----------------------------------------------------------------
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 #11889: [BEAM-10164] Flink Batch Runner: Memory efficient combine implementat…
Posted by GitBox <gi...@apache.org>.
dmvk commented on pull request #11889:
URL: https://github.com/apache/beam/pull/11889#issuecomment-637792716
Run Flink Validates Runner
----------------------------------------------------------------
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 #11889: [BEAM-10164] Flink Batch Runner: Memory efficient combine implementat…
Posted by GitBox <gi...@apache.org>.
dmvk commented on pull request #11889:
URL: https://github.com/apache/beam/pull/11889#issuecomment-637793402
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 pull request #11889: [BEAM-10164] Flink Batch Runner: Memory efficient combine implementat…
Posted by GitBox <gi...@apache.org>.
dmvk commented on pull request #11889:
URL: https://github.com/apache/beam/pull/11889#issuecomment-637769674
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] mxm commented on a change in pull request #11889: [BEAM-10164] Flink Batch Runner: Memory efficient combine implementat…
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #11889:
URL: https://github.com/apache/beam/pull/11889#discussion_r434568958
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkBatchTransformTranslators.java
##########
@@ -476,16 +476,15 @@ public void translateNode(
public void translateNode(
PTransform<PCollection<KV<K, InputT>>, PCollection<KV<K, OutputT>>> transform,
FlinkBatchTranslationContext context) {
- DataSet<WindowedValue<KV<K, InputT>>> inputDataSet =
+ final DataSet<WindowedValue<KV<K, InputT>>> inputDataSet =
context.getInputDataSet(context.getInput(transform));
-
- CombineFnBase.GlobalCombineFn<InputT, AccumT, OutputT> combineFn =
- ((Combine.PerKey) transform).getFn();
-
- KvCoder<K, InputT> inputCoder = (KvCoder<K, InputT>) context.getInput(transform).getCoder();
-
- Coder<AccumT> accumulatorCoder;
-
+ final Combine.PerKey<K, InputT, OutputT> combineTransform =
+ (Combine.PerKey<K, InputT, OutputT>) transform;
+ final CombineFnBase.GlobalCombineFn<InputT, AccumT, OutputT> combineFn =
+ (CombineFnBase.GlobalCombineFn<InputT, AccumT, OutputT>) combineTransform.getFn();
+ final KvCoder<K, InputT> inputCoder =
+ (KvCoder<K, InputT>) context.getInput(transform).getCoder();
+ final Coder<AccumT> accumulatorCoder;
Review comment:
Yes, it depends on the situation. I use `final` mostly when there is a clear assign-once relationship with commonly used types. Since type-parameterized types are used here, there is very little chance of accidentally overwriting a ref.
Just a discussion point. No need to revert.
----------------------------------------------------------------
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 merged pull request #11889: [BEAM-10164] Flink Batch Runner: Memory efficient combine implementat…
Posted by GitBox <gi...@apache.org>.
mxm merged pull request #11889:
URL: https://github.com/apache/beam/pull/11889
----------------------------------------------------------------
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 #11889: [BEAM-10164] Flink Batch Runner: Memory efficient combine implementat…
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #11889:
URL: https://github.com/apache/beam/pull/11889#discussion_r434515497
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkBatchTransformTranslators.java
##########
@@ -476,16 +476,15 @@ public void translateNode(
public void translateNode(
PTransform<PCollection<KV<K, InputT>>, PCollection<KV<K, OutputT>>> transform,
FlinkBatchTranslationContext context) {
- DataSet<WindowedValue<KV<K, InputT>>> inputDataSet =
+ final DataSet<WindowedValue<KV<K, InputT>>> inputDataSet =
context.getInputDataSet(context.getInput(transform));
-
- CombineFnBase.GlobalCombineFn<InputT, AccumT, OutputT> combineFn =
- ((Combine.PerKey) transform).getFn();
-
- KvCoder<K, InputT> inputCoder = (KvCoder<K, InputT>) context.getInput(transform).getCoder();
-
- Coder<AccumT> accumulatorCoder;
-
+ final Combine.PerKey<K, InputT, OutputT> combineTransform =
+ (Combine.PerKey<K, InputT, OutputT>) transform;
+ final CombineFnBase.GlobalCombineFn<InputT, AccumT, OutputT> combineFn =
+ (CombineFnBase.GlobalCombineFn<InputT, AccumT, OutputT>) combineTransform.getFn();
+ final KvCoder<K, InputT> inputCoder =
+ (KvCoder<K, InputT>) context.getInput(transform).getCoder();
+ final Coder<AccumT> accumulatorCoder;
Review comment:
Code style: It would be great if refs were immutable by default, but adding final to everything I find adds verbosity with little added benefit.
----------------------------------------------------------------
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 #11889: [BEAM-10164] Flink Batch Runner: Memory efficient combine implementat…
Posted by GitBox <gi...@apache.org>.
dmvk commented on a change in pull request #11889:
URL: https://github.com/apache/beam/pull/11889#discussion_r434528990
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkBatchTransformTranslators.java
##########
@@ -476,16 +476,15 @@ public void translateNode(
public void translateNode(
PTransform<PCollection<KV<K, InputT>>, PCollection<KV<K, OutputT>>> transform,
FlinkBatchTranslationContext context) {
- DataSet<WindowedValue<KV<K, InputT>>> inputDataSet =
+ final DataSet<WindowedValue<KV<K, InputT>>> inputDataSet =
context.getInputDataSet(context.getInput(transform));
-
- CombineFnBase.GlobalCombineFn<InputT, AccumT, OutputT> combineFn =
- ((Combine.PerKey) transform).getFn();
-
- KvCoder<K, InputT> inputCoder = (KvCoder<K, InputT>) context.getInput(transform).getCoder();
-
- Coder<AccumT> accumulatorCoder;
-
+ final Combine.PerKey<K, InputT, OutputT> combineTransform =
+ (Combine.PerKey<K, InputT, OutputT>) transform;
+ final CombineFnBase.GlobalCombineFn<InputT, AccumT, OutputT> combineFn =
+ (CombineFnBase.GlobalCombineFn<InputT, AccumT, OutputT>) combineTransform.getFn();
+ final KvCoder<K, InputT> inputCoder =
+ (KvCoder<K, InputT>) context.getInput(transform).getCoder();
+ final Coder<AccumT> accumulatorCoder;
Review comment:
I agree that the benefit is discutable.
Personally I find it helpful when it comes to long methods or trying to understand someone elses code, that you see that there would be no more assignents to the variable (if there is no final modifier, you know that there is something more going on).
Anyway, it's more of a personal taste. Do you want me to revert these?
----------------------------------------------------------------
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