[BEAM-2333] Go to proto and back before running a pipeline in Java DirectRunner #3334

kennknowles · 2017-06-08T23:31: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.
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.

This turn a Pipeline into a protobuf message and then decodes it prior to running it with the Java DirectRunner. Everything is still hardcoded to Java, but this proves that there is adequate information in the protobuf to correctly run a pipeline.

kennknowles · 2017-07-07T22:33:57Z

Seems to pass most tests (like all of ParDoTest etc), but mixes up composites/primitives for pipelines in PipelineTest that have composites with no primitives within them (for example returning their input directly or assembling PCollectionTuples, etc)

coveralls · 2017-07-08T00:31:12Z

Coverage increased (+0.008%) to 70.868% when pulling cd648c4 on kennknowles:Pipeline-rehydrate into 440c7d4 on apache:master.

kennknowles · 2017-07-10T22:34:25Z

The only remaining failure is DisplayData, which is simply a TODO, and then turning the hacks into better designs.

coveralls · 2017-07-11T06:01:23Z

Coverage decreased (-0.1%) to 70.751% when pulling 6cae654 on kennknowles:Pipeline-rehydrate into 0e89df3 on apache:master.

kennknowles · 2017-07-18T00:23:27Z

R: @tgroh

There are a couple of hacks left in here:

Determining if something is a primitive via set difference "outputs minus inputs"
Hardcoding the fact that ParDo is the only transform where evaluating it correctly uses additional inputs (specifically, non-additional inputs). I think the right way to do this is probably to just put the main input in the payload and have the evaluator use that.
Hitting the DisplayData methods to get them to crash, to not regress validation, while leaving them out of the proto since it isn't on the critical path. Developing the actual translation in parallel.

But it still seems like a decent milestone to perhaps checkpoint where things are with the DirectRunner, depending on how egregious you find said hacks. Suggestions welcome.

kennknowles · 2017-07-18T04:58:49Z

@tgroh @jkff I have a hunch that TextIOTest has been succeeding only because the pipeline is mutated before run. This PR first clones before run, causing stable unique name violations in TextIOReadTest.

kennknowles · 2017-07-18T05:09:24Z

Yup, hacking uniqueness works. Probably we should fix it. Filed https://issues.apache.org/jira/browse/BEAM-2632 as a starter ticket.

coveralls · 2017-07-18T06:03:14Z

Coverage decreased (-0.08%) to 70.53% when pulling 182fdd7 on kennknowles:Pipeline-rehydrate into 7c36318 on apache:master.

coveralls · 2017-07-18T16:50:52Z

Coverage decreased (-0.08%) to 70.53% when pulling 182fdd7 on kennknowles:Pipeline-rehydrate into 7c36318 on apache:master.

tgroh · 2017-07-19T01:11:56Z

sdks/java/core/src/main/java/org/apache/beam/sdk/runners/TransformHierarchy.java

