[BEAM-3566] Replace apply_* hooks in DirectRunner with PTransformOverrides by charlesccychen · Pull Request #4529 · apache/beam

charlesccychen · 2018-01-29T23:30:02Z

In the Python DirectRunner, we currently use apply_* overrides to override the operation of the default .expand() operation for certain transforms. For example, GroupByKey has a special implementation in the DirectRunner, so we use an apply_* override hook to replace the implementation of GroupByKey.expand().

However, this strategy has drawbacks. Because this override operation happens eagerly during graph construction, the pipeline graph is specialized and modified before a specific runner is bound to the pipeline's execution. This makes the pipeline graph non-portable and blocks full migration to using the Runner API pipeline representation in the DirectRunner.

By contrast, the SDK's PTransformOverride mechanism allows the expression of matchers that operate on the unspecialized graph, replacing PTransforms as necessary to produce a DirectRunner-specialized pipeline graph for execution.

We therefore want to replace these eager apply_* overrides with PTransformOverrides that operate on the completely constructed graph.

https://issues.apache.org/jira/browse/BEAM-3566

charlesccychen · 2018-02-01T23:56:58Z

R: @robertwb

This is ready for review.

robertwb

Nice, thanks! Just a couple of comments.

robertwb · 2018-02-02T20:56:28Z

sdks/python/apache_beam/io/gcp/pubsub_test.py


 import hamcrest as hc

+from apache_beam import Map


Use import apache_beam as beam, and then beam.Map for consistency with the rest of the codebase.

robertwb · 2018-02-02T20:59:50Z

sdks/python/apache_beam/io/gcp/pubsub_test.py

+    overrides = _get_transform_overrides(p.options)
+    p.replace_all(overrides)
+
+    # Note that the direct output of ReadStringsFromPubSub will be replaced


These tests seem rather brittle. Is there a better way to test this transform application than grabbing the internal source and verifying a couple of properties on it. https://beam.apache.org/documentation/pipelines/test-your-pipeline/

Ping on this comment.

I've addressed this here: #4529 (comment)

robertwb · 2018-02-02T22:14:45Z

sdks/python/apache_beam/runners/direct/direct_runner.py

-          ProcessKeyedElementsViaKeyedWorkItemsOverride()]
+
+  class CombinePerKeyOverride(PTransformOverride):
+    def get_matcher(self):


@chamikaramj I'm curious why there isn't just a is_match() method on PTransformOverride. (Or, even simpler, perhpas get_replacement_transform would return None for non-matches.)

I think this promotes developing simple (and possibly lightweight) matchers that might be shared between different overrides. We apply the matchers to every transform in the tree. get_replacement_transform() might be more heavy weight. This is inspired by the Java design.

Java (at least Java 7) didn't let you pass methods around as first class objects, thus forcing the design. Even with methods, sharding can happen (e.g. is_match = some_common_utility). We should make the common case the simple one (but probably not this PR). Could you file a bug?

Bug to replace get_matcher() by is_match() ? I think it's still good to keep is_match() and get_replacement_transform() separate to promote lightweight matching.

Yes. (The advantage of not having separate methods is that you can get rid of the whole need for a class, and just let them be callables (again, not possible in Java), but I'm fine with is_match (or matches).

Created https://issues.apache.org/jira/browse/BEAM-3614.

robertwb · 2018-02-02T22:17:08Z

sdks/python/apache_beam/runners/direct/direct_runner.py

+
+  overrides = [SplittableParDoOverride(),
+               ProcessKeyedElementsViaKeyedWorkItemsOverride(),
+               CombinePerKeyOverride(),]


Nit: only leave trailing commas if the closing is on its own line.

robertwb · 2018-02-02T22:18:46Z

sdks/python/apache_beam/runners/direct/direct_runner.py

+    from apache_beam.io.gcp import pubsub
+  except ImportError:
+    pass
+  if pubsub:


Factor this out into a method, and invoke it from the try block.

robertwb · 2018-02-02T22:23:55Z

sdks/python/apache_beam/transforms/combiners.py

-def curry_combine_fn(fn, args, kwargs):
-  if not args and not kwargs:
-    return fn
+class _CurriedFn(core.CombineFn):


This may be a performance regression.

Given that we're calling fn(..., *args, **kwargs) anyways, this shouldn't matter. Please ignore.

robertwb · 2018-02-02T22:24:18Z

sdks/python/apache_beam/transforms/combiners.py

+    return fn

-  return CurriedFn()
+  return _CurriedFn(fn, args, kwargs)


Put this in an else for symmetry.

…rides

charlesccychen

Thanks! PTAL.

charlesccychen · 2018-02-03T00:26:16Z

sdks/python/apache_beam/io/gcp/pubsub_test.py

+    overrides = _get_transform_overrides(p.options)
+    p.replace_all(overrides)
+
+    # Note that the direct output of ReadStringsFromPubSub will be replaced


robertwb wrote:
These tests seem rather brittle. Is there a better way to test this transform application than grabbing the internal source and verifying a couple of properties on it. https://beam.apache.org/documentation/pipelines/test-your-pipeline/

Thanks, and I agree, but the issue is that there really isn't anything to test.

The test here isn't a test of the transform; rather, it was (and still is) testing the behavior of replacing the transform with the correct DirectRunner replacement. The transform itself is just a wrapper with runner-specific overrides.

I changed the test name to reflect the intended target of the test.

Sorry, I meant https://beam.apache.org/contribute/ptransform-style-guide/#testing-transform-construction-and-validation

If this is about the direct runner, we should put it into the direct runner tests. Best is if we could create a mock/in memory PubSub and make sure this works end-to-end (on any runner).

But, as mentioned, fixing these existing tests should not block this PR. Please file a JIRA.

Thanks, filed https://issues.apache.org/jira/browse/BEAM-3619.

charlesccychen · 2018-02-03T00:26:16Z

sdks/python/apache_beam/runners/direct/direct_runner.py

-          ProcessKeyedElementsViaKeyedWorkItemsOverride()]
+
+  class CombinePerKeyOverride(PTransformOverride):
+    def get_matcher(self):


