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/04/07 18:40:18 UTC
[GitHub] [beam] rohdesamuel opened a new pull request #11335: [BEAM-9692]:
Make CombineValues portable
rohdesamuel opened a new pull request #11335: [BEAM-9692]: Make CombineValues portable
URL: https://github.com/apache/beam/pull/11335
Change-Id: If1835e4d14320149b3e18eea9c97a26d3d34ee48
**Please** add a meaningful description for your change here
------------------------
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_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
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11335:
[BEAM-9692]: Make CombineValues portable
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11335: [BEAM-9692]: Make CombineValues portable
URL: https://github.com/apache/beam/pull/11335#discussion_r407758508
##########
File path: sdks/python/apache_beam/runners/dataflow/dataflow_runner_test.py
##########
@@ -566,6 +566,19 @@ def test_get_default_gcp_region_ignores_error(
result = runner.get_default_gcp_region()
self.assertIsNone(result)
+ def test_combine_values_translation(self):
+ runner = DataflowRunner()
+
+ with beam.Pipeline(runner=runner,
+ options=PipelineOptions(self.default_properties)) as p:
+ ( # pylint: disable=expression-not-assigned
+ p
+ | beam.Create([('a', [1, 2]), ('b', [3, 4])])
+ | beam.CombineValues(lambda v, _: sum(v)))
+
+ job_dict = json.loads(str(runner.job))
+ self.assertEqual(job_dict[u'steps'][1][u'kind'], u'CombineValues')
Review comment:
Asserting that it's the first step seems brittle, maybe just assert that there is some step that has this kind?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11335:
[BEAM-9692]: Make CombineValues portable
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11335: [BEAM-9692]: Make CombineValues portable
URL: https://github.com/apache/beam/pull/11335#discussion_r408289779
##########
File path: sdks/python/apache_beam/runners/dataflow/ptransform_overrides.py
##########
@@ -111,3 +111,38 @@ def expand(self, pbegin):
return JrhRead().with_output_types(
ptransform.get_type_hints().simple_output_type('Read'))
+
+
+class CombineValuesPTransformOverride(PTransformOverride):
+ """A ``PTransformOverride`` for ``CombineValues``.
+
+ The DataflowRunner expects that the CombineValues PTransform acts as a
+ primitive. So this override replaces the CombineValues with a primitive.
+ """
+ def matches(self, applied_ptransform):
+ # Imported here to avoid circular dependencies.
+ # pylint: disable=wrong-import-order, wrong-import-position
+ from apache_beam import CombineValues
+
+ if isinstance(applied_ptransform.transform, CombineValues):
+ self.transform = applied_ptransform.transform
+ return True
+ return False
+
+ def get_replacement_transform(self, ptransform):
+ # Imported here to avoid circular dependencies.
+ # pylint: disable=wrong-import-order, wrong-import-position
+ from apache_beam import PTransform
+ from apache_beam.pvalue import PCollection
+
+ # The DataflowRunner still needs access to the CombineValues members to
Review comment:
Sorry, I don't understand what you mean. Can you elaborate? What does "let try and find methods for composites as well" mean? What would an alternative to PTransformOverrides be?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11335:
[BEAM-9692]: Make CombineValues portable
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11335: [BEAM-9692]: Make CombineValues portable
URL: https://github.com/apache/beam/pull/11335#discussion_r407760055
##########
File path: sdks/python/apache_beam/runners/dataflow/ptransform_overrides.py
##########
@@ -111,3 +111,38 @@ def expand(self, pbegin):
return JrhRead().with_output_types(
ptransform.get_type_hints().simple_output_type('Read'))
+
+
+class CombineValuesPTransformOverride(PTransformOverride):
+ """A ``PTransformOverride`` for ``CombineValues``.
+
+ The DataflowRunner expects that the CombineValues PTransform acts as a
+ primitive. So this override replaces the CombineValues with a primitive.
+ """
+ def matches(self, applied_ptransform):
+ # Imported here to avoid circular dependencies.
+ # pylint: disable=wrong-import-order, wrong-import-position
+ from apache_beam import CombineValues
+
+ if isinstance(applied_ptransform.transform, CombineValues):
+ self.transform = applied_ptransform.transform
+ return True
+ return False
+
+ def get_replacement_transform(self, ptransform):
+ # Imported here to avoid circular dependencies.
+ # pylint: disable=wrong-import-order, wrong-import-position
+ from apache_beam import PTransform
+ from apache_beam.pvalue import PCollection
+
+ # The DataflowRunner still needs access to the CombineValues members to
Review comment:
It would be preferable to simply let try and find methods for composites as well, rather than using PTransformOverrides. This would likely help with the GBK one too.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11335:
[BEAM-9692]: Make CombineValues portable
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11335: [BEAM-9692]: Make CombineValues portable
URL: https://github.com/apache/beam/pull/11335#discussion_r408461337
##########
File path: sdks/python/apache_beam/runners/dataflow/ptransform_overrides.py
##########
@@ -111,3 +111,38 @@ def expand(self, pbegin):
return JrhRead().with_output_types(
ptransform.get_type_hints().simple_output_type('Read'))
+
+
+class CombineValuesPTransformOverride(PTransformOverride):
+ """A ``PTransformOverride`` for ``CombineValues``.
+
+ The DataflowRunner expects that the CombineValues PTransform acts as a
+ primitive. So this override replaces the CombineValues with a primitive.
+ """
+ def matches(self, applied_ptransform):
+ # Imported here to avoid circular dependencies.
+ # pylint: disable=wrong-import-order, wrong-import-position
+ from apache_beam import CombineValues
+
+ if isinstance(applied_ptransform.transform, CombineValues):
+ self.transform = applied_ptransform.transform
+ return True
+ return False
+
+ def get_replacement_transform(self, ptransform):
+ # Imported here to avoid circular dependencies.
+ # pylint: disable=wrong-import-order, wrong-import-position
+ from apache_beam import PTransform
+ from apache_beam.pvalue import PCollection
+
+ # The DataflowRunner still needs access to the CombineValues members to
Review comment:
I was thinking that run_xxx could also be called for composites. That might, however, be a bigger change, so we can go with this approach.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on issue #11335: [BEAM-9692]: Make
CombineValues portable
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on issue #11335: [BEAM-9692]: Make CombineValues portable
URL: https://github.com/apache/beam/pull/11335#issuecomment-613023893
R: @robertwb
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11335:
[BEAM-9692]: Make CombineValues portable
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11335: [BEAM-9692]: Make CombineValues portable
URL: https://github.com/apache/beam/pull/11335#discussion_r409125734
##########
File path: sdks/python/apache_beam/runners/dataflow/dataflow_runner_test.py
##########
@@ -566,6 +566,19 @@ def test_get_default_gcp_region_ignores_error(
result = runner.get_default_gcp_region()
self.assertIsNone(result)
+ def test_combine_values_translation(self):
+ runner = DataflowRunner()
+
+ with beam.Pipeline(runner=runner,
+ options=PipelineOptions(self.default_properties)) as p:
+ ( # pylint: disable=expression-not-assigned
+ p
+ | beam.Create([('a', [1, 2]), ('b', [3, 4])])
+ | beam.CombineValues(lambda v, _: sum(v)))
+
+ job_dict = json.loads(str(runner.job))
+ self.assertEqual(job_dict[u'steps'][1][u'kind'], u'CombineValues')
Review comment:
Done, changed to assertIn
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11335:
[BEAM-9692]: Make CombineValues portable
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11335: [BEAM-9692]: Make CombineValues portable
URL: https://github.com/apache/beam/pull/11335#discussion_r407757918
##########
File path: sdks/python/apache_beam/runners/dataflow/dataflow_runner.py
##########
@@ -110,22 +110,27 @@ class DataflowRunner(PipelineRunner):
# Imported here to avoid circular dependencies.
# TODO: Remove the apache_beam.pipeline dependency in CreatePTransformOverride
+ from apache_beam.runners.dataflow.ptransform_overrides import CombineValuesPTransformOverride
from apache_beam.runners.dataflow.ptransform_overrides import CreatePTransformOverride
from apache_beam.runners.dataflow.ptransform_overrides import ReadPTransformOverride
from apache_beam.runners.dataflow.ptransform_overrides import JrhReadPTransformOverride
- _PTRANSFORM_OVERRIDES = [] # type: List[PTransformOverride]
+ # Thesse overrides should be applied before the proto representation of the
+ # graph is created.
+ _PTRANSFORM_OVERRIDES = [
+ CombineValuesPTransformOverride()
Review comment:
Seems this one should happen after too...
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [beam] rohdesamuel commented on a change in pull request #11335:
[BEAM-9692]: Make CombineValues portable
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11335: [BEAM-9692]: Make CombineValues portable
URL: https://github.com/apache/beam/pull/11335#discussion_r407783064
##########
File path: sdks/python/apache_beam/runners/dataflow/dataflow_runner.py
##########
@@ -110,22 +110,27 @@ class DataflowRunner(PipelineRunner):
# Imported here to avoid circular dependencies.
# TODO: Remove the apache_beam.pipeline dependency in CreatePTransformOverride
+ from apache_beam.runners.dataflow.ptransform_overrides import CombineValuesPTransformOverride
from apache_beam.runners.dataflow.ptransform_overrides import CreatePTransformOverride
from apache_beam.runners.dataflow.ptransform_overrides import ReadPTransformOverride
from apache_beam.runners.dataflow.ptransform_overrides import JrhReadPTransformOverride
- _PTRANSFORM_OVERRIDES = [] # type: List[PTransformOverride]
+ # Thesse overrides should be applied before the proto representation of the
+ # graph is created.
+ _PTRANSFORM_OVERRIDES = [
+ CombineValuesPTransformOverride()
Review comment:
This override should place the pipeline object into the same state as if the runner had defined an apply_CombineValues, what am I missing? Looking at the code, is it because other overrides might also use a CombineValues transform so it might needed to be replaced again?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11335:
[BEAM-9692]: Make CombineValues portable
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11335: [BEAM-9692]: Make CombineValues portable
URL: https://github.com/apache/beam/pull/11335#discussion_r408461679
##########
File path: sdks/python/apache_beam/runners/dataflow/dataflow_runner.py
##########
@@ -110,22 +110,27 @@ class DataflowRunner(PipelineRunner):
# Imported here to avoid circular dependencies.
# TODO: Remove the apache_beam.pipeline dependency in CreatePTransformOverride
+ from apache_beam.runners.dataflow.ptransform_overrides import CombineValuesPTransformOverride
from apache_beam.runners.dataflow.ptransform_overrides import CreatePTransformOverride
from apache_beam.runners.dataflow.ptransform_overrides import ReadPTransformOverride
from apache_beam.runners.dataflow.ptransform_overrides import JrhReadPTransformOverride
- _PTRANSFORM_OVERRIDES = [] # type: List[PTransformOverride]
+ # Thesse overrides should be applied before the proto representation of the
+ # graph is created.
+ _PTRANSFORM_OVERRIDES = [
+ CombineValuesPTransformOverride()
Review comment:
I suppose this is fine; it's just preserving an inconsistency in Dataflow vs. everything else.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [beam] pabloem commented on issue #11335: [BEAM-9692]: Make
CombineValues portable
Posted by GitBox <gi...@apache.org>.
pabloem commented on issue #11335: [BEAM-9692]: Make CombineValues portable
URL: https://github.com/apache/beam/pull/11335#issuecomment-612225788
retest this please
----------------------------------------------------------------
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
With regards,
Apache Git Services