[NEMO-338] SkewSamplingPass #193

johnyangk · 2019-02-13T03:32:37Z

JIRA: NEMO-338: SkewSamplingPass

Major changes:

SamplingSkewReshapingPass: Inserts SkewSampling and MessageBarrier vertices
SamplingVertex: Instantiated with (originalVertex, desiredSampleRate)
IRDAG: Automatically inserts IREdges from/to SamplingVertex objects, similar to other insert() methods
PhysicalPlanGenerator: Handles SamplingVertex objects appropriately
Stage: Uses getTaskIndices(), returns a subset of tasks if consists of SamplingVertex objects, to determine the tasks to execute

Minor changes to note:

Refactors other insert() methods to share code as much as possible

Tests for the changes:

PerKeyMedianITCase#testLargeShuffleSamplingSkew (combines large shuffle + skew handling optimizations)

Other comments:

Sanha(@sanha) wrote the original code. I refactored the code and added comments to create this PR.

Closes #193

johnyangk · 2019-02-13T03:33:17Z

@sanha Can you check if I missed out anything? Other reviewers are also welcome to take a look, of course.

sanha

Thanks for the work @johnyangk! Here's my first review.

sanha · 2019-02-14T10:58:55Z

common/src/main/java/org/apache/nemo/common/Util.java