robertwb wrote:
Yes. (The advantage of not having separate methods is that you can get rid of the whole need for a class, and just let them be callables (again, not possible in Java), but I'm fine with is_match (or matches).

Done.

charlesccychen · 2018-02-03T00:26:16Z

sdks/python/apache_beam/io/gcp/pubsub_test.py


 import hamcrest as hc

+from apache_beam import Map


robertwb wrote:
Use import apache_beam as beam, and then beam.Map for consistency with the rest of the codebase.

Done.

charlesccychen · 2018-02-03T00:26:16Z

sdks/python/apache_beam/runners/direct/direct_runner.py

+
+  overrides = [SplittableParDoOverride(),
+               ProcessKeyedElementsViaKeyedWorkItemsOverride(),
+               CombinePerKeyOverride(),]


robertwb wrote:
Nit: only leave trailing commas if the closing is on its own line.

Done.

charlesccychen · 2018-02-03T00:26:16Z

sdks/python/apache_beam/transforms/combiners.py

-def curry_combine_fn(fn, args, kwargs):
-  if not args and not kwargs:
-    return fn
+class _CurriedFn(core.CombineFn):


robertwb wrote:
Given that we're calling fn(..., *args, **kwargs) anyways, this shouldn't matter. Please ignore.

Acknowledged.

charlesccychen · 2018-02-03T00:26:17Z

sdks/python/apache_beam/runners/direct/direct_runner.py

+    from apache_beam.io.gcp import pubsub
+  except ImportError:
+    pass
+  if pubsub:


robertwb wrote:
Factor this out into a method, and invoke it from the try block.

Done.

charlesccychen · 2018-02-03T00:26:17Z

sdks/python/apache_beam/transforms/combiners.py

+    return fn

-  return CurriedFn()
+  return _CurriedFn(fn, args, kwargs)


robertwb wrote:
Put this in an else for symmetry.

Done.

robertwb

LGTM

robertwb · 2018-02-05T23:30:52Z

sdks/python/apache_beam/io/gcp/pubsub_test.py

+    overrides = _get_transform_overrides(p.options)
+    p.replace_all(overrides)
+
+    # Note that the direct output of ReadStringsFromPubSub will be replaced


But, as mentioned, fixing these existing tests should not block this PR. Please file a JIRA.

charlesccychen force-pushed the direct-apply-hooks branch from 39be828 to 27878b6 Compare February 1, 2018 23:54

robertwb reviewed Feb 2, 2018

View reviewed changes

charlesccychen added 2 commits February 2, 2018 15:58

[BEAM-3566] Replace apply_* hooks in DirectRunner with PTransformOver…

16e5fef

…rides

Address reviewer comments

e1a96eb

charlesccychen force-pushed the direct-apply-hooks branch from 27878b6 to e1a96eb Compare February 3, 2018 00:23

charlesccychen commented Feb 3, 2018

View reviewed changes

robertwb approved these changes Feb 5, 2018

View reviewed changes

robertwb merged commit 8e2092b into apache:master Feb 5, 2018

charlesccychen mentioned this pull request Feb 6, 2018

[BEAM-3207] Create a standard location to enumerate and document URNs. #4310

Merged

6 tasks

Conversation

charlesccychen commented Jan 29, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

charlesccychen commented Feb 1, 2018

Uh oh!

robertwb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

charlesccychen left a comment

Choose a reason for hiding this comment

Uh oh!

charlesccychen Feb 3, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

robertwb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

charlesccychen commented Jan 29, 2018 •

edited

Loading

charlesccychen Feb 3, 2018 •

edited

Loading