[BEAM-646] Get the #apply out of the DirectRunner by tgroh · Pull Request #2006 · apache/beam

tgroh · 2017-02-14T21:58:16Z

Be sure to do all of the following to help us incorporate your contribution
quickly and easily:

Make sure the PR title is formatted like:
[BEAM-<Jira issue #>] Description of pull request
Make sure tests pass via mvn clean verify. (Even better, enable
Travis-CI on your fork and ensure the whole test matrix passes).
Replace <Jira issue #> in the title with the actual Jira issue
number, if there is one.
If this contribution is large, please file an Apache
Individual Contributor License Agreement.

Use the Graph Surgery API in the DirectRunner.

The Forwarding View is currently required to ensure that the WriteView transform
is marked as the producer of the View proper. It is then replaced by the original view,
so the Pipeline writes the view to the correct location.

I'm going to revisit the producer-marking code soon, but this is safe to review.

I also missed invariant maintenance within TransformHierarchy within #1998,
so that's added here as well.

tgroh · 2017-02-14T22:00:22Z

R: @amitsela
CC/R/etc: @kennknowles

Whaboom!

coveralls · 2017-02-14T22:45:09Z

Changes Unknown when pulling a8bf815 on tgroh:surgery_ish into ** on apache:master**.

asfbot · 2017-02-14T22:45:44Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/7416/
--none--

tgroh · 2017-02-15T00:07:34Z

Good news! I removed the need for the crazy ForwardingView thingy

coveralls · 2017-02-15T00:52:25Z

Coverage decreased (-0.03%) to 69.678% when pulling 17a03e7 on tgroh:surgery_ish into 0d3389a on apache:master.

asfbot · 2017-02-15T00:52:54Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/7426/
--none--

kennknowles

Little nits. Great.

kennknowles · 2017-02-16T18:19:33Z

runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java

+          .put(
+              PTransformMatchers.stateOrTimerParDoSingle(),
+              new ParDoSingleViaMultiOverrideFactory())
+          .put(PTransformMatchers.splittableParDoMulti(), new ParDoMultiOverrideFactory())


When you have a sensitivity to ordering, comment about it it. For example "ParDoSingleViaMulti comes before ParDoMultiOverrideFactory because state and splittable are only set up to work on multi" and then later "ParDoSingleViaMulti comes after ParDoMultiOverrideFactory because the resulting expansions contains yet more ParDo singles"

Done. Most of the other things are independent of this ParDo chain, and/or are commented above.

kennknowles · 2017-02-16T18:19:55Z

runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java

  @Override
  public DirectPipelineResult run(Pipeline pipeline) {
+    for (Map.Entry<PTransformMatcher, PTransformOverrideFactory> override :
+        defaultTransformOverrides.entrySet()) {


Really nice. This is the perfect "finish line" for this API.

kennknowles · 2017-02-16T18:21:54Z

...direct-java/src/test/java/org/apache/beam/runners/direct/KeyedPValueTrackingVisitorTest.java

+                new DirectGroupAlsoByWindow<String, WindowedValue<KV<String, Integer>>>(
+                    WindowingStrategy.globalDefault(), WindowingStrategy.globalDefault()))
+            .apply(
+                ParDo.of(new ParDoMultiOverrideFactory.ToKeyedWorkItem<String, Integer>())


This is a good change. It tests only that the analysis works on the post-surgery graph containing real direct runner primitives.

kennknowles · 2017-02-16T18:23:05Z

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java

-          !isSplittable(getFn()),
-          "%s does not support Splittable DoFn",
-          input.getPipeline().getOptions().getRunner().getName());
+      // SplittableDoFn should be forbidden on the runner-side.


Are there JIRAs for the other runners to reject it? Otherwise I think this will just be silently busted.

+1 isn't SplittableDoFn to be handled by runners via SplittableParDo ?

And reject otherwise.

I've explicitly rejected SplittableDoFn in all of the existing runners.

amitsela · 2017-02-16T23:14:36Z

Besides joining Kenn's comment on removing assertions in ParDo, looks fine, but unless I'm missing something here it shouldn't be removed, unless this responsibility is passed to runners now, and if so it should still be as part of a separate PR (so still, not here).

Spark runner currently doesn't use apply but I already have a use for Pipeline#replace!

Remove Existing Outputs from Producer Map in replace. This permits replacements to produce the same PValue as the transform they are replacing, for example in CreatePCollectionView. In replace(), add the replacement input to Unexpanded Inputs.

Remove DirectRunner#apply(). This migrates the DirectRunner to work on a runner-agnostic graph.

coveralls · 2017-02-17T01:35:12Z

Coverage decreased (-0.06%) to 69.218% when pulling f965c3b on tgroh:surgery_ish into 9151676 on apache:master.

asfbot · 2017-02-17T01:35:52Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/7508/
--none--

kennknowles

Just nits on the error messages. LGTM when you fix them up.

kennknowles · 2017-02-17T01:56:50Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/TranslationUtils.java

+
+    if (signature.processElement().isSplittable()) {
+      throw new UnsupportedOperationException(
+          String.format("SparkRunner does not support SplittableDoFn: %s", doFn));


Error message nits:

For proper identifiers, interpolate when possible, as in SparkRunner.class.getSimpleName()

SplittableDoFn is not a proper identifier. It is just a splittable DoFn.

kennknowles · 2017-02-17T01:57:22Z

...s/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoBoundMultiTranslator.java


+    if (signature.processElement().isSplittable()) {
+      throw new UnsupportedOperationException(
+          String.format("ApexRunner does not support Splittable DoFn: %s", doFn));


Error message nits:

For proper identifiers, interpolate when possible, as in ApexRunner.class.getSimpleName()

Splittable shouldn't be capitalized. It is just a splittable DoFn.

kennknowles · 2017-02-17T01:57:34Z

runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoBoundTranslator.java


+    if (signature.processElement().isSplittable()) {
+      throw new UnsupportedOperationException(
+          String.format("ApexRunner does not support Splittable DoFn: %s", doFn));


Error message nits:

For proper identifiers, interpolate when possible, as in ApexRunner.class.getSimpleName()

Splittable shouldn't be capitalized. It is just a splittable DoFn.

kennknowles · 2017-02-17T01:58:14Z

.../src/main/java/org/apache/beam/runners/flink/translation/FlinkBatchTransformTranslators.java

+    DoFnSignature signature = DoFnSignatures.getSignature(doFn.getClass());
+    if (signature.processElement().isSplittable()) {
+      throw new UnsupportedOperationException(
+          String.format("FlinkRunner does not currently support Splittable DoFn: %s", doFn));


Error message nits:

For proper identifiers, interpolate when possible, as in FlinkRunner.class.getSimpleName()

Splittable shouldn't be capitalized. It is just a splittable DoFn.

kennknowles · 2017-02-17T01:58:20Z

.../main/java/org/apache/beam/runners/flink/translation/FlinkStreamingTransformTranslators.java

+    DoFnSignature signature = DoFnSignatures.getSignature(doFn.getClass());
+    if (signature.processElement().isSplittable()) {
+      throw new UnsupportedOperationException(
+          String.format("FlinkRunner does not currently support Splittable DoFn: %s", doFn));


Error message nits:

For proper identifiers, interpolate when possible, as in FlinkRunner.class.getSimpleName()

Splittable shouldn't be capitalized. It is just a splittable DoFn.

kennknowles · 2017-02-17T01:59:12Z

...dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java

    DoFnSignature signature = DoFnSignatures.getSignature(fn.getClass());
+    if (signature.processElement().isSplittable()) {
+      throw new UnsupportedOperationException(
+          String.format("FlinkRunner does not currently support splittable DoFn: %s", fn));


This is the DataflowRunner. Using identifier interpolation would have caught this :-)

The one time I don't artisinally craft a locally sourced error message I also fail to remember to fix my copypaste.

coveralls · 2017-02-17T02:42:43Z

Coverage decreased (-0.05%) to 69.226% when pulling e6eefea on tgroh:surgery_ish into 9151676 on apache:master.

asfbot · 2017-02-17T02:43:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/7512/
--none--

tgroh force-pushed the surgery_ish branch from a8bf815 to 17a03e7 Compare February 15, 2017 00:06

kennknowles requested changes Feb 16, 2017

View reviewed changes

tgroh added 2 commits February 16, 2017 16:00

Clean up TransformHierarchy#replace

b9b972c

Remove Existing Outputs from Producer Map in replace. This permits replacements to produce the same PValue as the transform they are replacing, for example in CreatePCollectionView. In replace(), add the replacement input to Unexpanded Inputs.

Use Graph Surgery in the DirectRunner

f965c3b

Remove DirectRunner#apply(). This migrates the DirectRunner to work on a runner-agnostic graph.

tgroh force-pushed the surgery_ish branch from 17a03e7 to f965c3b Compare February 17, 2017 00:21

kennknowles requested changes Feb 17, 2017

View reviewed changes

fixup! Use Graph Surgery in the DirectRunner

e6eefea

asfgit closed this in 2ca3bf6 Feb 17, 2017

Comments

Conversation

tgroh commented Feb 14, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

tgroh commented Feb 14, 2017

Uh oh!

coveralls commented Feb 14, 2017

Uh oh!

asfbot commented Feb 14, 2017

Uh oh!

tgroh commented Feb 15, 2017

Uh oh!

coveralls commented Feb 15, 2017

Uh oh!

asfbot commented Feb 15, 2017

Uh oh!

kennknowles 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!

amitsela commented Feb 16, 2017

Uh oh!

coveralls commented Feb 17, 2017

Uh oh!

asfbot commented Feb 17, 2017

Uh oh!

kennknowles 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!

coveralls commented Feb 17, 2017

Uh oh!

asfbot commented Feb 17, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

tgroh commented Feb 14, 2017 •

edited

Loading