[GitHub] johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric translation from Beam pipeline to IR DAG
johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric
translation from Beam pipeline to IR DAG
URL: https://github.com/apache/incubator-nemo/pull/104#discussion_r210535793
##
File path:
compiler/frontend/beam/src/main/java/edu/snu/nemo/compiler/frontend/beam/PipelineVisitor.java
##
@@ -0,0 +1,291 @@
+/*
+ * Copyright (C) 2018 Seoul National University
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.snu.nemo.compiler.frontend.beam;
+
+import edu.snu.nemo.common.dag.DAG;
+import edu.snu.nemo.common.dag.DAGBuilder;
+import edu.snu.nemo.common.dag.Edge;
+import edu.snu.nemo.common.dag.Vertex;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.runners.TransformHierarchy;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.values.PValue;
+
+import java.util.*;
+
+/**
+ * Traverses through the given Beam pipeline to construct a DAG of Beam
Transform,
+ * while preserving hierarchy of CompositeTransforms.
+ * This DAG will be later translated by {@link PipelineTranslator} into Nemo
IR DAG.
+ */
+public final class PipelineVisitor extends Pipeline.PipelineVisitor.Defaults {
+
+ private static final String TRANSFORM = "Transform-";
+ private static final String DATAFLOW = "Dataflow-";
+
+ private final Stack compositeTransformVertexStack
= new Stack<>();
+ private CompositeTransformVertex rootVertex = null;
+ private int nextIdx = 0;
+
+ @Override
+ public void visitPrimitiveTransform(final TransformHierarchy.Node node) {
+final PrimitiveTransformVertex vertex = new PrimitiveTransformVertex(node,
compositeTransformVertexStack.peek());
+compositeTransformVertexStack.peek().addVertex(vertex);
+vertex.getPValuesConsumed()
+.forEach(pValue -> {
+ final TransformVertex dst = getDestinationOfDataFlowEdge(vertex,
pValue);
+ dst.parent.addDataFlow(new
DataFlowEdge(dst.parent.getProducerOf(pValue), dst));
+});
+ }
+
+ @Override
+ public CompositeBehavior enterCompositeTransform(final
TransformHierarchy.Node node) {
+final CompositeTransformVertex vertex;
+if (compositeTransformVertexStack.isEmpty()) {
+ // There is always a top-level CompositeTransform that encompasses the
entire Beam pipeline.
+ vertex = new CompositeTransformVertex(node, null);
+} else {
+ vertex = new CompositeTransformVertex(node,
compositeTransformVertexStack.peek());
+}
+compositeTransformVertexStack.push(vertex);
+return CompositeBehavior.ENTER_TRANSFORM;
+ }
+
+ @Override
+ public void leaveCompositeTransform(final TransformHierarchy.Node node) {
+final CompositeTransformVertex vertex =
compositeTransformVertexStack.pop();
+vertex.build();
+if (compositeTransformVertexStack.isEmpty()) {
+ // The vertex is the root.
+ if (rootVertex != null) {
+throw new RuntimeException("The visitor already have traversed a Beam
pipeline. "
++ "Re-using a visitor is not allowed.");
+ }
+ rootVertex = vertex;
+} else {
+ // The CompositeTransformVertex is ready; adding it to its parent vertex.
+ compositeTransformVertexStack.peek().addVertex(vertex);
+}
+ }
+
+ /**
+ * @return A vertex representing the top-level CompositeTransform.
+ */
+ public CompositeTransformVertex getConvertedPipeline() {
+if (rootVertex == null) {
+ throw new RuntimeException("The visitor have not fully traversed through
a Beam pipeline.");
+}
+return rootVertex;
+ }
+
+ /**
+ * Represents a {@link org.apache.beam.sdk.transforms.PTransform} as a
vertex in DAG.
+ */
+ public abstract class TransformVertex extends Vertex {
Review comment:
Understood that a composite vertex can have a composite vertex as a child,
and so on like an onion.
For example:
(layer1) root composite - (layer2) child composite - (layer3) grandchild
composite - (layer4) primitives
My understanding of your comment is that with the proposed lazy execution
we're able to do cross-layer optimizations in the Beam frontend (although I
believe that's not being done at the moment). This is not possible with
directly using Beam's `TransformHierarchy.Node`, which does not provide access
to `Node#parts` that represent the children nodes of a composite node. Hence
the rationale for our own intermediary abstraction:
`TransformVertex(Prim
[GitHub] johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric translation from Beam pipeline to IR DAG
johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric
translation from Beam pipeline to IR DAG
URL: https://github.com/apache/incubator-nemo/pull/104#discussion_r210535793
##
File path:
compiler/frontend/beam/src/main/java/edu/snu/nemo/compiler/frontend/beam/PipelineVisitor.java
##
@@ -0,0 +1,291 @@
+/*
+ * Copyright (C) 2018 Seoul National University
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.snu.nemo.compiler.frontend.beam;
+
+import edu.snu.nemo.common.dag.DAG;
+import edu.snu.nemo.common.dag.DAGBuilder;
+import edu.snu.nemo.common.dag.Edge;
+import edu.snu.nemo.common.dag.Vertex;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.runners.TransformHierarchy;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.values.PValue;
+
+import java.util.*;
+
+/**
+ * Traverses through the given Beam pipeline to construct a DAG of Beam
Transform,
+ * while preserving hierarchy of CompositeTransforms.
+ * This DAG will be later translated by {@link PipelineTranslator} into Nemo
IR DAG.
+ */
+public final class PipelineVisitor extends Pipeline.PipelineVisitor.Defaults {
+
+ private static final String TRANSFORM = "Transform-";
+ private static final String DATAFLOW = "Dataflow-";
+
+ private final Stack compositeTransformVertexStack
= new Stack<>();
+ private CompositeTransformVertex rootVertex = null;
+ private int nextIdx = 0;
+
+ @Override
+ public void visitPrimitiveTransform(final TransformHierarchy.Node node) {
+final PrimitiveTransformVertex vertex = new PrimitiveTransformVertex(node,
compositeTransformVertexStack.peek());
+compositeTransformVertexStack.peek().addVertex(vertex);
+vertex.getPValuesConsumed()
+.forEach(pValue -> {
+ final TransformVertex dst = getDestinationOfDataFlowEdge(vertex,
pValue);
+ dst.parent.addDataFlow(new
DataFlowEdge(dst.parent.getProducerOf(pValue), dst));
+});
+ }
+
+ @Override
+ public CompositeBehavior enterCompositeTransform(final
TransformHierarchy.Node node) {
+final CompositeTransformVertex vertex;
+if (compositeTransformVertexStack.isEmpty()) {
+ // There is always a top-level CompositeTransform that encompasses the
entire Beam pipeline.
+ vertex = new CompositeTransformVertex(node, null);
+} else {
+ vertex = new CompositeTransformVertex(node,
compositeTransformVertexStack.peek());
+}
+compositeTransformVertexStack.push(vertex);
+return CompositeBehavior.ENTER_TRANSFORM;
+ }
+
+ @Override
+ public void leaveCompositeTransform(final TransformHierarchy.Node node) {
+final CompositeTransformVertex vertex =
compositeTransformVertexStack.pop();
+vertex.build();
+if (compositeTransformVertexStack.isEmpty()) {
+ // The vertex is the root.
+ if (rootVertex != null) {
+throw new RuntimeException("The visitor already have traversed a Beam
pipeline. "
++ "Re-using a visitor is not allowed.");
+ }
+ rootVertex = vertex;
+} else {
+ // The CompositeTransformVertex is ready; adding it to its parent vertex.
+ compositeTransformVertexStack.peek().addVertex(vertex);
+}
+ }
+
+ /**
+ * @return A vertex representing the top-level CompositeTransform.
+ */
+ public CompositeTransformVertex getConvertedPipeline() {
+if (rootVertex == null) {
+ throw new RuntimeException("The visitor have not fully traversed through
a Beam pipeline.");
+}
+return rootVertex;
+ }
+
+ /**
+ * Represents a {@link org.apache.beam.sdk.transforms.PTransform} as a
vertex in DAG.
+ */
+ public abstract class TransformVertex extends Vertex {
Review comment:
Understood that a composite vertex can have a composite vertex as a child,
and so on like an onion.
For example:
(layer1) root composite - (layer2) child composite - (layer3) grandchild
composite - (layer4) primitives
My understanding of your comment is that with the proposed lazy execution
we're able to do cross-layer optimizations in the Beam frontend (although I
believe that's not being done at the moment). This is not possible with
directly using Beam's `TransformHierarchy.Node`, which does not provide access
to `Node#parts` that represent the children nodes of a composite node. Hence
the rationale for our own intermediary abstraction:
`TransformVertex(Prim
[GitHub] johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric translation from Beam pipeline to IR DAG
johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric
translation from Beam pipeline to IR DAG
URL: https://github.com/apache/incubator-nemo/pull/104#discussion_r210523963
##
File path:
compiler/frontend/beam/src/main/java/edu/snu/nemo/compiler/frontend/beam/PipelineVisitor.java
##
@@ -0,0 +1,291 @@
+/*
+ * Copyright (C) 2018 Seoul National University
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.snu.nemo.compiler.frontend.beam;
+
+import edu.snu.nemo.common.dag.DAG;
+import edu.snu.nemo.common.dag.DAGBuilder;
+import edu.snu.nemo.common.dag.Edge;
+import edu.snu.nemo.common.dag.Vertex;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.runners.TransformHierarchy;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.values.PValue;
+
+import java.util.*;
+
+/**
+ * Traverses through the given Beam pipeline to construct a DAG of Beam
Transform,
+ * while preserving hierarchy of CompositeTransforms.
+ * This DAG will be later translated by {@link PipelineTranslator} into Nemo
IR DAG.
+ */
+public final class PipelineVisitor extends Pipeline.PipelineVisitor.Defaults {
+
+ private static final String TRANSFORM = "Transform-";
+ private static final String DATAFLOW = "Dataflow-";
+
+ private final Stack compositeTransformVertexStack
= new Stack<>();
+ private CompositeTransformVertex rootVertex = null;
Review comment:
Thanks that makes sense! I guess this corresponds to TransformHierarchy#root.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric translation from Beam pipeline to IR DAG
johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric
translation from Beam pipeline to IR DAG
URL: https://github.com/apache/incubator-nemo/pull/104#discussion_r210522713
##
File path:
compiler/frontend/beam/src/main/java/edu/snu/nemo/compiler/frontend/beam/PipelineVisitor.java
##
@@ -0,0 +1,291 @@
+/*
+ * Copyright (C) 2018 Seoul National University
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.snu.nemo.compiler.frontend.beam;
+
+import edu.snu.nemo.common.dag.DAG;
+import edu.snu.nemo.common.dag.DAGBuilder;
+import edu.snu.nemo.common.dag.Edge;
+import edu.snu.nemo.common.dag.Vertex;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.runners.TransformHierarchy;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.values.PValue;
+
+import java.util.*;
+
+/**
+ * Traverses through the given Beam pipeline to construct a DAG of Beam
Transform,
+ * while preserving hierarchy of CompositeTransforms.
+ * This DAG will be later translated by {@link PipelineTranslator} into Nemo
IR DAG.
+ */
+public final class PipelineVisitor extends Pipeline.PipelineVisitor.Defaults {
+
+ private static final String TRANSFORM = "Transform-";
+ private static final String DATAFLOW = "Dataflow-";
+
+ private final Stack compositeTransformVertexStack
= new Stack<>();
+ private CompositeTransformVertex rootVertex = null;
+ private int nextIdx = 0;
+
+ @Override
+ public void visitPrimitiveTransform(final TransformHierarchy.Node node) {
+final PrimitiveTransformVertex vertex = new PrimitiveTransformVertex(node,
compositeTransformVertexStack.peek());
+compositeTransformVertexStack.peek().addVertex(vertex);
+vertex.getPValuesConsumed()
+.forEach(pValue -> {
+ final TransformVertex dst = getDestinationOfDataFlowEdge(vertex,
pValue);
+ dst.parent.addDataFlow(new
DataFlowEdge(dst.parent.getProducerOf(pValue), dst));
+});
+ }
+
+ @Override
+ public CompositeBehavior enterCompositeTransform(final
TransformHierarchy.Node node) {
+final CompositeTransformVertex vertex;
+if (compositeTransformVertexStack.isEmpty()) {
+ // There is always a top-level CompositeTransform that encompasses the
entire Beam pipeline.
+ vertex = new CompositeTransformVertex(node, null);
+} else {
+ vertex = new CompositeTransformVertex(node,
compositeTransformVertexStack.peek());
Review comment:
Thanks! Understood that the parent-child relationship here refers to a
parent composite transform containing children transforms.
Maybe rename `parent` to `enclosingVertex` as in Beam's
TransformHierarchy.Node?
```java
// @param enclosingNode the composite node containing this node
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric translation from Beam pipeline to IR DAG
johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric
translation from Beam pipeline to IR DAG
URL: https://github.com/apache/incubator-nemo/pull/104#discussion_r210491889
##
File path:
compiler/frontend/beam/src/main/java/edu/snu/nemo/compiler/frontend/beam/PipelineVisitor.java
##
@@ -0,0 +1,291 @@
+/*
+ * Copyright (C) 2018 Seoul National University
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.snu.nemo.compiler.frontend.beam;
+
+import edu.snu.nemo.common.dag.DAG;
+import edu.snu.nemo.common.dag.DAGBuilder;
+import edu.snu.nemo.common.dag.Edge;
+import edu.snu.nemo.common.dag.Vertex;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.runners.TransformHierarchy;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.values.PValue;
+
+import java.util.*;
+
+/**
+ * Traverses through the given Beam pipeline to construct a DAG of Beam
Transform,
+ * while preserving hierarchy of CompositeTransforms.
+ * This DAG will be later translated by {@link PipelineTranslator} into Nemo
IR DAG.
+ */
+public final class PipelineVisitor extends Pipeline.PipelineVisitor.Defaults {
+
+ private static final String TRANSFORM = "Transform-";
+ private static final String DATAFLOW = "Dataflow-";
+
+ private final Stack compositeTransformVertexStack
= new Stack<>();
+ private CompositeTransformVertex rootVertex = null;
Review comment:
Are we assuming a single root? (i.e., a single source)
How does this work with multiple sources?
(Do we perhaps have tests to detect this?)
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric translation from Beam pipeline to IR DAG
johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric
translation from Beam pipeline to IR DAG
URL: https://github.com/apache/incubator-nemo/pull/104#discussion_r210500940
##
File path:
compiler/frontend/beam/src/main/java/edu/snu/nemo/compiler/frontend/beam/PipelineVisitor.java
##
@@ -0,0 +1,291 @@
+/*
+ * Copyright (C) 2018 Seoul National University
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.snu.nemo.compiler.frontend.beam;
+
+import edu.snu.nemo.common.dag.DAG;
+import edu.snu.nemo.common.dag.DAGBuilder;
+import edu.snu.nemo.common.dag.Edge;
+import edu.snu.nemo.common.dag.Vertex;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.runners.TransformHierarchy;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.values.PValue;
+
+import java.util.*;
+
+/**
+ * Traverses through the given Beam pipeline to construct a DAG of Beam
Transform,
+ * while preserving hierarchy of CompositeTransforms.
+ * This DAG will be later translated by {@link PipelineTranslator} into Nemo
IR DAG.
+ */
+public final class PipelineVisitor extends Pipeline.PipelineVisitor.Defaults {
+
+ private static final String TRANSFORM = "Transform-";
+ private static final String DATAFLOW = "Dataflow-";
+
+ private final Stack compositeTransformVertexStack
= new Stack<>();
+ private CompositeTransformVertex rootVertex = null;
+ private int nextIdx = 0;
+
+ @Override
+ public void visitPrimitiveTransform(final TransformHierarchy.Node node) {
+final PrimitiveTransformVertex vertex = new PrimitiveTransformVertex(node,
compositeTransformVertexStack.peek());
+compositeTransformVertexStack.peek().addVertex(vertex);
+vertex.getPValuesConsumed()
+.forEach(pValue -> {
+ final TransformVertex dst = getDestinationOfDataFlowEdge(vertex,
pValue);
+ dst.parent.addDataFlow(new
DataFlowEdge(dst.parent.getProducerOf(pValue), dst));
+});
+ }
+
+ @Override
+ public CompositeBehavior enterCompositeTransform(final
TransformHierarchy.Node node) {
+final CompositeTransformVertex vertex;
+if (compositeTransformVertexStack.isEmpty()) {
+ // There is always a top-level CompositeTransform that encompasses the
entire Beam pipeline.
+ vertex = new CompositeTransformVertex(node, null);
+} else {
+ vertex = new CompositeTransformVertex(node,
compositeTransformVertexStack.peek());
+}
+compositeTransformVertexStack.push(vertex);
+return CompositeBehavior.ENTER_TRANSFORM;
+ }
+
+ @Override
+ public void leaveCompositeTransform(final TransformHierarchy.Node node) {
+final CompositeTransformVertex vertex =
compositeTransformVertexStack.pop();
+vertex.build();
+if (compositeTransformVertexStack.isEmpty()) {
+ // The vertex is the root.
+ if (rootVertex != null) {
+throw new RuntimeException("The visitor already have traversed a Beam
pipeline. "
++ "Re-using a visitor is not allowed.");
+ }
+ rootVertex = vertex;
+} else {
+ // The CompositeTransformVertex is ready; adding it to its parent vertex.
+ compositeTransformVertexStack.peek().addVertex(vertex);
+}
+ }
+
+ /**
+ * @return A vertex representing the top-level CompositeTransform.
+ */
+ public CompositeTransformVertex getConvertedPipeline() {
+if (rootVertex == null) {
+ throw new RuntimeException("The visitor have not fully traversed through
a Beam pipeline.");
+}
+return rootVertex;
+ }
+
+ /**
+ * Represents a {@link org.apache.beam.sdk.transforms.PTransform} as a
vertex in DAG.
+ */
+ public abstract class TransformVertex extends Vertex {
Review comment:
My understanding of this change is that
- Before: (Beam) TransformHierarchy.Node -> (Nemo) IRVertex
- After: (Beam) TransformHierarchy.Node -> (Nemo)
TransformVertex(Primitive/Composite) -> (Nemo) IRVertex
In the previous method, we are in the 'eager mode' going directly from a
Beam primitive transform to an IRVertex without much considerations for Beam's
composite transforms. In this new approach, we are in the 'lazy mode' where we
create another full intermediary DAG, and conveniently translate composite
transforms using our `CompositeTransformVertex` abstraction.
I like how the new approach enables grouping multiple Beam primitive
transforms belonging to th
[GitHub] johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric translation from Beam pipeline to IR DAG
johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric
translation from Beam pipeline to IR DAG
URL: https://github.com/apache/incubator-nemo/pull/104#discussion_r210493130
##
File path:
compiler/frontend/beam/src/main/java/edu/snu/nemo/compiler/frontend/beam/PipelineTranslator.java
##
@@ -0,0 +1,544 @@
+/*
+ * Copyright (C) 2018 Seoul National University
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.snu.nemo.compiler.frontend.beam;
+
+import edu.snu.nemo.common.dag.DAG;
+import edu.snu.nemo.common.dag.DAGBuilder;
+import edu.snu.nemo.common.ir.edge.IREdge;
+import edu.snu.nemo.common.ir.edge.executionproperty.*;
+import edu.snu.nemo.common.ir.vertex.IRVertex;
+import edu.snu.nemo.common.ir.vertex.LoopVertex;
+import edu.snu.nemo.common.ir.vertex.OperatorVertex;
+import edu.snu.nemo.common.ir.vertex.transform.Transform;
+import edu.snu.nemo.compiler.frontend.beam.PipelineVisitor.*;
+import edu.snu.nemo.compiler.frontend.beam.coder.BeamDecoderFactory;
+import edu.snu.nemo.compiler.frontend.beam.coder.BeamEncoderFactory;
+import edu.snu.nemo.compiler.frontend.beam.source.BeamBoundedSourceVertex;
+import edu.snu.nemo.compiler.frontend.beam.transform.*;
+import org.apache.beam.sdk.coders.*;
+import org.apache.beam.sdk.io.Read;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.transforms.*;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.transforms.windowing.WindowFn;
+import org.apache.beam.sdk.values.*;
+
+import java.lang.annotation.*;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Stack;
+import java.util.function.BiFunction;
+
+/**
+ * Converts DAG of Beam pipeline to Nemo IR DAG.
+ * For a {@link PrimitiveTransformVertex}, it defines mapping to the
corresponding {@link IRVertex}.
+ * For a {@link CompositeTransformVertex}, it defines how to setup and clear
{@link TranslationContext}
+ * before start translating inner Beam transform hierarchy.
+ */
+public final class PipelineTranslator
+implements BiFunction> {
+
+ private static final PipelineTranslator INSTANCE = new PipelineTranslator();
+
+ private final Map, Method>
primitiveTransformToTranslator = new HashMap<>();
+ private final Map, Method>
compositeTransformToTranslator = new HashMap<>();
+
+ /**
+ * Static translator method.
+ * @param pipeline Top-level Beam transform hierarchy, usually given by
{@link PipelineVisitor}
+ * @param pipelineOptions {@link PipelineOptions}
+ * @return Nemo IR DAG
+ */
+ public static DAG translate(final CompositeTransformVertex
pipeline,
+final PipelineOptions
pipelineOptions) {
+return INSTANCE.apply(pipeline, pipelineOptions);
+ }
+
+ /**
+ * Creates the translator, while building a map between {@link PTransform}s
and the corresponding translators.
+ */
+ private PipelineTranslator() {
+for (final Method translator : getClass().getDeclaredMethods()) {
+ final PrimitiveTransformTranslator primitive =
translator.getAnnotation(PrimitiveTransformTranslator.class);
+ final CompositeTransformTranslator composite =
translator.getAnnotation(CompositeTransformTranslator.class);
+ if (primitive != null) {
+for (final Class transform : primitive.value()) {
+ if (primitiveTransformToTranslator.containsKey(transform)) {
+throw new RuntimeException(String.format("Translator for primitive
transform %s is"
++ "already registered: %s", transform,
primitiveTransformToTranslator.get(transform)));
+ }
+ primitiveTransformToTranslator.put(transform, translator);
+}
+ }
+ if (composite != null) {
+for (final Class transform : composite.value()) {
+ if (compositeTransformToTranslator.containsKey(transform)) {
+throw new RuntimeException(String.format("Translator for composite
transform %s is"
++ "already registered: %s", transform,
compositeTransformToTranslator.get(transform)));
+ }
+ compositeTransformToTranslator.put(transform, translator);
+}
+ }
+}
+ }
+
+ @PrimitiveTransformTranslator(Read.Bounded.class)
+ private static void boundedReadTranslator(final TranslationContext ctx,
[GitHub] johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric translation from Beam pipeline to IR DAG
johnyangk commented on a change in pull request #104: [NEMO-183] DAG-centric
translation from Beam pipeline to IR DAG
URL: https://github.com/apache/incubator-nemo/pull/104#discussion_r210491619
##
File path:
compiler/frontend/beam/src/main/java/edu/snu/nemo/compiler/frontend/beam/PipelineVisitor.java
##
@@ -0,0 +1,291 @@
+/*
+ * Copyright (C) 2018 Seoul National University
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.snu.nemo.compiler.frontend.beam;
+
+import edu.snu.nemo.common.dag.DAG;
+import edu.snu.nemo.common.dag.DAGBuilder;
+import edu.snu.nemo.common.dag.Edge;
+import edu.snu.nemo.common.dag.Vertex;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.runners.TransformHierarchy;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.View;
+import org.apache.beam.sdk.values.PValue;
+
+import java.util.*;
+
+/**
+ * Traverses through the given Beam pipeline to construct a DAG of Beam
Transform,
+ * while preserving hierarchy of CompositeTransforms.
+ * This DAG will be later translated by {@link PipelineTranslator} into Nemo
IR DAG.
+ */
+public final class PipelineVisitor extends Pipeline.PipelineVisitor.Defaults {
+
+ private static final String TRANSFORM = "Transform-";
+ private static final String DATAFLOW = "Dataflow-";
+
+ private final Stack compositeTransformVertexStack
= new Stack<>();
+ private CompositeTransformVertex rootVertex = null;
+ private int nextIdx = 0;
+
+ @Override
+ public void visitPrimitiveTransform(final TransformHierarchy.Node node) {
+final PrimitiveTransformVertex vertex = new PrimitiveTransformVertex(node,
compositeTransformVertexStack.peek());
+compositeTransformVertexStack.peek().addVertex(vertex);
+vertex.getPValuesConsumed()
+.forEach(pValue -> {
+ final TransformVertex dst = getDestinationOfDataFlowEdge(vertex,
pValue);
+ dst.parent.addDataFlow(new
DataFlowEdge(dst.parent.getProducerOf(pValue), dst));
+});
+ }
+
+ @Override
+ public CompositeBehavior enterCompositeTransform(final
TransformHierarchy.Node node) {
+final CompositeTransformVertex vertex;
+if (compositeTransformVertexStack.isEmpty()) {
+ // There is always a top-level CompositeTransform that encompasses the
entire Beam pipeline.
+ vertex = new CompositeTransformVertex(node, null);
+} else {
+ vertex = new CompositeTransformVertex(node,
compositeTransformVertexStack.peek());
Review comment:
In the following DAG
A->C
B->C
If we visit in the order of (A, B, C), then does A becomes the parent of B?
(Do we perhaps have tests to detect this?)
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
