SAMZA-1067: Physical execution graph and planner for fluent API #75

xinyuiscool · 2017-03-03T22:32:04Z

Initial commit for the physical graph and plan. Design is there: https://issues.apache.org/jira/secure/attachment/12856670/SAMZA-1067.0.pdf.

The commit includes:

Physical ProcessorGraph, where each processor represents a physical execution unit (e.g. a job in Yarn).
A planner does the following:
- create ProcessorGraph from StreamGraph. For this phase, the graph only contains a single node (single stage);
- figure out the partitions of intermediate topics
- create the topics

Please note currently the planner is used in the remote runner for now. Further changes/refactoring/cleanup are expected to be integrated with local runner.

…fluent API

prateekm

Thanks @xinyuiscool, looks pretty good!

I have three general categories of comments here:
Readability/Naming: related to naming, code readability or documentation.
Assertions: My current understanding of the code, please correct me if I'm wrong. Will also be helpful to document these explicitly via comments so that others can verify/understand them. Would be even better to assert these in unit tests.
Issues: General questions and concerns.

Additionally, would be good to make these classes and methods immutable wherever possible, and otherwise document mutations and state updates clearly.

prateekm · 2017-03-07T18:58:18Z

samza-core/src/main/java/org/apache/samza/processorgraph/ProcessorGraph.java

+  }
+
+  /**
+   * Validate the sinks should have outdegree being 0 and indegree greater than 0


There was some discussion about not making sendTo terminal. Let's check with @nickpan47 for more context.

I will talk to Yi about this.

I think if sendTo is not terminal, then the stream should be in both input and output streams of the StreamGraph. So it will be counted as an intermediate StreamEdge in the processorGraph. So the validation here won't be affected. But need to confirm with @nickpan47 about this.

Don't think it's necessary to make them intermediate streams. We can use the MessageStream that's the input to the sendTo as an input to the next operator too.

prateekm · 2017-03-07T23:26:32Z

samza-core/src/main/java/org/apache/samza/processorgraph/ExecutionPlanner.java

+      calculatePartitions(streamGraph, processorGraph, sysAdmins);
+
+      // create the streams
+      createStreams(processorGraph, sysAdmins);


Issue: I don't think the Planner should be creating the streams. Probably better for the ExecutionEnvironment/ApplicationRunner or a separate class to do this.

You're exactly right! The stream creation shouldn't be part of planner. I need to refactor this into a new class named as StreamManager, so it can be used in both Remote/Local runners. Do you mind if I add the refactoring into the new patch (ApplicationRunner patch)? The reason I didn't include here is because at that time I didn't have a clear picture how to break down this part for LocalRunner.

Makes sense, we can do it later.

prateekm · 2017-03-07T23:28:00Z

samza-core/src/main/java/org/apache/samza/processorgraph/ExecutionPlanner.java

+    ProcessorGraph processorGraph = new ProcessorGraph(config);
+    Set<StreamSpec> sourceStreams = new HashSet<>(streamGraph.getInStreams().keySet());
+    Set<StreamSpec> sinkStreams = new HashSet<>(streamGraph.getOutStreams().keySet());
+    Set<StreamSpec> intStreams = new HashSet<>(sourceStreams);


Issue: Isn't there a StreamGraph#getIntermediateStreams()? If not, would it be helpful for you if it did?

Current there isn't such a method. Adding it will definitely make this part of logic easier to understand I think.

prateekm · 2017-03-07T23:41:13Z

samza-core/src/main/java/org/apache/samza/processorgraph/ExecutionPlanner.java

+    sinkStreams.removeAll(intStreams);
+
+    // add sources
+    sourceStreams.forEach(spec -> processorGraph.addSource(spec, processorId));


Issue: It'd be better if ProcessorGraph was immutable, took sources, sinks and intermediate streams as parameters and did the validation itself, either immediately on construction or in a validate() method.

Since the planner decides the processors to create, and how to connect them using the intermediate streams, the construction of the graph is here instead inside the graph. The graph is immutable to classes outside the package since all the update methods are package private. I couldn't think of a better way to make it immutable. Any ideas?

I see. Makes sense to keep it here for when the planner is doing more work and deciding which Processors to create.

prateekm · 2017-03-07T23:42:12Z

samza-core/src/main/java/org/apache/samza/processorgraph/ExecutionPlanner.java

+
+  /**
+   * Figure out the number of partitions of intermediate streams
+   * Package private for testing


Readability: I like the convention of adding /* package private */ before the method declaration. E.g.,

