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 2022/07/15 23:58:57 UTC
[GitHub] [beam] robertwb opened a new pull request, #22308: Deduplicate identical environments in a pipeline.
robertwb opened a new pull request, #22308:
URL: https://github.com/apache/beam/pull/22308
This is especially useful when external transforms from the same environment
are used multiple times.
------------------------
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`).
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on pull request #22308: Deduplicate identical environments in a pipeline.
Posted by GitBox <gi...@apache.org>.
robertwb commented on PR #22308:
URL: https://github.com/apache/beam/pull/22308#issuecomment-1187711824
R: @chamikaramj
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on pull request #22308: Deduplicate identical environments in a pipeline.
Posted by GitBox <gi...@apache.org>.
robertwb commented on PR #22308:
URL: https://github.com/apache/beam/pull/22308#issuecomment-1192064667
Run Python_PVR_Flink PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #22308: Deduplicate identical environments in a pipeline.
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #22308:
URL: https://github.com/apache/beam/pull/22308#issuecomment-1187718310
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb merged pull request #22308: Deduplicate identical environments in a pipeline.
Posted by GitBox <gi...@apache.org>.
robertwb merged PR #22308:
URL: https://github.com/apache/beam/pull/22308
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] commented on pull request #22308: Deduplicate identical environments in a pipeline.
Posted by GitBox <gi...@apache.org>.
codecov[bot] commented on PR #22308:
URL: https://github.com/apache/beam/pull/22308#issuecomment-1186035555
# [Codecov](https://codecov.io/gh/apache/beam/pull/22308?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#22308](https://codecov.io/gh/apache/beam/pull/22308?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (85ae056) into [master](https://codecov.io/gh/apache/beam/commit/67e6726ffeb47d2ada0122369fa230833ce0f026?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (67e6726) will **decrease** coverage by `0.05%`.
> The diff coverage is `86.66%`.
```diff
@@ Coverage Diff @@
## master #22308 +/- ##
==========================================
- Coverage 74.26% 74.20% -0.06%
==========================================
Files 702 703 +1
Lines 93019 93118 +99
==========================================
+ Hits 69076 69094 +18
- Misses 22676 22757 +81
Partials 1267 1267
```
| Flag | Coverage Δ | |
|---|---|---|
| python | `83.54% <86.66%> (-0.10%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/22308?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/python/apache\_beam/pipeline.py](https://codecov.io/gh/apache/beam/pull/22308/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcGlwZWxpbmUucHk=) | `92.28% <85.71%> (-0.15%)` | :arrow_down: |
| [...ache\_beam/runners/portability/local\_job\_service.py](https://codecov.io/gh/apache/beam/pull/22308/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9wb3J0YWJpbGl0eS9sb2NhbF9qb2Jfc2VydmljZS5weQ==) | `82.42% <100.00%> (+0.14%)` | :arrow_up: |
| [sdks/python/apache\_beam/io/source\_test\_utils.py](https://codecov.io/gh/apache/beam/pull/22308/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vc291cmNlX3Rlc3RfdXRpbHMucHk=) | `88.01% <0.00%> (-1.39%)` | :arrow_down: |
| [...eam/runners/portability/fn\_api\_runner/execution.py](https://codecov.io/gh/apache/beam/pull/22308/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9wb3J0YWJpbGl0eS9mbl9hcGlfcnVubmVyL2V4ZWN1dGlvbi5weQ==) | `92.44% <0.00%> (-0.65%)` | :arrow_down: |
| [sdks/python/apache\_beam/runners/direct/executor.py](https://codecov.io/gh/apache/beam/pull/22308/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9kaXJlY3QvZXhlY3V0b3IucHk=) | `96.46% <0.00%> (-0.55%)` | :arrow_down: |
| [...dks/python/apache\_beam/options/pipeline\_options.py](https://codecov.io/gh/apache/beam/pull/22308/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vb3B0aW9ucy9waXBlbGluZV9vcHRpb25zLnB5) | `94.34% <0.00%> (-0.42%)` | :arrow_down: |
| [.../runners/portability/fn\_api\_runner/translations.py](https://codecov.io/gh/apache/beam/pull/22308/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9wb3J0YWJpbGl0eS9mbl9hcGlfcnVubmVyL3RyYW5zbGF0aW9ucy5weQ==) | `92.65% <0.00%> (-0.34%)` | :arrow_down: |
| [sdks/python/apache\_beam/transforms/external.py](https://codecov.io/gh/apache/beam/pull/22308/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHJhbnNmb3Jtcy9leHRlcm5hbC5weQ==) | `79.73% <0.00%> (-0.11%)` | :arrow_down: |
| [sdks/python/apache\_beam/io/kafka.py](https://codecov.io/gh/apache/beam/pull/22308/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8va2Fma2EucHk=) | `80.00% <0.00%> (ø)` | |
| [sdks/python/apache\_beam/ml/inference/base.py](https://codecov.io/gh/apache/beam/pull/22308/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvaW5mZXJlbmNlL2Jhc2UucHk=) | `95.37% <0.00%> (ø)` | |
| ... and [2 more](https://codecov.io/gh/apache/beam/pull/22308/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/22308?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/22308?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [67e6726...85ae056](https://codecov.io/gh/apache/beam/pull/22308?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on pull request #22308: Deduplicate identical environments in a pipeline.
Posted by GitBox <gi...@apache.org>.
robertwb commented on PR #22308:
URL: https://github.com/apache/beam/pull/22308#issuecomment-1190891231
Run Python PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a diff in pull request #22308: Deduplicate identical environments in a pipeline.
Posted by GitBox <gi...@apache.org>.
robertwb commented on code in PR #22308:
URL: https://github.com/apache/beam/pull/22308#discussion_r926110124
##########
sdks/python/apache_beam/pipeline.py:
##########
@@ -919,11 +918,49 @@ def visit_transform(self, transform_node):
requirements=context.requirements())
proto.components.transforms[root_transform_id].unique_name = (
root_transform_id)
+ self.merge_compatible_environments(proto)
if return_context:
return proto, context # type: ignore # too complicated for now
else:
return proto
+ @staticmethod
+ def merge_compatible_environments(proto):
+ """Tries to minimize the number of distinct environments by merging
+ those that are compatible (currently defined as identical).
+
+ Mutates proto as contexts may have references to proto.components.
+ """
+ env_map = {}
+ canonical_env = {}
+ files_by_hash = {}
+ for env_id, env in proto.components.environments.items():
+ # First deduplicate any file dependencies by their hash.
+ for dep in env.dependencies:
+ if dep.type_urn == common_urns.artifact_types.FILE.urn:
+ file_payload = beam_runner_api_pb2.ArtifactFilePayload.FromString(
+ dep.type_payload)
+ if file_payload.sha256:
+ if file_payload.sha256 in files_by_hash:
+ file_payload.path = files_by_hash[file_payload.sha256]
Review Comment:
Yeah. We could probably clean that up a lot once Runner v1 goes away.
##########
sdks/python/apache_beam/pipeline.py:
##########
@@ -919,11 +918,49 @@ def visit_transform(self, transform_node):
requirements=context.requirements())
proto.components.transforms[root_transform_id].unique_name = (
root_transform_id)
+ self.merge_compatible_environments(proto)
if return_context:
return proto, context # type: ignore # too complicated for now
else:
return proto
+ @staticmethod
+ def merge_compatible_environments(proto):
+ """Tries to minimize the number of distinct environments by merging
+ those that are compatible (currently defined as identical).
+
+ Mutates proto as contexts may have references to proto.components.
+ """
+ env_map = {}
+ canonical_env = {}
+ files_by_hash = {}
+ for env_id, env in proto.components.environments.items():
+ # First deduplicate any file dependencies by their hash.
+ for dep in env.dependencies:
Review Comment:
It's not really well defined whether dependencies are ordered (e.g. the order of jar files matters for class resolution) so I'm preserving the order here just in case.
##########
sdks/python/apache_beam/pipeline_test.py:
##########
@@ -1291,6 +1291,56 @@ def CompositeTransform(pcoll):
count += 1
assert count == 2
+ def test_environments_are_deduplicated(self):
+ def file_artifact(path, hash, staged_name):
+ return beam_runner_api_pb2.ArtifactInformation(
+ type_urn=common_urns.artifact_types.FILE.urn,
+ type_payload=beam_runner_api_pb2.ArtifactFilePayload(
+ path=path, sha256=hash).SerializeToString(),
+ role_urn=common_urns.artifact_roles.STAGING_TO.urn,
+ role_payload=beam_runner_api_pb2.ArtifactStagingToRolePayload(
+ staged_name=staged_name).SerializeToString(),
+ )
+
+ proto = beam_runner_api_pb2.Pipeline(
+ components=beam_runner_api_pb2.Components(
+ transforms={
+ 'transform1': beam_runner_api_pb2.PTransform(
+ environment_id='e1'),
+ 'transform2': beam_runner_api_pb2.PTransform(
+ environment_id='e2'),
+ 'transform3': beam_runner_api_pb2.PTransform(
+ environment_id='e3'),
+ 'transform4': beam_runner_api_pb2.PTransform(
+ environment_id='e4'),
+ },
+ environments={
+ 'e1': beam_runner_api_pb2.Environment(
+ dependencies=[file_artifact('a1', 'x', 'a')]),
+ 'e2': beam_runner_api_pb2.Environment(
+ dependencies=[file_artifact('a2', 'x', 'a')]),
+ 'e3': beam_runner_api_pb2.Environment(
+ dependencies=[file_artifact('a3', 'y', 'a')]),
Review Comment:
Done.
##########
sdks/python/apache_beam/pipeline_test.py:
##########
@@ -1291,6 +1291,56 @@ def CompositeTransform(pcoll):
count += 1
assert count == 2
+ def test_environments_are_deduplicated(self):
+ def file_artifact(path, hash, staged_name):
+ return beam_runner_api_pb2.ArtifactInformation(
+ type_urn=common_urns.artifact_types.FILE.urn,
+ type_payload=beam_runner_api_pb2.ArtifactFilePayload(
+ path=path, sha256=hash).SerializeToString(),
+ role_urn=common_urns.artifact_roles.STAGING_TO.urn,
+ role_payload=beam_runner_api_pb2.ArtifactStagingToRolePayload(
+ staged_name=staged_name).SerializeToString(),
+ )
+
+ proto = beam_runner_api_pb2.Pipeline(
+ components=beam_runner_api_pb2.Components(
+ transforms={
+ 'transform1': beam_runner_api_pb2.PTransform(
+ environment_id='e1'),
+ 'transform2': beam_runner_api_pb2.PTransform(
+ environment_id='e2'),
+ 'transform3': beam_runner_api_pb2.PTransform(
+ environment_id='e3'),
+ 'transform4': beam_runner_api_pb2.PTransform(
+ environment_id='e4'),
+ },
+ environments={
+ 'e1': beam_runner_api_pb2.Environment(
+ dependencies=[file_artifact('a1', 'x', 'a')]),
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on pull request #22308: Deduplicate identical environments in a pipeline.
Posted by GitBox <gi...@apache.org>.
robertwb commented on PR #22308:
URL: https://github.com/apache/beam/pull/22308#issuecomment-1192064614
Container build failed pulling licenses.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #22308: Deduplicate identical environments in a pipeline.
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #22308:
URL: https://github.com/apache/beam/pull/22308#issuecomment-1186072895
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @ryanthompson591 for label python.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] chamikaramj commented on a diff in pull request #22308: Deduplicate identical environments in a pipeline.
Posted by GitBox <gi...@apache.org>.
chamikaramj commented on code in PR #22308:
URL: https://github.com/apache/beam/pull/22308#discussion_r925042609
##########
sdks/python/apache_beam/pipeline.py:
##########
@@ -919,11 +918,49 @@ def visit_transform(self, transform_node):
requirements=context.requirements())
proto.components.transforms[root_transform_id].unique_name = (
root_transform_id)
+ self.merge_compatible_environments(proto)
if return_context:
return proto, context # type: ignore # too complicated for now
else:
return proto
+ @staticmethod
+ def merge_compatible_environments(proto):
+ """Tries to minimize the number of distinct environments by merging
+ those that are compatible (currently defined as identical).
+
+ Mutates proto as contexts may have references to proto.components.
+ """
+ env_map = {}
+ canonical_env = {}
+ files_by_hash = {}
+ for env_id, env in proto.components.environments.items():
+ # First deduplicate any file dependencies by their hash.
+ for dep in env.dependencies:
+ if dep.type_urn == common_urns.artifact_types.FILE.urn:
+ file_payload = beam_runner_api_pb2.ArtifactFilePayload.FromString(
+ dep.type_payload)
+ if file_payload.sha256:
+ if file_payload.sha256 in files_by_hash:
+ file_payload.path = files_by_hash[file_payload.sha256]
Review Comment:
FYI, we de-duplicate artifacts again before staging: https://github.com/apache/beam/blob/ba62704a4640979763655a25fd1d603e60bc8e80/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py#L663
##########
sdks/python/apache_beam/pipeline.py:
##########
@@ -919,11 +918,49 @@ def visit_transform(self, transform_node):
requirements=context.requirements())
proto.components.transforms[root_transform_id].unique_name = (
root_transform_id)
+ self.merge_compatible_environments(proto)
if return_context:
return proto, context # type: ignore # too complicated for now
else:
return proto
+ @staticmethod
+ def merge_compatible_environments(proto):
+ """Tries to minimize the number of distinct environments by merging
+ those that are compatible (currently defined as identical).
+
+ Mutates proto as contexts may have references to proto.components.
+ """
+ env_map = {}
+ canonical_env = {}
+ files_by_hash = {}
+ for env_id, env in proto.components.environments.items():
+ # First deduplicate any file dependencies by their hash.
+ for dep in env.dependencies:
Review Comment:
Probably we should sort dependencies before de-duping ?
Otherwise different environments with the same normalized set of files might end up picking different files based on the order.
##########
sdks/python/apache_beam/pipeline_test.py:
##########
@@ -1291,6 +1291,56 @@ def CompositeTransform(pcoll):
count += 1
assert count == 2
+ def test_environments_are_deduplicated(self):
+ def file_artifact(path, hash, staged_name):
+ return beam_runner_api_pb2.ArtifactInformation(
+ type_urn=common_urns.artifact_types.FILE.urn,
+ type_payload=beam_runner_api_pb2.ArtifactFilePayload(
+ path=path, sha256=hash).SerializeToString(),
+ role_urn=common_urns.artifact_roles.STAGING_TO.urn,
+ role_payload=beam_runner_api_pb2.ArtifactStagingToRolePayload(
+ staged_name=staged_name).SerializeToString(),
+ )
+
+ proto = beam_runner_api_pb2.Pipeline(
+ components=beam_runner_api_pb2.Components(
+ transforms={
+ 'transform1': beam_runner_api_pb2.PTransform(
+ environment_id='e1'),
+ 'transform2': beam_runner_api_pb2.PTransform(
+ environment_id='e2'),
+ 'transform3': beam_runner_api_pb2.PTransform(
+ environment_id='e3'),
+ 'transform4': beam_runner_api_pb2.PTransform(
+ environment_id='e4'),
+ },
+ environments={
+ 'e1': beam_runner_api_pb2.Environment(
+ dependencies=[file_artifact('a1', 'x', 'a')]),
+ 'e2': beam_runner_api_pb2.Environment(
+ dependencies=[file_artifact('a2', 'x', 'a')]),
+ 'e3': beam_runner_api_pb2.Environment(
+ dependencies=[file_artifact('a3', 'y', 'a')]),
Review Comment:
Probably also should add a negative test for the case where artifacts are the same but other properties (for example, capabilities) are different.
##########
sdks/python/apache_beam/pipeline_test.py:
##########
@@ -1291,6 +1291,56 @@ def CompositeTransform(pcoll):
count += 1
assert count == 2
+ def test_environments_are_deduplicated(self):
+ def file_artifact(path, hash, staged_name):
+ return beam_runner_api_pb2.ArtifactInformation(
+ type_urn=common_urns.artifact_types.FILE.urn,
+ type_payload=beam_runner_api_pb2.ArtifactFilePayload(
+ path=path, sha256=hash).SerializeToString(),
+ role_urn=common_urns.artifact_roles.STAGING_TO.urn,
+ role_payload=beam_runner_api_pb2.ArtifactStagingToRolePayload(
+ staged_name=staged_name).SerializeToString(),
+ )
+
+ proto = beam_runner_api_pb2.Pipeline(
+ components=beam_runner_api_pb2.Components(
+ transforms={
+ 'transform1': beam_runner_api_pb2.PTransform(
+ environment_id='e1'),
+ 'transform2': beam_runner_api_pb2.PTransform(
+ environment_id='e2'),
+ 'transform3': beam_runner_api_pb2.PTransform(
+ environment_id='e3'),
+ 'transform4': beam_runner_api_pb2.PTransform(
+ environment_id='e4'),
+ },
+ environments={
+ 'e1': beam_runner_api_pb2.Environment(
+ dependencies=[file_artifact('a1', 'x', 'a')]),
Review Comment:
Add a test for multiple-artifacts ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on pull request #22308: Deduplicate identical environments in a pipeline.
Posted by GitBox <gi...@apache.org>.
robertwb commented on PR #22308:
URL: https://github.com/apache/beam/pull/22308#issuecomment-1190891496
Run Python_PVR_Flink PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org