[BEAM-9001, BEAM-6327] Ensure that all transforms (except for required runner implemented transforms) have an environment id. #11670

lukecwik · 2020-05-11T22:54:08Z

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) 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.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Gearpump	Samza
Go		---	---	---	---
Java
Python		---		---	---
XLang	---	---	---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

…d runner implemented transforms) have an environment id.

lukecwik · 2020-05-11T22:54:55Z

R: @chamikaramj @robertwb
CC: @ibzib @mxm @lostluck @CraigChambersG

lostluck

Approved for Go.

lukecwik · 2020-05-11T23:51:39Z

Run Go Flink ValidatesRunner

lukecwik · 2020-05-12T04:04:31Z

Run Go Flink ValidatesRunner

lukecwik · 2020-05-12T05:24:38Z

Run Python PreCommit

ibzib · 2020-05-12T18:34:21Z

...ain/java/org/apache/beam/runners/core/construction/graph/TrivialNativeTransformExpander.java

-public class PipelineTrimmer {
-  private static final Logger LOG = LoggerFactory.getLogger(PipelineTrimmer.class);
+/**
+ * TrivialNativeTransformExpander is used to replace transforms with known URNs with their native


What does "Trivial" mean here? Are there "Native" transforms that are not "Trivial"?

Not currently but the "trivial" is to imply that you don't need anything other then payload on the transform itself and that you don't need to inspect the transform or any of its children when constructing the native transform.

The "trivial" is meant to apply to the expander part.

chamikaramj · 2020-05-12T21:15:58Z

sdks/go/pkg/beam/core/runtime/graphx/translate.go

@@ -375,6 +375,7 @@ func (m *marshaller) addMultiEdge(edge NamedEdge) []string {
 		payload := &pipepb.WindowIntoPayload{
 			WindowFn: makeWindowFn(edge.Edge.WindowFn),
 		}
+		transformEnvID = m.addDefaultEnv()


This should be native as well, right ?

No, Window.intorequires execution of assignWindows that is part of the windowing fn (which could be a well known window fn or a custom user window fn).

It could be lifted into the runner if it understands the windowing fn but that could break fusion since Window.into may occur between ParDos and far away from GroupByKey.

Ah ok. I read this as GBK by mistake. Can we do this in a common place and skip the two known runner implemented transforms similar to other SDKs ?

chamikaramj

Thanks!

chamikaramj · 2020-05-13T16:09:58Z

...ain/java/org/apache/beam/runners/core/construction/graph/TrivialNativeTransformExpander.java

-      if (knownUrns.contains(
-          pipeline.getComponents().getTransformsOrThrow(ptransformId).getSpec().getUrn())) {
-        LOG.debug("Removing descendants of known PTransform {}" + ptransformId);
+      // Skip over previously removed transforms from the original pipeline.


Probably we should expand the comment here to describe why this is needed.

chamikaramj · 2020-05-13T16:11:10Z

sdks/go/pkg/beam/core/runtime/graphx/translate.go

@@ -375,6 +375,7 @@ func (m *marshaller) addMultiEdge(edge NamedEdge) []string {
 		payload := &pipepb.WindowIntoPayload{
 			WindowFn: makeWindowFn(edge.Edge.WindowFn),
 		}
+		transformEnvID = m.addDefaultEnv()


Ah ok. I read this as GBK by mistake. Can we do this in a common place and skip the two known runner implemented transforms similar to other SDKs ?

chamikaramj · 2020-05-13T16:17:50Z

sdks/python/apache_beam/pipeline.py

@@ -123,20 +123,16 @@ class Pipeline(object):
  should be used to designate new names
  (e.g. ``input | "label" >> my_transform``).
  """
-
-  # TODO: BEAM-9001 - set environment ID in all transforms and allow runners to
-  # override.
  @classmethod


Can we add an assert similar to PipelineValidator.java ?

Where do you suggest the assert go?

The PipelineValidator does the assertion since it is used by the Runner once the entire pipeline is constructed.

Probably we can introduce a new visitor and update runners to use that similar to following ?
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py#L557

But this is not a blocker.

lukecwik · 2020-05-13T17:59:15Z

Run Java PreCommit

lukecwik · 2020-05-13T17:59:24Z

Run Python PreCommit

lukecwik · 2020-05-13T19:23:19Z

Run Python PreCommit

lostluck · 2020-05-13T19:34:19Z

sdks/go/pkg/beam/core/runtime/graphx/translate.go

@@ -213,6 +213,7 @@ func (m *marshaller) addScopeTree(s *ScopeTree) string {
 	transform := &pipepb.PTransform{
 		UniqueName:    s.Scope.Name,
 		Subtransforms: subtransforms,
+		EnvironmentId: m.addDefaultEnv(),


So, it's now the case that Composite Transforms should have environments?

Yes.

It was to cover the case where the runner knows what the composite transform means such as a combiner and lifts it appropriately.

lukecwik · 2020-05-13T20:15:39Z

Run Python PreCommit

aaltay · 2020-05-13T22:40:03Z

Is this ready to be merged? This is the only blocker left on the release.

This change is large for a cherry pick. What would be impacted, what tests need to be run? If we cherry pick this to the release branch there will be a risk of introducing other issues.

chamikaramj

Thanks. LGTM.

chamikaramj · 2020-05-13T22:40:06Z

sdks/python/apache_beam/pipeline.py

@@ -123,20 +123,16 @@ class Pipeline(object):
  should be used to designate new names
  (e.g. ``input | "label" >> my_transform``).
  """
-
-  # TODO: BEAM-9001 - set environment ID in all transforms and allow runners to
-  # override.
  @classmethod


Probably we can introduce a new visitor and update runners to use that similar to following ?
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py#L557

But this is not a blocker.

lukecwik · 2020-05-13T22:45:07Z

Is this ready to be merged? This is the only blocker left on the release.

This change is large for a cherry pick. What would be impacted, what tests need to be run? If we cherry pick this to the release branch there will be a risk of introducing other issues.

We only would need to cherry pick the Python portion of this change which is one file. The Java/Go portion aren't important for 2.21.

aaltay · 2020-05-13T22:46:37Z

We only would need to cherry pick the Python portion of this change which is one file. The Java/Go portion aren't important for 2.21.

Thank you!

…d runner implemented transforms) have an environment id. (apache#11670) * [BEAM-9001, BEAM-6327] Ensure that all transforms (except for required runner implemented transforms) have an environment id. * fixup! Fix native transform expander to not reinsert deleted transforms. * fixup! Address chamikara's PR comments

[BEAM-9001, BEAM-6327] Ensure that all transforms (except for require…

3fbbb9e

…d runner implemented transforms) have an environment id.

probot-autolabeler bot added core flink go model python runners spark labels May 11, 2020

lostluck approved these changes May 11, 2020

View reviewed changes

fixup! Fix native transform expander to not reinsert deleted transforms.

58a4709

ibzib reviewed May 12, 2020

View reviewed changes

chamikaramj reviewed May 12, 2020

View reviewed changes

chamikaramj reviewed May 13, 2020

View reviewed changes

fixup! Address chamikara's PR comments

bb9a55b

lostluck reviewed May 13, 2020

View reviewed changes

chamikaramj reviewed May 13, 2020

View reviewed changes

lukecwik merged commit a5b2046 into apache:master May 13, 2020

ibzib mentioned this pull request May 13, 2020

[BEAM-9001, BEAM-6327] Ensure that all transforms (except for require… #11703

Merged

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-9001, BEAM-6327] Ensure that all transforms (except for required runner implemented transforms) have an environment id. #11670

[BEAM-9001, BEAM-6327] Ensure that all transforms (except for required runner implemented transforms) have an environment id. #11670

lukecwik commented May 11, 2020

lukecwik commented May 11, 2020 •

edited

lostluck left a comment

lukecwik commented May 11, 2020

lukecwik commented May 12, 2020

lukecwik commented May 12, 2020

ibzib May 12, 2020

lukecwik May 12, 2020

lukecwik May 12, 2020

chamikaramj May 12, 2020

lukecwik May 12, 2020 •

edited

chamikaramj May 13, 2020

lukecwik May 13, 2020

chamikaramj left a comment

chamikaramj May 13, 2020

lukecwik May 13, 2020

chamikaramj May 13, 2020

chamikaramj May 13, 2020

lukecwik May 13, 2020 •

edited

chamikaramj May 13, 2020

lukecwik commented May 13, 2020

lukecwik commented May 13, 2020

lukecwik commented May 13, 2020

lostluck May 13, 2020

lukecwik May 13, 2020 •

edited

lukecwik commented May 13, 2020

aaltay commented May 13, 2020

chamikaramj left a comment

chamikaramj May 13, 2020

lukecwik commented May 13, 2020

aaltay commented May 13, 2020

[BEAM-9001, BEAM-6327] Ensure that all transforms (except for required runner implemented transforms) have an environment id. #11670

[BEAM-9001, BEAM-6327] Ensure that all transforms (except for required runner implemented transforms) have an environment id. #11670

Conversation

lukecwik commented May 11, 2020

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

lukecwik commented May 11, 2020 • edited

lostluck left a comment

Choose a reason for hiding this comment

lukecwik commented May 11, 2020

lukecwik commented May 12, 2020

lukecwik commented May 12, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukecwik May 12, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukecwik May 13, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukecwik commented May 13, 2020

lukecwik commented May 13, 2020

lukecwik commented May 13, 2020

Choose a reason for hiding this comment

lukecwik May 13, 2020 • edited

Choose a reason for hiding this comment

lukecwik commented May 13, 2020

aaltay commented May 13, 2020

chamikaramj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukecwik commented May 13, 2020

aaltay commented May 13, 2020

lukecwik commented May 11, 2020 •

edited

lukecwik May 12, 2020 •

edited

lukecwik May 13, 2020 •

edited

lukecwik May 13, 2020 •

edited