/* package private */ void calculatePartitions() {

Applies to other classes as well.

Fixed. Thanks for the recommendation.

prateekm · 2017-03-08T20:26:20Z

samza-core/src/main/java/org/apache/samza/processorgraph/ProcessorGraph.java

+      }
+
+      if (sortedNodes.size() < pnodes.size()) {
+        // The remaining nodes have circles


Readability: s/circles/cycles

prateekm · 2017-03-08T20:31:36Z

samza-core/src/main/java/org/apache/samza/processorgraph/ProcessorGraph.java

+        sortedNodes.add(node);
+        node.getOutEdges().stream().flatMap(edge -> edge.getTargetNodes().stream()).forEach(n -> {
+            String nid = n.getId();
+            Long degree = indegree.get(nid) - 1;


Readability: Will be good to add a comment about what this whole block is trying to achieve.

Also, why decrement indegree? Are we trying to sort by indegree? If so, will be clearer to keep an incrementing counter and subtract that instead of mutating the indegree map directly (since then it's not really indegree anymore.

I added comments about what this algo does. Basically it implemented an indegree-based approach similar to this: http://www.geeksforgeeks.org/topological-sorting-indegree-based-solution/. It's different from the original approach in that it won't need to change the graph itself.

Yup, the logic looks good. I meant to say that it'll be clearer to keep a running count and subtract that from the indegree, rather than mutating the indegree map directly.

Still not very clear about it. Let chat.

prateekm · 2017-03-08T22:18:36Z

samza-core/src/main/java/org/apache/samza/processorgraph/ProcessorGraph.java

+
+
+  /**
+   * Validate the graph


Readability: Document what validation entails.

Added the docs.

prateekm · 2017-03-08T22:23:00Z

samza-core/src/main/java/org/apache/samza/util/ConfigInheritance.java

+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ConfigInheritance {


Issue: Looks like this is currently only used once. Is this going to be re-used in multiple places?

Readability: Needs documentation for what this is for and what it does. Also, the name doesn't make sense by itself, maybe 'ConfigInheritance[Helper|Util]'?

Right. It was used in multiple places before but the code got refactored and now only one place needs. it. I get rid of this class.

prateekm · 2017-03-08T22:36:41Z

samza-core/src/main/java/org/apache/samza/processorgraph/ExecutionPlanner.java

+ * under the License.
+ */
+
+package org.apache.samza.processorgraph;


Naming: Probably org.apache.samza.[execution|runner] or something similar would be more appropriate.

The reason I put it here is that this class can access some package private functions of ProcessorGraph so the graph cannot be changed after it's created. Since it's tightly coupled with ProcessorGraph, I will keep this here for now.

Yeah, keeping them together makes sense. What I meant is that maybe "execution" or "runner" makes more sense as a top level package name than "processorgraph".

right, execution seems to be a better package name. Thanks for the suggestion!

prateekm

Minor documentation related comments and nitpicks, otherwise looks good to me. Thanks for the updates!

prateekm · 2017-03-13T17:50:02Z

samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpecs.java

@@ -149,6 +149,7 @@ public void init(Config config, TaskContext context) {
   * @param sinkFn  the sink function
   * @param stream  the {@link OutputStream} where the message is sent to
   * @param <M>  type of input message
+   * @param opId operator ID


Documentation: Should be before type parameters (the <M>)

prateekm · 2017-03-13T17:50:59Z

samza-core/src/main/java/org/apache/samza/processorgraph/ExecutionPlanner.java

+ * under the License.
+ */
+
+package org.apache.samza.processorgraph;


Yeah, keeping them together makes sense. What I meant is that maybe "execution" or "runner" makes more sense as a top level package name than "processorgraph".

prateekm · 2017-03-13T17:51:26Z

samza-core/src/main/java/org/apache/samza/processorgraph/ExecutionPlanner.java

+      calculatePartitions(streamGraph, processorGraph, sysAdmins);
+
+      // create the streams
+      createStreams(processorGraph, sysAdmins);


Makes sense, we can do it later.

prateekm · 2017-03-13T17:51:45Z

samza-core/src/main/java/org/apache/samza/processorgraph/ExecutionPlanner.java

+  /**
+   * Create the physical graph from StreamGraph
+   */
+  /* package private for testing */ ProcessorGraph createProcessorGraph(StreamGraph streamGraph) {


Documentation: Don't need "for testing"

prateekm · 2017-03-13T18:10:33Z

samza-core/src/main/java/org/apache/samza/processorgraph/ExecutionPlanner.java

+    sinkStreams.removeAll(intStreams);
+
+    // add sources
+    sourceStreams.forEach(spec -> processorGraph.addSource(spec, processorId));


I see. Makes sense to keep it here for when the planner is doing more work and deciding which Processors to create.

prateekm · 2017-03-13T21:01:05Z

samza-core/src/main/java/org/apache/samza/processorgraph/ProcessorGraph.java

+ * It contains the topology of execution processors connected with source/sink/intermediate streams.
+ * High level APIs are transformed into ProcessorGraph for planing, validation and execution.
+ * Source/sink streams are external streams while intermediate streams are created and managed by Samza.
+ * Note that intermediate streams can be both the input and output of a ProcessorNode in ProcessorGraph.


Documentation: s/can be/is?

prateekm · 2017-03-13T22:05:37Z

samza-core/src/main/java/org/apache/samza/processorgraph/ProcessorGraph.java

+
+  /**
+   * Returns the processors to be executed in the topological order
+   * @return list of {@link ProcessorNode}


Documentation: "unmodifiable list/set of ", here and other methods.

prateekm · 2017-03-13T22:08:19Z

samza-core/src/main/java/org/apache/samza/processorgraph/ProcessorGraph.java

+  }
+
+  /**
+   * Validate the sinks should have outdegree being 0 and indegree greater than 0


Don't think it's necessary to make them intermediate streams. We can use the MessageStream that's the input to the sendTo as an input to the next operator too.

prateekm · 2017-03-13T22:10:24Z

samza-core/src/main/java/org/apache/samza/processorgraph/ProcessorGraph.java

+        sortedNodes.add(node);
+        node.getOutEdges().stream().flatMap(edge -> edge.getTargetNodes().stream()).forEach(n -> {
+            String nid = n.getId();
+            Long degree = indegree.get(nid) - 1;


Yup, the logic looks good. I meant to say that it'll be clearer to keep a running count and subtract that from the indegree, rather than mutating the indegree map directly.

prateekm · 2017-03-13T22:11:17Z

samza-core/src/main/java/org/apache/samza/processorgraph/ProcessorNode.java

+    Config scopedConfig = fullConfig.subset(configPrefix);
+
+    Config[] configPrecedence;
+    if (INHERIT_ROOT_CONFIGS) {


Can probably remove this and the field since this method isn't shared.

This was a left-over from the early pipeline code. Let me remove it. Thanks for catching it.

vjagadish1989 · 2017-03-15T17:40:26Z

samza-core/src/main/java/org/apache/samza/execution/ExecutionPlanner.java

+   */
+  /* package private */ ProcessorGraph createProcessorGraph(StreamGraph streamGraph) {
+    // For this phase, we are going to create a processor for the whole dag
+    String processorId = config.get(JobConfig.JOB_NAME()); // only one processor, use the job name


Don't we need some combination of Job name and Job Id here? I thought the tuple [Job name, Job Id] identifies a job uniquely?

Good question. I use job name for processorId since it's easier to generate multi-stage job names by appending indexes afterwards, like test-1, test-2. I don't expect multiple instances of the same job exists in the same application so I didn't include both. I agree if we want to support this, we need to have a tuple data structure like ProcessorUid which has both inside.

vjagadish1989 · 2017-03-15T18:46:12Z

samza-core/src/main/java/org/apache/samza/execution/ExecutionPlanner.java

+      // set the partitions of a stream to its StreamEdge
+      streamToMetadata.forEach((stream, data) -> {
+          int partitions = data.getSystemStreamPartitionMetadata().size();
+          streamToStreamEdge.get(stream).setPartitions(partitions);


nit: setPartitionCount(partitions)

vjagadish1989 · 2017-03-15T20:53:44Z

samza-core/src/main/java/org/apache/samza/execution/StreamEdge.java

+  private final StreamSpec streamSpec;
+  private final List<ProcessorNode> sourceNodes = new ArrayList<>();
+  private final List<ProcessorNode> targetNodes = new ArrayList<>();
+  private final Config config;


Appears that the config is no longer accessed (for instance, config does not have getters). Is this for a future use-case?

StreamEdge in the graph does not need the config object when the streamSpec and the enclosing ProcessorGraph already have the reference to the config right?

The constructor can instead be:
StreamEdge(StreamSpec spec)?

Good catch! Removed the config in this class.

vjagadish1989 · 2017-03-15T21:00:10Z

samza-core/src/main/java/org/apache/samza/execution/StreamEdge.java

+ */
+public class StreamEdge {
+  private final StreamSpec streamSpec;
+  private final List<ProcessorNode> sourceNodes = new ArrayList<>();


Currently, there is only one ProcessorNode created right for the entire application?

That's right!

vjagadish1989 · 2017-03-15T21:53:32Z

samza-core/src/main/java/org/apache/samza/execution/StreamEdge.java

+    return targetNodes;
+  }
+
+  int getPartitions() {


Document that we return -1 if not assigned yet. Even better, return a final constant and use it to consistently to check in all places. A couple of places in ExecutionPlanner check for -1.

Sounds great. I made a final constant for -1 and used it in other places.

vjagadish1989 · 2017-03-15T22:31:00Z

samza-core/src/main/java/org/apache/samza/execution/ExecutionPlanner.java

+      }
+      // assign the partition count
+      for (StreamEdge edge : joinSpecToStreamEdges.get(join)) {
+        if (edge.getPartitions() <= 0) {


What conditions would cause an edge.getPartitions() == -1 at this point? Will be super-helpful to document this.

vjagadish1989

Thanks for the contributions!

looks good. approved.

nickpan47 · 2017-03-14T05:48:27Z

samza-core/src/main/java/org/apache/samza/execution/ExecutionPlanner.java

+
+
+/**
+ * The ExecutionPlanner creates the physical execution graph for the StreamGraph, and


nit: please use {@link xxx} for the class names in Javadoc.

nickpan47 · 2017-03-14T05:57:14Z

samza-core/src/main/java/org/apache/samza/execution/ExecutionPlanner.java

+
+    if (!processorGraph.getIntermediateStreams().isEmpty()) {
+      // figure out the partitions for internal streams
+      calculatePartitions(streamGraph, processorGraph, sysAdmins);


minor comment: we can wrap streamGraph as a constructor parameter into ProcessorGraph.

Initial commit for the physical graph and plan. Design is there: https://issues.apache.org/jira/secure/attachment/12856670/SAMZA-1067.0.pdf. The commit includes: 1) Physical ProcessorGraph, where each processor represents a physical execution unit (e.g. a job in Yarn). 2) A planner does the following: - create ProcessorGraph from StreamGraph. For this phase, the graph only contains a single node (single stage); - figure out the partitions of intermediate topics - create the topics Please note currently the planner is used in the remote runner for now. Further changes/refactoring/cleanup are expected to be integrated with local runner. Author: Xinyu Liu <xiliu@xiliu-ld.linkedin.biz> Reviewers: Jagadish Venkatraman <jvenkatraman@linkedin.com> Closes apache#75 from xinyuiscool/SAMZA-1067

SAMZA-1067: Physical execution plan for logic expressions written in …

ef6548f

…fluent API

prateekm reviewed Mar 8, 2017

View reviewed changes

Xinyu Liu added 5 commits March 9, 2017 17:52

SAMZA-1067: Physical execution graph and planner for fluent API

ebae986

Merge branch 'master' into SAMZA-1067

d00031e

Add unit tests for SAMZA-1067

ee77211

Merge branch 'master' into SAMZA-1067

1c48d6c

Updates based on merging from master

ca71616

prateekm approved these changes Mar 13, 2017

View reviewed changes

Xinyu Liu added 2 commits March 13, 2017 15:52

Some refactoring based on pr comments

2cd467b

Refactor test case package name

136aab7

vjagadish1989 reviewed Mar 15, 2017

View reviewed changes

vjagadish1989 approved these changes Mar 16, 2017

View reviewed changes

Update based on Jagadish's review

dfa52bf

asfgit closed this in 1676eab Mar 16, 2017

nickpan47 reviewed Mar 16, 2017

View reviewed changes



		/**
		* The ExecutionPlanner creates the physical execution graph for the StreamGraph, and

SAMZA-1067: Physical execution graph and planner for fluent API #75

SAMZA-1067: Physical execution graph and planner for fluent API #75

Conversation

xinyuiscool commented Mar 3, 2017 • edited

prateekm left a 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

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

prateekm left a 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

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

vjagadish1989 left a comment • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xinyuiscool commented Mar 3, 2017 •

edited

vjagadish1989 left a comment •

edited