+    if (edgeToClone.getPropertySnapshot().containsKey(EncoderProperty.class)) {
+      clone.setProperty(edgeToClone.getPropertySnapshot().get(EncoderProperty.class));
+    } else {
+      clone.setProperty(EncoderProperty.of(edgeToClone.getPropertyValue(EncoderProperty.class).get()));


Please use orElseThrow or something like that instead of get.

sanha · 2019-02-14T10:59:07Z

common/src/main/java/org/apache/nemo/common/Util.java

+    if (edgeToClone.getPropertySnapshot().containsKey(DecoderProperty.class)) {
+      clone.setProperty(edgeToClone.getPropertySnapshot().get(DecoderProperty.class));
+    } else {
+      clone.setProperty(DecoderProperty.of(edgeToClone.getPropertyValue(DecoderProperty.class).get()));


Please use orElseThrow or something like that instead of get.

sanha · 2019-02-15T10:02:28Z

common/src/main/java/org/apache/nemo/common/Util.java

+    });
+
+    edgeToClone.getPropertyValue(PartitionerProperty.class).ifPresent(p -> {
+      if (p.right() == PartitionerProperty.NUM_EQUAL_TO_DST_PARALLELISM) {


Is this code needed? The result will be the same without this if clause.

It is needed. Please see: https://github.com/apache/incubator-nemo/blob/master/common/src/main/java/org/apache/nemo/common/ir/edge/executionproperty/PartitionerProperty.java#L66

sanha · 2019-02-19T09:46:41Z

common/src/main/java/org/apache/nemo/common/Util.java

+   * @param dst vertex.
+   * @return the control edge.
+   */
+  public static IREdge createControlEdge(final IRVertex src, final IRVertex dst) {


Please add some method level comment for this method.

sanha · 2019-02-19T09:57:35Z

common/src/main/java/org/apache/nemo/common/ir/IRDAG.java

   * @param messageBarrierVertex to insert.
   * @param messageAggregatorVertex to insert.
   * @param mbvOutputEncoder to use.
   * @param mbvOutputDecoder to use.
   * @param edgesToGetStatisticsOf to examine.
+   * @param edgesToOptimize to optimize.
   */
  public void insert(final MessageBarrierVertex messageBarrierVertex,


The inner BiFunction (messageFunction) of the messageBarrierVertex is only used in this method.
Why don't we receive the BiFunction instead of MessageBarrierVertex?

Created a JIRA for this: https://issues.apache.org/jira/browse/NEMO-341

sanha · 2019-02-19T10:13:56Z

common/src/main/java/org/apache/nemo/common/ir/IRDAG.java

+   * @param samplingVertices to insert.
+   * @param executeAfterSamplingVertices that must be executed after samplingVertices.
+   */
+  public void insert(final Set<SamplingVertex> samplingVertices,


Why don't we get the Set of the original vertices to sample instead of already sampled vertices?
If we receive the sampled vertices, opt pass builders can give some already connected sampling vertices.

Created a JIRA for this: https://issues.apache.org/jira/browse/NEMO-341
I also added the assertNonExistence() checker in this method and the other insert() methods.

sanha · 2019-02-19T10:25:28Z

...org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SamplingSkewReshapingPass.java

+ *
+ * Then, this pass will produce something like:
+ * P1' - P1 - P2
+ *          - P2' - P2


Why do we need to sample P2?

sanha · 2019-02-19T10:28:47Z

...org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SamplingSkewReshapingPass.java

+            .orElseThrow(() -> new IllegalStateException());
+          final IREdge clonedShuffleEdge = rightBeforeShuffle.getCloneOfOriginalEdge(e);
+
+          final KeyExtractor keyExtractor = e.getPropertyValue(KeyExtractorProperty.class).get();


We need to add control edges from the message aggregation vertex to the partitionSources instead of the vertex that receives the original shuffle edge.

For the example DAG that is partitioned into two sub-DAGs as follows: P1 -(shuffle)- P2,
the expected outcome looks like P1' -(o2o)- MCV -(shuffle)- MAV -(control)- P1 -(shuffle)- p2.

This is because that we must optimze the partitioning way of the target shuffle edge before the execution of P1.

Also, the P1' and message collection vertex must be in a single stage. If not, the whole intermediate data will be duplicated.

Thanks for bringing this up!

(1) Regarding pipelining MessageBarrierVertex within a single stage with parent sampling vertices:

I've changed the semantics of insert() to use SamplingVertex(NewVertex) instead of the NewVertex, if an existing vertex that the NewVertex will connect to is a SamplingVertex. I think this is a reasonable assumption as new vertices that consume outputs from sampling vertices will process a subset of data anyways, and no such new vertex will reach the original DAG except via control edges. With this change Nemo is able to pipeline the MessageBarrierVertex (wrapped inside a SamplingVertex), avoiding duplicate data materialization.

(2) Regarding connecting the message aggregation vertex to the partition Sources:

I'd prefer not to do this, at least considering current use cases we have.

Here's the physical DAG diagram of PerKeyMedianITCase#testLargeShuffleSamplingSkew (including the fix for (1))
https://nemo.snuspl.snu.ac.kr:50443/nemo-dag-out/7a1c136ac24f427ebb4c34a43712da3f.svg

In the diagram the ScheduleGroup property is set such that the sampling partition does always execute prior to the original partition. In particular the ordering ScheduleGroup0(Stage3+Stage5) ==> ScheduleGroup1(Stage4+Stage6) is enforced although Stage4 ==> Stage6 is PUSH (which makes sense when considering each schedule group as a big vertex). The sampling should also happen prior to the execution of the original partition when Stage4 ==> Stage6 is PULL as well, although the schedule groups may differ in this case. I think this shows that a sequence of insert(samplingVertex) and insert(messageVertex) captures our intention fairly well.

I did write some code to try to 'extend' the control edges from (sampling vertices) to (existing vertices), by adding new control edges from (new vertices that connect to sampling vertices) to (existing vertices) upon each insert(). However, I ultimately I felt that this approach complicates the code quite a bit, and reverted the code back to the current approach which I think works for the current use cases.

For (2), I understood why the current version of the pass works normally.
However, scheduling indirectly depending on the logic of scheduling group is at a risk. (The order of scheduling according to the scheduling group is implicit and might be changed.)
If it is not simple to add the control dependency as I mentioned, please create an issue about it and mark as TODO.

Agreed. I've filed the JIRA: https://issues.apache.org/jira/browse/NEMO-343

johnyangk

Thanks @sanha! I've addressed your comments.

johnyangk · 2019-02-20T06:40:28Z

common/src/main/java/org/apache/nemo/common/Util.java

+    });
+
+    edgeToClone.getPropertyValue(PartitionerProperty.class).ifPresent(p -> {
+      if (p.right() == PartitionerProperty.NUM_EQUAL_TO_DST_PARALLELISM) {


It is needed. Please see: https://github.com/apache/incubator-nemo/blob/master/common/src/main/java/org/apache/nemo/common/ir/edge/executionproperty/PartitionerProperty.java#L66

johnyangk · 2019-02-20T06:42:34Z

common/src/main/java/org/apache/nemo/common/Util.java

+   * @param dst vertex.
+   * @return the control edge.
+   */
+  public static IREdge createControlEdge(final IRVertex src, final IRVertex dst) {


johnyangk · 2019-02-20T06:43:52Z

common/src/main/java/org/apache/nemo/common/ir/IRDAG.java

   * @param messageBarrierVertex to insert.
   * @param messageAggregatorVertex to insert.
   * @param mbvOutputEncoder to use.
   * @param mbvOutputDecoder to use.
   * @param edgesToGetStatisticsOf to examine.
+   * @param edgesToOptimize to optimize.
   */
  public void insert(final MessageBarrierVertex messageBarrierVertex,


Created a JIRA for this: https://issues.apache.org/jira/browse/NEMO-341

johnyangk · 2019-02-20T06:45:35Z

common/src/main/java/org/apache/nemo/common/ir/IRDAG.java

+   * @param samplingVertices to insert.
+   * @param executeAfterSamplingVertices that must be executed after samplingVertices.
+   */
+  public void insert(final Set<SamplingVertex> samplingVertices,


Created a JIRA for this: https://issues.apache.org/jira/browse/NEMO-341
I also added the assertNonExistence() checker in this method and the other insert() methods.

johnyangk · 2019-02-20T07:25:36Z

...org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SamplingSkewReshapingPass.java

+            .orElseThrow(() -> new IllegalStateException());
+          final IREdge clonedShuffleEdge = rightBeforeShuffle.getCloneOfOriginalEdge(e);
+
+          final KeyExtractor keyExtractor = e.getPropertyValue(KeyExtractorProperty.class).get();


Thanks for bringing this up!

(1) Regarding pipelining MessageBarrierVertex within a single stage with parent sampling vertices:

I've changed the semantics of insert() to use SamplingVertex(NewVertex) instead of the NewVertex, if an existing vertex that the NewVertex will connect to is a SamplingVertex. I think this is a reasonable assumption as new vertices that consume outputs from sampling vertices will process a subset of data anyways, and no such new vertex will reach the original DAG except via control edges. With this change Nemo is able to pipeline the MessageBarrierVertex (wrapped inside a SamplingVertex), avoiding duplicate data materialization.

(2) Regarding connecting the message aggregation vertex to the partition Sources:

I'd prefer not to do this, at least considering current use cases we have.

Here's the physical DAG diagram of PerKeyMedianITCase#testLargeShuffleSamplingSkew (including the fix for (1))
https://nemo.snuspl.snu.ac.kr:50443/nemo-dag-out/7a1c136ac24f427ebb4c34a43712da3f.svg

In the diagram the ScheduleGroup property is set such that the sampling partition does always execute prior to the original partition. In particular the ordering ScheduleGroup0(Stage3+Stage5) ==> ScheduleGroup1(Stage4+Stage6) is enforced although Stage4 ==> Stage6 is PUSH (which makes sense when considering each schedule group as a big vertex). The sampling should also happen prior to the execution of the original partition when Stage4 ==> Stage6 is PULL as well, although the schedule groups may differ in this case. I think this shows that a sequence of insert(samplingVertex) and insert(messageVertex) captures our intention fairly well.

I did write some code to try to 'extend' the control edges from (sampling vertices) to (existing vertices), by adding new control edges from (new vertices that connect to sampling vertices) to (existing vertices) upon each insert(). However, I ultimately I felt that this approach complicates the code quite a bit, and reverted the code back to the current approach which I think works for the current use cases.

sanha

Thanks for the change! I left some minor comments. Please check it out.

sanha · 2019-02-21T04:27:51Z

common/src/main/java/org/apache/nemo/common/ir/vertex/utility/SamplingVertex.java

+  }
+
+  public IRVertex getCloneOfOriginalVertex() {
+    this.copyExecutionPropertiesTo(cloneOfOriginalVertex);


If the purpose of this method is creating a new clone of the original vertex (for every call), let's create a new clone and copy the EPs instead of returning already cloned cloneOfOriginalVertex.
If not, let's copy the EPs in construction.

Plus, please add some method level comments for these methods.

sanha · 2019-02-21T04:29:09Z

...org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SamplingSkewReshapingPass.java

+ * This pass effectively partitions the IRDAG by non-oneToOne edges, clones each subDAG partition using SamplingVertex
+ * to process sampled data, and executes each cloned partition prior to executing the corresponding original partition.
+ *
+ * Suppose the IRDAG is partitioned into three sub-DAGs as follows:


three sub-DAGs connected with shuffle edges?

sanha · 2019-02-21T04:29:40Z

...org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SamplingSkewReshapingPass.java

+ *
+ * Then, this pass will produce something like:
+ * P1' - P1 - P2
+ *          - P2' - P2 - P3


Why the P2 is executed twice?

sanha · 2019-02-21T04:35:13Z

...org/apache/nemo/compiler/optimizer/pass/compiletime/reshaping/SamplingSkewReshapingPass.java

+            .orElseThrow(() -> new IllegalStateException());
+          final IREdge clonedShuffleEdge = rightBeforeShuffle.getCloneOfOriginalEdge(e);
+
+          final KeyExtractor keyExtractor = e.getPropertyValue(KeyExtractorProperty.class).get();


For (2), I understood why the current version of the pass works normally.
However, scheduling indirectly depending on the logic of scheduling group is at a risk. (The order of scheduling according to the scheduling group is implicit and might be changed.)
If it is not simple to add the control dependency as I mentioned, please create an issue about it and mark as TODO.

johnyangk · 2019-02-21T05:51:23Z

Thanks @sanha! I've addressed the comments.

sanha

Thanks, @johnyangk! LGTM. I'll merge this.

John Yang added 17 commits February 7, 2019 16:20

init

c35039f

save

5fc685d

nit

99592da

chkpt

3ee41bb

chkpt

ec12b39

save

f284cf0

chkpt

c6f8b7c

save

24bbded

backend

c88107a

refactor

fd38ac9

itcase chkpt

87cebbc

save

6906214

save

868eebf

reuse code

323a342

ready for checkstyle

4d0197d

checkstyle

62f82b0

one test left

a67154a

johnyangk self-assigned this Feb 13, 2019

johnyangk requested a review from sanha February 13, 2019 03:32

John Yang added 3 commits February 13, 2019 16:35

nits

4a9a125

nit

7202882

nits

f6ea04a

sanha requested changes Feb 19, 2019

View reviewed changes

johnyangk added 5 commits February 20, 2019 10:51

all comments except the last

43aa13e

save edge extensions

a5c1c9d

simplify builder.build()

1231463

checkstyle

91cdf03

comment

208a7a1

johnyangk commented Feb 20, 2019

View reviewed changes

johnyangk added 2 commits February 20, 2019 16:37

merge master

f5d6985

checkstyle

7ab75af

sanha requested changes Feb 21, 2019

View reviewed changes

johnyangk added 2 commits February 21, 2019 14:49

second comments

ed6f81b

add todo

fe9af90

nit

bab9a44

sanha approved these changes Feb 21, 2019

View reviewed changes

sanha merged commit a2b02dc into apache:master Feb 21, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[NEMO-338] SkewSamplingPass #193

[NEMO-338] SkewSamplingPass #193

johnyangk commented Feb 13, 2019 •

edited

Loading

johnyangk commented Feb 13, 2019

sanha left a comment

sanha Feb 14, 2019

sanha Feb 14, 2019

sanha Feb 15, 2019

johnyangk Feb 20, 2019

sanha Feb 19, 2019

johnyangk Feb 20, 2019

sanha Feb 19, 2019

johnyangk Feb 20, 2019

sanha Feb 19, 2019

johnyangk Feb 20, 2019

sanha Feb 19, 2019

sanha Feb 19, 2019

johnyangk Feb 20, 2019

sanha Feb 21, 2019

johnyangk Feb 21, 2019

johnyangk left a comment

johnyangk Feb 20, 2019

johnyangk Feb 20, 2019

johnyangk Feb 20, 2019

johnyangk Feb 20, 2019

johnyangk Feb 20, 2019

sanha left a comment

sanha Feb 21, 2019 •

edited

Loading

sanha Feb 21, 2019

sanha Feb 21, 2019

sanha Feb 21, 2019

johnyangk commented Feb 21, 2019

sanha left a comment

[NEMO-338] SkewSamplingPass #193

[NEMO-338] SkewSamplingPass #193

Conversation

johnyangk commented Feb 13, 2019 • edited Loading

johnyangk commented Feb 13, 2019

sanha 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

johnyangk 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

sanha left a comment

Choose a reason for hiding this comment

sanha Feb 21, 2019 • edited Loading

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

johnyangk commented Feb 21, 2019

sanha left a comment

Choose a reason for hiding this comment

johnyangk commented Feb 13, 2019 •

edited

Loading

sanha Feb 21, 2019 •

edited

Loading