[BEAM-3565] Add FusedPipeline#toPipeline by tgroh · Pull Request #4777 · apache/beam

tgroh · 2018-03-01T02:34:37Z

The FusedPipeline is the physical plan corresponding with an original, logical Pipeline. Converting it back into a proto allows that plan to be manipulated with existing libraries, and for runners to interact with that plan for runner-specific implementation reasons.

Follow this checklist to help us incorporate your contribution quickly and easily:

tgroh · 2018-03-01T02:34:44Z

@bsidhom

lukecwik · 2018-03-06T19:32:16Z

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

   */
-  public Components asComponents(Components base) {
-    Builder newComponents = base.toBuilder().clearTransforms();
+  private Map<String, PTransform> getTopLevelTransforms(Components base) {


getRunnerExecutedPrimitiveTransforms?

If we go with my suggestion of treating executablestage as a primitive, then this can be called getPrimitiveTransforms or something along those lines. Even asComponents would still make sense.

getExecutableTransforms is what I am calling this now.

All of the transforms are primitives, but there are non-executable transforms (within a stage) that get merged into the components as well.

lukecwik · 2018-03-06T19:33:17Z

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

+
  /**
   * Return a {@link Components} like the {@code base} components, but with the only transforms
   * equal to this fused pipeline.


but with the only transforms equal to this fused pipeline. -> but with the set of transforms required to be executed by a runner.

lukecwik · 2018-03-06T19:39:27Z

...on-java/src/main/java/org/apache/beam/runners/core/construction/graph/QueryablePipeline.java

+   * PCollections} are not yet modelled by {@link QueryablePipeline}, so the input {@link
+   * Components} should be treatable as though each node is a primitive.
+   */
+  static QueryablePipeline forComponents(Components components) {


It would make sense to make ExecutableStage a primitive and have its payload be the subtransforms, coders, pcollections, ...

This will address some of the coder issues and the duality where you want the runner based pipeline representation to differ from the SDK based pipeline segment which will now be completely embedded inside the executable stage.

#4844 performs a lot of this change, without embedding the components within the stage; it will still require us to create a partial components, but should ultimately cause this construction to go through the same path as the original queryablePipeline

lukecwik · 2018-03-14T23:28:03Z

Please address all the comments.

tgroh · 2018-03-21T00:01:06Z

PTAL

lukecwik · 2018-03-21T21:02:19Z

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

-   * <p>The only composites will be the stages returned by {@link #getFusedStages()}.
+   * <p>The transforms that are present in the returned map are the union of the results of {@link
+   * #getRunnerExecutedTransforms()} and {@link #getFusedStages()}, where each {@link
+   * ExecutableStage}.


where each executable stage?

lukecwik · 2018-03-21T21:06:06Z

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

  /**
-   * Return a {@link Components} like the {@code base} components, but with the only transforms
-   * equal to this fused pipeline.
+   * Return a {@link Components} like the {@code base} components, but with the set of transforms to


This comment seems like it should apply toPipeline and not the private method getExecutableStages

Also, Return a {@link Pipeline}

This comment still seems out of date since it refers to base components.

lukecwik · 2018-03-21T21:15:46Z

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

+  /** The {@link PTransform PTransforms} that a runner is responsible for executing. */
  public abstract Set<PTransformNode> getRunnerExecutedTransforms();

+  public RunnerApi.Pipeline toPipeline(Components initialComponents) {


We were given the Pipeline when we constructed the FusedStage via GreedyPipelineFuser, why do we need initialComponents to be passed in again here?

Made into a property.

lukecwik · 2018-03-21T21:28:52Z

...on-java/src/main/java/org/apache/beam/runners/core/construction/graph/QueryablePipeline.java

+   * Get the PCollections which are not consumed by any {@link PTransformNode} in this {@link
+   * QueryablePipeline}.
+   */
+  private Set<PCollectionNode> getLeafPCollections() {


This isn't used anywhere, what is the intent of adding it right now?

Pulled in accidentally, I think.

lukecwik · 2018-03-21T21:32:42Z

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

+        fused.getFusedStages().size() == 2,
+        "Unexpected number of fused stages %s",
+        fused.getFusedStages());
+    RunnerApi.Pipeline fusedProto = fused.toPipeline(protoPipeline.getComponents());


fusedProto -> fusedProtoPipeline?

lukecwik · 2018-03-21T21:39:25Z

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

+              PCollection.class.getSimpleName(), fusedProto.getRootTransformIds(i)),
+          producedPCollections,
+          hasItems(rootTransform.getInputsMap().values().toArray(new String[0])));
+      for (String consumed : consumedPCollections) {


Don't all roots have zero consumed PCollections so this is just checking that no two roots produce the same PCollection.

the Pipeline root transform ID's is really the top-level transforms (transforms not contained in any other transform)

I see, not DAG roots.

lukecwik

Please address minor comment nits.

lukecwik · 2018-03-21T23:00:33Z

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

  /**
-   * Return a {@link Components} like the {@code base} components, but with the only transforms
-   * equal to this fused pipeline.
+   * Return a {@link Components} like the {@code base} components, but with the set of transforms to


This comment still seems out of date since it refers to base components.

lukecwik · 2018-03-21T23:04:26Z

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

+   * PTransforms} present in the original Pipeline that this {@link FusedPipeline} was created from,
+   * plus all of the {@link ExecutableStage ExecutableStages} contained within this {@link
+   * FusedPipeline}. The Root Transform IDs will contain all of the runner executed transforms and
+   * all of the ExecutableStages contained within the Pipeline.


ExecutableStages -> {@link ExecutableStage executable stages}

lukecwik · 2018-03-21T23:08:01Z

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

+   * <p>The {@link Components} of the returned pipeline will contain all of the {@link PTransform
+   * PTransforms} present in the original Pipeline that this {@link FusedPipeline} was created from,
+   * plus all of the {@link ExecutableStage ExecutableStages} contained within this {@link
+   * FusedPipeline}. The Root Transform IDs will contain all of the runner executed transforms and


The upper casing on Root Transform IDs is strange, would you rather link the Pipeline root transform ids method?

lukecwik · 2018-03-21T23:18:01Z

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

+              PCollection.class.getSimpleName(), fusedProto.getRootTransformIds(i)),
+          producedPCollections,
+          hasItems(rootTransform.getInputsMap().values().toArray(new String[0])));
+      for (String consumed : consumedPCollections) {


I see, not DAG roots.

Any pipeline node should be identifiable within the components the pipeline was built from, and this allows cleaner comparisons based on the id.

Given a Pipeline Components, this constructs the fused representation, including a shallow topologically ordered set of root transforms.

tgroh · 2018-03-21T23:46:29Z

Done to all

tgroh force-pushed the fused_pipeline_network_topo branch 3 times, most recently from 40cd510 to 025f9f7 Compare March 6, 2018 18:22

lukecwik requested changes Mar 6, 2018

View reviewed changes

tgroh force-pushed the fused_pipeline_network_topo branch from 025f9f7 to 2641890 Compare March 13, 2018 23:36

tgroh force-pushed the fused_pipeline_network_topo branch 3 times, most recently from e2ba80f to 361211e Compare March 20, 2018 21:26

tgroh force-pushed the fused_pipeline_network_topo branch from 361211e to 6e82d2e Compare March 21, 2018 18:13

lukecwik requested changes Mar 21, 2018

View reviewed changes

lukecwik approved these changes Mar 21, 2018

View reviewed changes

tgroh added 2 commits March 21, 2018 16:46

Move getId to the top-level pipeline node

043c615

Any pipeline node should be identifiable within the components the pipeline was built from, and this allows cleaner comparisons based on the id.

Add FusedPipeline#toPipeline

d254651

Given a Pipeline Components, this constructs the fused representation, including a shallow topologically ordered set of root transforms.

tgroh force-pushed the fused_pipeline_network_topo branch from e74aa23 to d254651 Compare March 21, 2018 23:46

bsidhom mentioned this pull request Mar 22, 2018

Fix compile bsidhom/beam#26

Merged

tgroh merged commit 3978b21 into apache:master Mar 23, 2018

Conversation

tgroh commented Mar 1, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

tgroh commented Mar 1, 2018

Uh oh!

lukecwik Mar 6, 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!

tgroh Mar 13, 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!

lukecwik commented Mar 14, 2018

Uh oh!

tgroh commented Mar 21, 2018

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!

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

tgroh commented Mar 21, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

tgroh commented Mar 1, 2018 •

edited

Loading

lukecwik Mar 6, 2018 •

edited

Loading

tgroh Mar 13, 2018 •

edited

Loading