You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "Polber (via GitHub)" <gi...@apache.org> on 2024/04/05 01:19:00 UTC
[PR] merging non-canonical environments fails [beam]
Polber opened a new pull request, #30864:
URL: https://github.com/apache/beam/pull/30864
There are cases where environments are specified at the transform/windowing level that are not included in the top-level components' environments.
This is particularly present when running sql-backed provider on multiple inputs (i.e. when joining 2 pcollections) in Beam YAML which creates 2 default environments, but only declares one in the proto.components.environments section.
Perhaps the real issue lies in how Beam YAML providers create transforms, but this is a nice safeguard regardless.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] 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://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-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)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+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 or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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
Re: [PR] merging non-canonical environments fails [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on PR #30864:
URL: https://github.com/apache/beam/pull/30864#issuecomment-2040341056
> @robertwb After further investigation, it appears this occurs anytime that 2 subsequent ExternalTransforms are declared that use the same gradle target, and the first is given multiple inputs (PCollectionTuple). I haven't been able to repro unless the input to the first ExternalTransform is a dict of PCollections.
Note: changing the gradle target for `LogForTesting` on either example I posted will result in a successful run.
The transforms also do not have to be connected to each other:
```
service = external.JavaJarExpansionService(
subprocess_server.JavaJarServer.path_to_beam_jar(
gradle_target='sdks:java:extensions:sql:expansion-service:shadowJar',
artifact_id=None
)
)
result = {'i1': i1, 'i2': i2} | 'Sql1' >> external.ExternalTransform(
'beam:external:java:sql:v1',
external.ImplicitSchemaPayloadBuilder(
{'query': 'SELECT * FROM i1 INNER JOIN i2 ON i1.id = i2.id'}
).payload(),
service)
result2 = {'i1': i1, 'i2': i2} | 'Flatten' >> external.SchemaAwareExternalTransform(
'beam:schematransform:org.apache.beam:yaml:flatten:v1',
service, rearrange_based_on_discovery=True)
```
and it only fails if one transform is `ExternalTransform` and one is `SchemaAwareExternalTransform`. i.e. the following does not fail:
```
service = external.JavaJarExpansionService(
subprocess_server.JavaJarServer.path_to_beam_jar(
gradle_target='sdks:java:extensions:sql:expansion-service:shadowJar',
artifact_id=None
)
)
result2 = {'i1': i1, 'i2': i2} | 'Flatten' >> external.SchemaAwareExternalTransform(
'beam:schematransform:org.apache.beam:yaml:flatten:v1',
service, rearrange_based_on_discovery=True
) | 'LogForTesting' >> external.SchemaAwareExternalTransform(
'beam:schematransform:org.apache.beam:yaml:log_for_testing:v1',
service, rearrange_based_on_discovery=True)
```
--
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
Re: [PR] merging non-canonical environments fails [beam]
Posted by "robertwb (via GitHub)" <gi...@apache.org>.
robertwb merged PR #30864:
URL: https://github.com/apache/beam/pull/30864
--
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
Re: [PR] merging non-canonical environments fails [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on code in PR #30864:
URL: https://github.com/apache/beam/pull/30864#discussion_r1554320826
##########
sdks/python/apache_beam/runners/common.py:
##########
@@ -1981,28 +1981,27 @@ def env_key(env):
base_env_key(e)
for e in environments.expand_anyof_environments(env)))
- cannonical_enviornments = collections.defaultdict(list)
+ canonical_environments = collections.defaultdict(list)
for env_id, env in pipeline_proto.components.environments.items():
- cannonical_enviornments[env_key(env)].append(env_id)
+ canonical_environments[env_key(env)].append(env_id)
- if len(cannonical_enviornments) == len(
- pipeline_proto.components.environments):
+ if len(canonical_environments) == len(pipeline_proto.components.environments):
# All environments are already sufficiently distinct.
return pipeline_proto
environment_remappings = {
e: es[0]
- for es in cannonical_enviornments.values() for e in es
+ for es in canonical_environments.values() for e in es
}
if not inplace:
pipeline_proto = copy.copy(pipeline_proto)
for t in pipeline_proto.components.transforms.values():
Review Comment:
Fixed
--
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
Re: [PR] merging non-canonical environments fails [beam]
Posted by "robertwb (via GitHub)" <gi...@apache.org>.
robertwb commented on PR #30864:
URL: https://github.com/apache/beam/pull/30864#issuecomment-2039987623
That would definitely be an erroneously constructed graph. Can you give a concrete example of how this is crated?
--
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
Re: [PR] merging non-canonical environments fails [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #30864:
URL: https://github.com/apache/beam/pull/30864#issuecomment-2038577539
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
Re: [PR] merging non-canonical environments fails [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on PR #30864:
URL: https://github.com/apache/beam/pull/30864#issuecomment-2040321400
@robertwb After further investigation, it appears this occurs anytime that 2 subsequent ExternalTransforms are declared that use the same gradle target, and the first is given multiple inputs (PCollectionTuple). I haven't been able to repro unless the input to the first ExternalTransform is a dict of PCollections.
Python example:
```
import apache_beam as beam
from apache_beam.transforms import external
import logging
from apache_beam.utils import subprocess_server
logging.getLogger().setLevel(logging.INFO)
with beam.Pipeline('DirectRunner') as p:
i1 = p | "i1" >> beam.Create([beam.Row(name='john', id=1)])
i2 = p | "i2" >> beam.Create([beam.Row(name='jane', id=1)])
result = {'i1': i1, 'i2': i2} | 'Sql1' >> external.ExternalTransform(
'beam:external:java:sql:v1',
external.ImplicitSchemaPayloadBuilder(
{'query': 'SELECT * FROM i1 INNER JOIN i2 ON i1.id = i2.id'}
).payload(),
external.JavaJarExpansionService(
subprocess_server.JavaJarServer.path_to_beam_jar(
gradle_target='sdks:java:extensions:sql:expansion-service:shadowJar',
artifact_id=None
)
)) | 'LogForTesting' >> external.SchemaAwareExternalTransform(
'beam:schematransform:org.apache.beam:yaml:log_for_testing:v1',
external.JavaJarExpansionService(
subprocess_server.JavaJarServer.path_to_beam_jar(
gradle_target='sdks:java:extensions:sql:expansion-service:shadowJar',
artifact_id=None
)
), rearrange_based_on_discovery=True)
```
YAML example:
```
pipeline:
transforms:
- type: Create
name: table1
config:
elements:
- name: "john"
id: 1
- type: Create
name: table2
config:
elements:
- name: "jane"
id: 1
- type: Sql
name: Join
input:
i1: table1
i2: table2
config:
query: "SELECT * FROM i1 INNER JOIN i2 ON i1.id = i2.id"
- type: LogForTestingJava
input: Sql
# Force same gradle target variant of LogForTesting
providers:
- type: 'beamJar'
config:
gradle_target: 'sdks:java:extensions:sql:expansion-service:shadowJar'
transforms:
LogForTestingJava: 'beam:schematransform:org.apache.beam:yaml:log_for_testing:v1'
```
--
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
Re: [PR] merging non-canonical environments fails [beam]
Posted by "robertwb (via GitHub)" <gi...@apache.org>.
robertwb commented on code in PR #30864:
URL: https://github.com/apache/beam/pull/30864#discussion_r1554254408
##########
sdks/python/apache_beam/runners/common.py:
##########
@@ -1981,28 +1981,27 @@ def env_key(env):
base_env_key(e)
for e in environments.expand_anyof_environments(env)))
- cannonical_enviornments = collections.defaultdict(list)
+ canonical_environments = collections.defaultdict(list)
for env_id, env in pipeline_proto.components.environments.items():
- cannonical_enviornments[env_key(env)].append(env_id)
+ canonical_environments[env_key(env)].append(env_id)
- if len(cannonical_enviornments) == len(
- pipeline_proto.components.environments):
+ if len(canonical_environments) == len(pipeline_proto.components.environments):
# All environments are already sufficiently distinct.
return pipeline_proto
environment_remappings = {
e: es[0]
- for es in cannonical_enviornments.values() for e in es
+ for es in canonical_environments.values() for e in es
}
if not inplace:
pipeline_proto = copy.copy(pipeline_proto)
for t in pipeline_proto.components.transforms.values():
Review Comment:
This makes it a bit clearer what's going on here.
##########
sdks/python/apache_beam/runners/common.py:
##########
@@ -1981,28 +1981,27 @@ def env_key(env):
base_env_key(e)
for e in environments.expand_anyof_environments(env)))
- cannonical_enviornments = collections.defaultdict(list)
+ canonical_environments = collections.defaultdict(list)
for env_id, env in pipeline_proto.components.environments.items():
- cannonical_enviornments[env_key(env)].append(env_id)
+ canonical_environments[env_key(env)].append(env_id)
- if len(cannonical_enviornments) == len(
- pipeline_proto.components.environments):
+ if len(canonical_environments) == len(pipeline_proto.components.environments):
# All environments are already sufficiently distinct.
return pipeline_proto
environment_remappings = {
e: es[0]
- for es in cannonical_enviornments.values() for e in es
+ for es in canonical_environments.values() for e in es
}
if not inplace:
pipeline_proto = copy.copy(pipeline_proto)
for t in pipeline_proto.components.transforms.values():
Review Comment:
```suggestion
if t.environment_id not in pipeline_proto.components.environments:
# TODO(https://github.com/apache/beam/issues/30876): Remove this workaround.
continue
if t.environment_id:
t.environment_id = environment_remappings[t.environment_id]
for w in pipeline_proto.components.windowing_strategies.values():
if w.environment_id not in pipeline_proto.components.environments:
# TODO(https://github.com/apache/beam/issues/30876): Remove this workaround.
continue
if w.environment_id:
w.environment_id = environment_remappings[w.environment_id]```
--
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
Re: [PR] merging non-canonical environments fails [beam]
Posted by "Polber (via GitHub)" <gi...@apache.org>.
Polber commented on PR #30864:
URL: https://github.com/apache/beam/pull/30864#issuecomment-2038576770
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org