@@ -274,7 +317,8 @@ public Node getCurrent() {

    private final String fullName;

-    // Nodes for sub-transforms of a composite transform.
+    // Nodes for sub-transforms of a composite transform, in shallow topological
+    // order so visiting in this order yields a recursively topological traversal.


This is not strictly required; visit will ensure that the topological order is respected.

Will it? That's new. Great! Reverted.

tgroh · 2017-07-19T01:12:49Z

sdks/java/core/src/test/java/org/apache/beam/sdk/io/TextIOReadTest.java

@@ -296,15 +297,18 @@ private static void writeToStreamAndClose(List<String> lines, OutputStream outpu
  private void assertReadingCompressedFileMatchesExpected(
      File file, CompressionType compressionType, List<String> expected) {

+    int thisUniquifier = ++uniquifier;


Can this be pulled out as an independent change?

tgroh · 2017-07-19T16:59:51Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+        RehydratedPTransform.of(
+            transformSpec.getUrn(), transformSpec.getParameter(), additionalInputs);
+
+    // HACK: A primitive is something with outputs that are not in its input


I don't know if this is as hacky as this comment claims, as we discussed.

Worth factoring into an isPrimitive method, though

Done. And reversed polarity of if accordingly.

tgroh · 2017-07-19T20:14:44Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+
+          @Override
+          public void visitPrimitiveTransform(Node node) {
+            // TODO: Include DisplayData in the proto


Can you include a stub implementation that puts in an empty DisplayData (including a DisplayDataTranslation class, so once it's implemented it's already wired in?

An explicit JIRA for "Implement Runner API Display Data" is probably worthwhile as well.

Done. Moved that to PTransformTranslation. Filed BEAM-2645.

Consider moving the TODO to that location as well.

tgroh · 2017-07-19T20:16:04Z

...on-java/src/test/java/org/apache/beam/runners/core/construction/PipelineTranslationTest.java

+                .withAllowedLateness(Duration.standardMinutes(3L)));
+    final WindowingStrategy<?, ?> windowedStrategy = windowed.getWindowingStrategy();
+    PCollection<KV<String, Long>> keyed = windowed.apply(WithKeys.<String, Long>of("foo"));
+    PCollection<KV<String, Iterable<Long>>> grouped =


Worth including a combine and something with side inputs?

Is this done?

Done now. Minor change to the verification. We discussed that the verification could be made both more thorough and less brittle by doing something other than counting, but I'd love to do that in a follow-up...

tgroh · 2017-07-19T20:18:59Z

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

+    } catch (IOException exc) {
+      throw new IllegalArgumentException(
+          String.format(
+              "%s received transform that did not contain a Java-serialized %s: %s",


That specific reason seems like an exception that should be thrown out of the getWindowFn call, as it's returning an object of type WindowFn; This should just report that it couldn't get the WindowFn out of the transform and attach the cause.

tgroh · 2017-07-19T20:20:14Z

sdks/java/core/src/main/java/org/apache/beam/sdk/runners/TransformHierarchy.java

+     * @param inputs the expanded inputs to the transform
+     * @param outputs the expanded outputs of the transform
+     */
+    private Node(


Should finalized be set to true here? all places we use it we then immediately set the bit to true.

No strong opinion. I think it is a bit uniform and readable as-is.

kennknowles

I'm also going to break out some of these sub-pieces for quick-and-easy reviews.

kennknowles · 2017-07-20T02:37:40Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+
+          @Override
+          public void visitPrimitiveTransform(Node node) {
+            // TODO: Include DisplayData in the proto


Done. Moved that to PTransformTranslation. Filed BEAM-2645.

kennknowles · 2017-07-20T02:41:18Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+        RehydratedPTransform.of(
+            transformSpec.getUrn(), transformSpec.getParameter(), additionalInputs);
+
+    // HACK: A primitive is something with outputs that are not in its input


Done. And reversed polarity of if accordingly.

coveralls · 2017-07-22T00:23:10Z

Coverage decreased (-0.08%) to 70.497% when pulling 8cca62b on kennknowles:Pipeline-rehydrate into 81c2e90 on apache:master.

kennknowles · 2017-07-22T03:28:43Z

Seem this PR lost a race with some changes to WriteFiles that cause NPEs in the display data because filenamePolicy ends up null, perhaps due to serialization. @jkff @reuvenlax does this sound familiar?

coveralls · 2017-07-22T04:40:01Z

Coverage decreased (-0.09%) to 70.491% when pulling 12a285f on kennknowles:Pipeline-rehydrate into 81c2e90 on apache:master.

kennknowles · 2017-07-22T04:59:19Z

Turns out they were just broken tests actually passing null for non-nullable values. And we don't aggressively check for null because the modern presumption is that nothing is nullable unless marked. We should probably be more defensive about user input, though.

kennknowles · 2017-07-22T05:00:40Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+    // Only ParDo really separates main from side inputs
+    Map<TupleTag<?>, PValue> additionalInputs = Collections.emptyMap();
+
+    // TODO: move this ownership into the ParDoTranslator


This bothers me a bit. I think we've discussed and the obvious plan is to also have ParDoTranslator registered by URN to do the rehydration, rather than one uniform class for RehydratedPTransform.

I think that's sensible. Doesn't need to happen here, but it's likely worth filing a JIRA to follow-up.

kennknowles · 2017-07-22T05:01:55Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+      additionalInputs = PCollectionViews.toAdditionalInputs(views);
+    }
+
+    // TODO: move this ownership into the CombineTranslator


See above comment for PAR_DO. This is another case where we want a custom rehydration to produce a RawPTransform that knows what components to register. In this case, the issue is not correct execution, but correct re-dehydration. A bit of a corner case, but should be designed right, and helpful for testing.

coveralls · 2017-07-22T05:53:26Z

Changes Unknown when pulling 5c293f8 on kennknowles:Pipeline-rehydrate into ** on apache:master**.

coveralls · 2017-07-22T06:24:29Z

Changes Unknown when pulling 5c293f8 on kennknowles:Pipeline-rehydrate into ** on apache:master**.

tgroh · 2017-07-24T16:42:39Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+
+          @Override
+          public void visitPrimitiveTransform(Node node) {
+            // TODO: Include DisplayData in the proto


Consider moving the TODO to that location as well.

tgroh · 2017-07-24T16:43:25Z

...ion-java/src/main/java/org/apache/beam/runners/core/construction/DisplayDataTranslation.java

+/** Utilities for going to/from DisplayData protos. */
+public class DisplayDataTranslation {
+  public static RunnerApi.DisplayData toProto(DisplayData displayData) {
+    // TODO https://issues.apache.org/jira/browse/BEAM-2645


We could consider putting in an "isStubImplementation=true" to signal to anyone looking at the result that it's not empty because it was actually empty.

tgroh · 2017-07-24T16:46:17Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+    // Only ParDo really separates main from side inputs
+    Map<TupleTag<?>, PValue> additionalInputs = Collections.emptyMap();
+
+    // TODO: move this ownership into the ParDoTranslator


I think that's sensible. Doesn't need to happen here, but it's likely worth filing a JIRA to follow-up.

tgroh · 2017-07-24T16:47:35Z

...on-java/src/test/java/org/apache/beam/runners/core/construction/PipelineTranslationTest.java

+                .withAllowedLateness(Duration.standardMinutes(3L)));
+    final WindowingStrategy<?, ?> windowedStrategy = windowed.getWindowingStrategy();
+    PCollection<KV<String, Long>> keyed = windowed.apply(WithKeys.<String, Long>of("foo"));
+    PCollection<KV<String, Iterable<Long>>> grouped =


Is this done?

…pipeline in Java DirectRunner Dehydrate then rehydrate Pipeline before DirectRunner.run() Add Pipeline rehydration from proto Fix tests that passed invalid input to DynamicDestinations Add stub DisplayDataTranslation

coveralls · 2017-07-25T05:57:29Z

Coverage decreased (-0.09%) to 70.503% when pulling 8ca4591 on kennknowles:Pipeline-rehydrate into 0064fb3 on apache:master.

kennknowles force-pushed the Pipeline-rehydrate branch 2 times, most recently from eff523d to 8dd7dfb Compare June 10, 2017 03:30

kennknowles force-pushed the Pipeline-rehydrate branch 3 times, most recently from 113973e to f4b11b3 Compare July 7, 2017 05:13

kennknowles force-pushed the Pipeline-rehydrate branch from c3918c6 to c80fb3f Compare July 8, 2017 03:15

kennknowles mentioned this pull request Jul 10, 2017

[BEAM-2371] Use URNs, not Java classes, in immutability enforcements #3536

Merged

4 tasks

kennknowles force-pushed the Pipeline-rehydrate branch from 94724e4 to 6cae654 Compare July 11, 2017 04:48

kennknowles force-pushed the Pipeline-rehydrate branch from 6cae654 to 22e2a02 Compare July 18, 2017 00:17

kennknowles force-pushed the Pipeline-rehydrate branch 2 times, most recently from bbeba80 to 735673b Compare July 18, 2017 04:39

kennknowles force-pushed the Pipeline-rehydrate branch from 735673b to e99b8f8 Compare July 18, 2017 05:04

kennknowles force-pushed the Pipeline-rehydrate branch from e99b8f8 to 182fdd7 Compare July 18, 2017 05:11

kennknowles mentioned this pull request Jul 18, 2017

[BEAM-2453] Perform a Multi-step combine in the DirectRunner #3579

Closed

4 tasks

tgroh requested changes Jul 19, 2017

View reviewed changes

kennknowles force-pushed the Pipeline-rehydrate branch from 9db4a17 to 834339c Compare July 20, 2017 14:29

kennknowles mentioned this pull request Jul 20, 2017

[BEAM-2333] Use RehydratedComponents for memoized rehydration #3601

Merged

6 tasks

kennknowles force-pushed the Pipeline-rehydrate branch from 834339c to 48905bd Compare July 20, 2017 14:57

kennknowles mentioned this pull request Jul 21, 2017

Uniquify application nodes in TextIOReadTest #3608

Merged

kennknowles force-pushed the Pipeline-rehydrate branch from 48905bd to a2b9260 Compare July 21, 2017 03:09

kennknowles commented Jul 21, 2017

View reviewed changes

This was referenced Jul 21, 2017

[BEAM-2371] Use dehydration-insensitive APIs in WindowEvaluatorFactory #3609

Merged

[BEAM-2371] Use dehydration-insensitive APIs in ParDoEvaluatorFactory #3610

Merged

kennknowles force-pushed the Pipeline-rehydrate branch 2 times, most recently from ec93c33 to 8cca62b Compare July 21, 2017 22:50

kennknowles force-pushed the Pipeline-rehydrate branch from 12a285f to 5c293f8 Compare July 22, 2017 04:57

kennknowles commented Jul 22, 2017

View reviewed changes

tgroh reviewed Jul 24, 2017

View reviewed changes

tgroh approved these changes Jul 24, 2017

View reviewed changes

kennknowles added 4 commits July 24, 2017 18:53

Add stub DisplayDataTranslation

efe2dc1

Fix tests that passed invalid input to DynamicDestinations

12c277f

Add Pipeline rehydration from proto

4348159

Dehydrate then rehydrate Pipeline before DirectRunner.run()

8ca4591

kennknowles force-pushed the Pipeline-rehydrate branch from fb5097f to 8ca4591 Compare July 25, 2017 01:59

asfgit merged commit 8ca4591 into apache:master Jul 25, 2017

takidau mentioned this pull request Aug 1, 2017

[BEAM-2441] move dsls/sql to sdks/java/extensions/sql #3666

Merged

kennknowles deleted the Pipeline-rehydrate branch September 29, 2017 21:26

[BEAM-2333] Go to proto and back before running a pipeline in Java DirectRunner #3334

[BEAM-2333] Go to proto and back before running a pipeline in Java DirectRunner #3334

Conversation

kennknowles commented Jun 8, 2017 • edited

kennknowles commented Jul 7, 2017

coveralls commented Jul 8, 2017

kennknowles commented Jul 10, 2017

coveralls commented Jul 11, 2017

kennknowles commented Jul 18, 2017

kennknowles commented Jul 18, 2017 • edited

kennknowles commented Jul 18, 2017

coveralls commented Jul 18, 2017

coveralls commented Jul 18, 2017

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

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

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

Choose a reason for hiding this comment

kennknowles left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

coveralls commented Jul 22, 2017

kennknowles commented Jul 22, 2017 • edited

coveralls commented Jul 22, 2017

kennknowles commented Jul 22, 2017

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

coveralls commented Jul 22, 2017

coveralls commented Jul 22, 2017

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

coveralls commented Jul 25, 2017

kennknowles commented Jun 8, 2017 •

edited

kennknowles commented Jul 18, 2017 •

edited

kennknowles commented Jul 22, 2017 •

edited