You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by bi...@apache.org on 2013/08/30 02:59:49 UTC
git commit: TEZ-410. Refactor EdgeProperties to be more clear
Updated Branches:
refs/heads/master 9326d6983 -> 33d0dcefb
TEZ-410. Refactor EdgeProperties to be more clear
Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/33d0dcef
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/33d0dcef
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/33d0dcef
Branch: refs/heads/master
Commit: 33d0dcefb61037fef6255a3f56a7a29ccfe5b78d
Parents: 9326d69
Author: Bikas Saha <bi...@apache.org>
Authored: Thu Aug 29 17:57:27 2013 -0700
Committer: Bikas Saha <bi...@apache.org>
Committed: Thu Aug 29 17:57:27 2013 -0700
----------------------------------------------------------------------
.../main/java/org/apache/tez/dag/api/DAG.java | 23 +++-
.../apache/tez/dag/api/DagTypeConverters.java | 83 ++++++++------
.../org/apache/tez/dag/api/EdgeProperty.java | 102 +++++++++++++----
tez-dag-api/src/main/proto/DAGApiRecords.proto | 28 +++--
.../org/apache/tez/dag/api/TestDAGPlan.java | 10 +-
.../org/apache/tez/dag/api/TestDAGVerify.java | 110 ++++++++++++++++---
.../apache/tez/dag/app/dag/impl/DAGImpl.java | 4 +-
.../dag/app/dag/impl/ShuffleVertexManager.java | 4 +-
.../apache/tez/dag/app/dag/impl/VertexImpl.java | 4 +-
.../tez/dag/app/dag/impl/TestDAGImpl.java | 45 +++++---
.../tez/dag/app/dag/impl/TestVertexImpl.java | 35 +++---
.../dag/app/dag/impl/TestVertexScheduler.java | 37 +++++--
.../tez/mapreduce/examples/MRRSleepJob.java | 8 +-
.../mapreduce/examples/OrderedWordCount.java | 8 +-
.../apache/tez/mapreduce/TestMRRJobsDAGApi.java | 17 ++-
.../org/apache/tez/mapreduce/YARNRunner.java | 8 +-
16 files changed, 370 insertions(+), 156 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-dag-api/src/main/java/org/apache/tez/dag/api/DAG.java
----------------------------------------------------------------------
diff --git a/tez-dag-api/src/main/java/org/apache/tez/dag/api/DAG.java b/tez-dag-api/src/main/java/org/apache/tez/dag/api/DAG.java
index 3958b95..9cb602c 100644
--- a/tez-dag-api/src/main/java/org/apache/tez/dag/api/DAG.java
+++ b/tez-dag-api/src/main/java/org/apache/tez/dag/api/DAG.java
@@ -31,7 +31,9 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.Resource;
-import org.apache.tez.dag.api.EdgeProperty.ConnectionPattern;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
+import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
import org.apache.tez.dag.api.VertexLocationHint.TaskLocationHint;
import org.apache.tez.dag.api.records.DAGProtos.ConfigurationProto;
import org.apache.tez.dag.api.records.DAGProtos.DAGPlan;
@@ -169,11 +171,21 @@ public class DAG { // FIXME rename to Topology
if(restricted){
for(Edge e : edges){
vertexMap.get(e.getInputVertex().getVertexName()).outDegree++;
- if (e.getEdgeProperty().getConnectionPattern() !=
- ConnectionPattern.BIPARTITE) {
+ if (e.getEdgeProperty().getDataMovementType() !=
+ DataMovementType.SCATTER_GATHER) {
throw new IllegalStateException(
"Unsupported connection pattern on edge. " + e);
}
+ if (e.getEdgeProperty().getDataSourceType() !=
+ DataSourceType.PERSISTED) {
+ throw new IllegalStateException(
+ "Unsupported source type on edge. " + e);
+ }
+ if (e.getEdgeProperty().getSchedulingType() !=
+ SchedulingType.SEQUENTIAL) {
+ throw new IllegalStateException(
+ "Unsupported scheduling type on edge. " + e);
+ }
}
for(AnnotatedVertex av: vertexMap.values()){
if (av.outDegree > 1) {
@@ -338,8 +350,9 @@ public class DAG { // FIXME rename to Topology
edgeBuilder.setId(edge.getId());
edgeBuilder.setInputVertexName(edge.getInputVertex().getVertexName());
edgeBuilder.setOutputVertexName(edge.getOutputVertex().getVertexName());
- edgeBuilder.setConnectionPattern(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().connectionPattern));
- edgeBuilder.setSourceType(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().getSourceType()));
+ edgeBuilder.setDataMovementType(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().getDataMovementType()));
+ edgeBuilder.setDataSourceType(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().getDataSourceType()));
+ edgeBuilder.setSchedulingType(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().getSchedulingType()));
edgeBuilder.setEdgeSource(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().getEdgeSource()));
edgeBuilder.setEdgeDestination(DagTypeConverters.convertToDAGPlan(edge.getEdgeProperty().getEdgeDestination()));
dagBuilder.addEdge(edgeBuilder);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-dag-api/src/main/java/org/apache/tez/dag/api/DagTypeConverters.java
----------------------------------------------------------------------
diff --git a/tez-dag-api/src/main/java/org/apache/tez/dag/api/DagTypeConverters.java b/tez-dag-api/src/main/java/org/apache/tez/dag/api/DagTypeConverters.java
index 143186d..fd07b5b 100644
--- a/tez-dag-api/src/main/java/org/apache/tez/dag/api/DagTypeConverters.java
+++ b/tez-dag-api/src/main/java/org/apache/tez/dag/api/DagTypeConverters.java
@@ -31,13 +31,15 @@ import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.URL;
import org.apache.hadoop.yarn.api.records.impl.pb.LocalResourcePBImpl;
import org.apache.hadoop.yarn.util.ConverterUtils;
-import org.apache.tez.dag.api.EdgeProperty.ConnectionPattern;
-import org.apache.tez.dag.api.EdgeProperty.SourceType;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
+import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
import org.apache.tez.dag.api.VertexLocationHint.TaskLocationHint;
import org.apache.tez.dag.api.records.DAGProtos.ConfigurationProto;
import org.apache.tez.dag.api.records.DAGProtos.EdgePlan;
-import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeConnectionPattern;
-import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeSourceType;
+import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeDataMovementType;
+import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeDataSourceType;
+import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeSchedulingType;
import org.apache.tez.dag.api.records.DAGProtos.PlanKeyValuePair;
import org.apache.tez.dag.api.records.DAGProtos.PlanLocalResource;
import org.apache.tez.dag.api.records.DAGProtos.PlanLocalResourceType;
@@ -56,7 +58,7 @@ public class DagTypeConverters {
case PUBLIC : return PlanLocalResourceVisibility.PUBLIC;
case PRIVATE : return PlanLocalResourceVisibility.PRIVATE;
case APPLICATION : return PlanLocalResourceVisibility.APPLICATION;
- default : throw new RuntimeException("unknown 'visibility'");
+ default : throw new RuntimeException("unknown 'visibility': " + visibility);
}
}
@@ -65,43 +67,59 @@ public class DagTypeConverters {
case PUBLIC : return LocalResourceVisibility.PUBLIC;
case PRIVATE : return LocalResourceVisibility.PRIVATE;
case APPLICATION : return LocalResourceVisibility.APPLICATION;
- default : throw new RuntimeException("unknown 'visibility'");
+ default : throw new RuntimeException("unknown 'visibility': " + visibility);
}
}
- public static PlanEdgeSourceType convertToDAGPlan(SourceType sourceType){
+ public static PlanEdgeDataSourceType convertToDAGPlan(DataSourceType sourceType){
switch(sourceType){
- case STABLE : return PlanEdgeSourceType.STABLE;
- case STABLE_PERSISTED : return PlanEdgeSourceType.STABLE_PERSISTED;
- case STREAMING : return PlanEdgeSourceType.STREAMING;
- default : throw new RuntimeException("unknown 'sourceType'");
+ case PERSISTED : return PlanEdgeDataSourceType.PERSISTED;
+ case PERSISTED_RELIABLE : return PlanEdgeDataSourceType.PERSISTED_RELIABLE;
+ case EPHEMERAL : return PlanEdgeDataSourceType.EPHEMERAL;
+ default : throw new RuntimeException("unknown 'dataSourceType': " + sourceType);
}
}
- public static SourceType convertFromDAGPlan(PlanEdgeSourceType sourceType){
+ public static DataSourceType convertFromDAGPlan(PlanEdgeDataSourceType sourceType){
switch(sourceType){
- case STABLE : return SourceType.STABLE;
- case STABLE_PERSISTED : return SourceType.STABLE_PERSISTED;
- case STREAMING : return SourceType.STREAMING;
- default : throw new RuntimeException("unknown 'sourceType'");
+ case PERSISTED : return DataSourceType.PERSISTED;
+ case PERSISTED_RELIABLE : return DataSourceType.PERSISTED_RELIABLE;
+ case EPHEMERAL : return DataSourceType.EPHEMERAL;
+ default : throw new RuntimeException("unknown 'dataSourceType': " + sourceType);
}
}
- public static PlanEdgeConnectionPattern convertToDAGPlan(ConnectionPattern pattern){
- switch(pattern){
- case ONE_TO_ONE : return PlanEdgeConnectionPattern.ONE_TO_ONE;
- case ONE_TO_ALL : return PlanEdgeConnectionPattern.ONE_TO_ALL;
- case BIPARTITE : return PlanEdgeConnectionPattern.BIPARTITE;
- default : throw new RuntimeException("unknown 'pattern'");
+ public static PlanEdgeDataMovementType convertToDAGPlan(DataMovementType type){
+ switch(type){
+ case ONE_TO_ONE : return PlanEdgeDataMovementType.ONE_TO_ONE;
+ case BROADCAST : return PlanEdgeDataMovementType.BROADCAST;
+ case SCATTER_GATHER : return PlanEdgeDataMovementType.SCATTER_GATHER;
+ default : throw new RuntimeException("unknown 'dataMovementType': " + type);
+ }
+ }
+
+ public static DataMovementType convertFromDAGPlan(PlanEdgeDataMovementType type){
+ switch(type){
+ case ONE_TO_ONE : return DataMovementType.ONE_TO_ONE;
+ case BROADCAST : return DataMovementType.BROADCAST;
+ case SCATTER_GATHER : return DataMovementType.SCATTER_GATHER;
+ default : throw new IllegalArgumentException("unknown 'dataMovementType': " + type);
}
}
- public static ConnectionPattern convertFromDAGPlan(PlanEdgeConnectionPattern pattern){
- switch(pattern){
- case ONE_TO_ONE : return ConnectionPattern.ONE_TO_ONE;
- case ONE_TO_ALL : return ConnectionPattern.ONE_TO_ALL;
- case BIPARTITE : return ConnectionPattern.BIPARTITE;
- default : throw new IllegalArgumentException("unknown 'pattern'");
+ public static PlanEdgeSchedulingType convertToDAGPlan(SchedulingType type){
+ switch(type){
+ case SEQUENTIAL : return PlanEdgeSchedulingType.SEQUENTIAL;
+ case CONCURRENT : return PlanEdgeSchedulingType.CONCURRENT;
+ default : throw new RuntimeException("unknown 'SchedulingType': " + type);
+ }
+ }
+
+ public static SchedulingType convertFromDAGPlan(PlanEdgeSchedulingType type){
+ switch(type){
+ case SEQUENTIAL : return SchedulingType.SEQUENTIAL;
+ case CONCURRENT : return SchedulingType.CONCURRENT;
+ default : throw new IllegalArgumentException("unknown 'SchedulingType': " + type);
}
}
@@ -110,7 +128,7 @@ public class DagTypeConverters {
case ARCHIVE : return PlanLocalResourceType.ARCHIVE;
case FILE : return PlanLocalResourceType.FILE;
case PATTERN : return PlanLocalResourceType.PATTERN;
- default : throw new IllegalArgumentException("unknown 'type'");
+ default : throw new IllegalArgumentException("unknown 'type': " + type);
}
}
@@ -119,7 +137,7 @@ public class DagTypeConverters {
case ARCHIVE : return LocalResourceType.ARCHIVE;
case FILE : return LocalResourceType.FILE;
case PATTERN : return LocalResourceType.PATTERN;
- default : throw new IllegalArgumentException("unknown 'type'");
+ default : throw new IllegalArgumentException("unknown 'type': " + type);
}
}
@@ -198,8 +216,9 @@ public class DagTypeConverters {
for(EdgePlan edge: edgeList){
map.put(edge.getId(),
new EdgeProperty(
- convertFromDAGPlan(edge.getConnectionPattern()),
- convertFromDAGPlan(edge.getSourceType()),
+ convertFromDAGPlan(edge.getDataMovementType()),
+ convertFromDAGPlan(edge.getDataSourceType()),
+ convertFromDAGPlan(edge.getSchedulingType()),
convertOutputDescriptorFromDAGPlan(edge.getEdgeSource()),
convertInputDescriptorFromDAGPlan(edge.getEdgeDestination())
)
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-dag-api/src/main/java/org/apache/tez/dag/api/EdgeProperty.java
----------------------------------------------------------------------
diff --git a/tez-dag-api/src/main/java/org/apache/tez/dag/api/EdgeProperty.java b/tez-dag-api/src/main/java/org/apache/tez/dag/api/EdgeProperty.java
index d9ddd23..326d3d0 100644
--- a/tez-dag-api/src/main/java/org/apache/tez/dag/api/EdgeProperty.java
+++ b/tez-dag-api/src/main/java/org/apache/tez/dag/api/EdgeProperty.java
@@ -17,49 +17,107 @@
*/
package org.apache.tez.dag.api;
-public class EdgeProperty { // FIXME rename to ChannelProperty
+public class EdgeProperty {
- public enum ConnectionPattern {
+ /**
+ * Defines the manner of data movement between source and destination tasks.
+ * Determines which destination tasks have access to data produced on this
+ * edge by a source task. A destination task may choose to read any portion of
+ * the data available to it.
+ */
+ public enum DataMovementType {
+ /**
+ * Output on this edge produced by the i-th source task is available to the
+ * i-th destination task.
+ */
ONE_TO_ONE,
- ONE_TO_ALL,
- BIPARTITE // FIXME rename to SHUFFLE
+ /**
+ * Output on this edge produced by any source task is available to all
+ * destination tasks.
+ */
+ BROADCAST,
+ /**
+ * The i-th output on this edge produced by all source tasks is available to
+ * the same destination task. Source tasks scatter their outputs and they
+ * are gathered by designated destination tasks.
+ */
+ SCATTER_GATHER
+ }
+
+ /**
+ * Determines the lifetime of the data produced on this edge by a source task.
+ */
+ public enum DataSourceType {
+ /**
+ * Data produced by the source is persisted and available even when the
+ * task is not running. The data may become unavailable and may cause the
+ * source task to be re-executed.
+ */
+ PERSISTED,
+ /**
+ * Source data is stored reliably and will always be available
+ */
+ PERSISTED_RELIABLE,
+ /**
+ * Data produced by the source task is available only while the source task
+ * is running. This requires the destination task to run concurrently with
+ * the source task.
+ */
+ EPHEMERAL
}
- public enum SourceType {
- STABLE,
- STABLE_PERSISTED,
- STREAMING
+ /**
+ * Determines when the destination task is eligible to run, once the source
+ * task is eligible to run.
+ */
+ public enum SchedulingType {
+ /**
+ * Destination task is eligible to run after one or more of its source tasks
+ * have started or completed.
+ */
+ SEQUENTIAL,
+ /**
+ * Destination task must run concurrently with the source task
+ */
+ CONCURRENT
}
- ConnectionPattern connectionPattern;
- SourceType sourceType;
+ DataMovementType dataMovementType;
+ DataSourceType dataSourceType;
+ SchedulingType schedulingType;
InputDescriptor inputDescriptor;
OutputDescriptor outputDescriptor;
/**
- * @param connectionPattern
- * @param sourceType
+ * @param dataMovementType
+ * @param dataSourceType
* @param edgeSource
* The {@link OutputDescriptor} that generates data on the edge.
* @param edgeDestination
* The {@link InputDescriptor} which will consume data from the edge.
*/
- public EdgeProperty(ConnectionPattern connectionPattern,
- SourceType sourceType,
+ public EdgeProperty(DataMovementType dataMovementType,
+ DataSourceType dataSourceType,
+ SchedulingType schedulingType,
OutputDescriptor edgeSource,
InputDescriptor edgeDestination) {
- this.connectionPattern = connectionPattern;
- this.sourceType = sourceType;
+ this.dataMovementType = dataMovementType;
+ this.dataSourceType = dataSourceType;
+ this.schedulingType = schedulingType;
this.inputDescriptor = edgeDestination;
this.outputDescriptor = edgeSource;
}
- public ConnectionPattern getConnectionPattern() {
- return connectionPattern;
+ public DataMovementType getDataMovementType() {
+ return dataMovementType;
+ }
+
+ public DataSourceType getDataSourceType() {
+ return dataSourceType;
}
- public SourceType getSourceType() {
- return sourceType;
+ public SchedulingType getSchedulingType() {
+ return schedulingType;
}
/**
@@ -82,8 +140,8 @@ public class EdgeProperty { // FIXME rename to ChannelProperty
@Override
public String toString() {
- return "{ " + connectionPattern + " : " + inputDescriptor.getClassName()
- + " >> " + sourceType + " >> " + outputDescriptor.getClassName() + " }";
+ return "{ " + dataMovementType + " : " + inputDescriptor.getClassName()
+ + " >> " + dataSourceType + " >> " + outputDescriptor.getClassName() + " }";
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-dag-api/src/main/proto/DAGApiRecords.proto
----------------------------------------------------------------------
diff --git a/tez-dag-api/src/main/proto/DAGApiRecords.proto b/tez-dag-api/src/main/proto/DAGApiRecords.proto
index 85016b0..4385749 100644
--- a/tez-dag-api/src/main/proto/DAGApiRecords.proto
+++ b/tez-dag-api/src/main/proto/DAGApiRecords.proto
@@ -35,16 +35,21 @@ enum PlanVertexType {
OUTPUT = 2;
}
-enum PlanEdgeConnectionPattern {
+enum PlanEdgeDataMovementType {
ONE_TO_ONE = 0;
- ONE_TO_ALL = 1;
- BIPARTITE = 2;
+ BROADCAST = 1;
+ SCATTER_GATHER = 2;
}
-enum PlanEdgeSourceType {
- STABLE = 0;
- STABLE_PERSISTED = 1;
- STREAMING = 2;
+enum PlanEdgeDataSourceType {
+ PERSISTED = 0;
+ PERSISTED_RELIABLE = 1;
+ EPHEMERAL = 2;
+}
+
+enum PlanEdgeSchedulingType {
+ SEQUENTIAL = 0;
+ CONCURRENT = 1;
}
message PlanKeyValuePair {
@@ -111,10 +116,11 @@ message EdgePlan {
required string id = 1;
required string inputVertexName = 2;
required string outputVertexName = 3;
- required PlanEdgeConnectionPattern connectionPattern = 4;
- required PlanEdgeSourceType sourceType = 5;
- optional TezEntityDescriptorProto edge_source = 6;
- optional TezEntityDescriptorProto edge_destination = 7;
+ required PlanEdgeDataMovementType dataMovementType = 4;
+ required PlanEdgeDataSourceType dataSourceType = 5;
+ required PlanEdgeSchedulingType schedulingType = 6;
+ optional TezEntityDescriptorProto edge_source = 7;
+ optional TezEntityDescriptorProto edge_destination = 8;
}
message ConfigurationProto {
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-dag-api/src/test/java/org/apache/tez/dag/api/TestDAGPlan.java
----------------------------------------------------------------------
diff --git a/tez-dag-api/src/test/java/org/apache/tez/dag/api/TestDAGPlan.java b/tez-dag-api/src/test/java/org/apache/tez/dag/api/TestDAGPlan.java
index 8dd0826..c8d09d0 100644
--- a/tez-dag-api/src/test/java/org/apache/tez/dag/api/TestDAGPlan.java
+++ b/tez-dag-api/src/test/java/org/apache/tez/dag/api/TestDAGPlan.java
@@ -29,8 +29,9 @@ import java.util.Map;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.Resource;
-import org.apache.tez.dag.api.EdgeProperty.ConnectionPattern;
-import org.apache.tez.dag.api.EdgeProperty.SourceType;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
+import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
import org.apache.tez.dag.api.records.DAGProtos.DAGPlan;
import org.apache.tez.dag.api.records.DAGProtos.EdgePlan;
import org.apache.tez.dag.api.records.DAGProtos.PlanTaskConfiguration;
@@ -110,8 +111,9 @@ public class TestDAGPlan {
setUserPayload("inputBytes".getBytes());
OutputDescriptor outputDescriptor = new OutputDescriptor("output").
setUserPayload("outputBytes".getBytes());
- Edge edge = new Edge(v1, v2, new EdgeProperty(ConnectionPattern.BIPARTITE,
- SourceType.STABLE, outputDescriptor, inputDescriptor));
+ Edge edge = new Edge(v1, v2, new EdgeProperty(
+ DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL, outputDescriptor, inputDescriptor));
dag.addVertex(v1).addVertex(v2).addEdge(edge);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-dag-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
----------------------------------------------------------------------
diff --git a/tez-dag-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java b/tez-dag-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
index e66df54..b33f3a6 100644
--- a/tez-dag-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
+++ b/tez-dag-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
@@ -19,8 +19,9 @@
package org.apache.tez.dag.api;
import org.apache.hadoop.yarn.api.records.Resource;
-import org.apache.tez.dag.api.EdgeProperty.ConnectionPattern;
-import org.apache.tez.dag.api.EdgeProperty.SourceType;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
+import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
import org.junit.Assert;
import org.junit.Test;
@@ -44,7 +45,68 @@ public class TestDAGVerify {
new ProcessorDescriptor("MapProcessor"),
dummyTaskCount, dummyTaskResource);
Edge e1 = new Edge(v1, v2,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
+ new OutputDescriptor(dummyOutputClassName),
+ new InputDescriptor(dummyInputClassName)));
+ DAG dag = new DAG("testDag");
+ dag.addVertex(v1);
+ dag.addVertex(v2);
+ dag.addEdge(e1);
+ dag.verify();
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testVerify2() {
+ Vertex v1 = new Vertex("v1",
+ new ProcessorDescriptor(dummyProcessorClassName),
+ dummyTaskCount, dummyTaskResource);
+ Vertex v2 = new Vertex("v2",
+ new ProcessorDescriptor("MapProcessor"),
+ dummyTaskCount, dummyTaskResource);
+ Edge e1 = new Edge(v1, v2,
+ new EdgeProperty(DataMovementType.ONE_TO_ONE,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
+ new OutputDescriptor(dummyOutputClassName),
+ new InputDescriptor(dummyInputClassName)));
+ DAG dag = new DAG("testDag");
+ dag.addVertex(v1);
+ dag.addVertex(v2);
+ dag.addEdge(e1);
+ dag.verify();
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testVerify3() {
+ Vertex v1 = new Vertex("v1",
+ new ProcessorDescriptor(dummyProcessorClassName),
+ dummyTaskCount, dummyTaskResource);
+ Vertex v2 = new Vertex("v2",
+ new ProcessorDescriptor("MapProcessor"),
+ dummyTaskCount, dummyTaskResource);
+ Edge e1 = new Edge(v1, v2,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.EPHEMERAL, SchedulingType.SEQUENTIAL,
+ new OutputDescriptor(dummyOutputClassName),
+ new InputDescriptor(dummyInputClassName)));
+ DAG dag = new DAG("testDag");
+ dag.addVertex(v1);
+ dag.addVertex(v2);
+ dag.addEdge(e1);
+ dag.verify();
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testVerify4() {
+ Vertex v1 = new Vertex("v1",
+ new ProcessorDescriptor(dummyProcessorClassName),
+ dummyTaskCount, dummyTaskResource);
+ Vertex v2 = new Vertex("v2",
+ new ProcessorDescriptor("MapProcessor"),
+ dummyTaskCount, dummyTaskResource);
+ Edge e1 = new Edge(v1, v2,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.EPHEMERAL, SchedulingType.CONCURRENT,
new OutputDescriptor(dummyOutputClassName),
new InputDescriptor(dummyInputClassName)));
DAG dag = new DAG("testDag");
@@ -75,19 +137,23 @@ public class TestDAGVerify {
new ProcessorDescriptor("MapProcessor"),
dummyTaskCount, dummyTaskResource);
Edge e1 = new Edge(v1, v2,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
Edge e2 = new Edge(v2, v3,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
Edge e3 = new Edge(v2, v4,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
Edge e4 = new Edge(v4, v1,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
DAG dag = new DAG("testDag");
@@ -131,19 +197,23 @@ public class TestDAGVerify {
new ProcessorDescriptor("MapProcessor"),
dummyTaskCount, dummyTaskResource);
Edge e1 = new Edge(v1, v2,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
Edge e2 = new Edge(v2, v3,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
Edge e3 = new Edge(v2, v4,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
Edge e4 = new Edge(v3, v2,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
DAG dag = new DAG("testDag");
@@ -206,11 +276,13 @@ public class TestDAGVerify {
new ProcessorDescriptor("ReduceProcessor"),
dummyTaskCount, dummyTaskResource);
Edge e1 = new Edge(v1, v3,
- new EdgeProperty(ConnectionPattern.ONE_TO_ONE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.ONE_TO_ONE,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
Edge e2 = new Edge(v2, v3,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
DAG dag = new DAG("testDag");
@@ -245,11 +317,13 @@ public class TestDAGVerify {
new ProcessorDescriptor("ReduceProcessor"),
dummyTaskCount, dummyTaskResource);
Edge e1 = new Edge(v1, v3,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
Edge e2 = new Edge(v2, v3,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
DAG dag = new DAG("testDag");
@@ -278,11 +352,13 @@ public class TestDAGVerify {
new ProcessorDescriptor("MapProcessor"),
dummyTaskCount, dummyTaskResource);
Edge e1 = new Edge(v1, v2,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
Edge e2 = new Edge(v1, v2,
- new EdgeProperty(ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
new OutputDescriptor("dummy output class"),
new InputDescriptor("dummy input class")));
DAG dag = new DAG("testDag");
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java
index 0e93421..2ddcdd7 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java
@@ -874,14 +874,14 @@ public class DAGImpl implements org.apache.tez.dag.app.dag.DAG,
Map<Vertex, EdgeProperty> inVertices = vertex.getInputVertices();
if (!(outVertices == null || outVertices.isEmpty() || (outVertices
.size() == 1 && outVertices.values().iterator().next()
- .getConnectionPattern() == EdgeProperty.ConnectionPattern.BIPARTITE))) {
+ .getDataMovementType() == EdgeProperty.DataMovementType.SCATTER_GATHER))) {
// more than 1 output OR single output is not bipartite
isMRR = false;
break;
}
if (!(inVertices == null || inVertices.isEmpty() || (inVertices
.size() == 1 && inVertices.values().iterator().next()
- .getConnectionPattern() == EdgeProperty.ConnectionPattern.BIPARTITE))) {
+ .getDataMovementType() == EdgeProperty.DataMovementType.SCATTER_GATHER))) {
// more than 1 output OR single output is not bipartite
isMRR = false;
break;
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ShuffleVertexManager.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ShuffleVertexManager.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ShuffleVertexManager.java
index e221483..bb6e2ee 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ShuffleVertexManager.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ShuffleVertexManager.java
@@ -33,7 +33,7 @@ import org.apache.tez.common.TezJobConfig;
import org.apache.tez.dag.api.EdgeProperty;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezUncheckedException;
-import org.apache.tez.dag.api.EdgeProperty.ConnectionPattern;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
import org.apache.tez.dag.app.dag.Vertex;
import org.apache.tez.dag.app.dag.VertexScheduler;
import org.apache.tez.dag.records.TezTaskAttemptID;
@@ -74,7 +74,7 @@ public class ShuffleVertexManager implements VertexScheduler {
this.managedVertex = managedVertex;
Map<Vertex, EdgeProperty> inputs = managedVertex.getInputVertices();
for(Map.Entry<Vertex, EdgeProperty> entry : inputs.entrySet()) {
- if(entry.getValue().getConnectionPattern() == ConnectionPattern.BIPARTITE) {
+ if(entry.getValue().getDataMovementType() == DataMovementType.SCATTER_GATHER) {
Vertex vertex = entry.getKey();
bipartiteSources.put(vertex.getVertexId(), vertex);
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
index c0137e2..49a5fb2 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
@@ -55,7 +55,7 @@ import org.apache.tez.common.OutputSpec;
import org.apache.tez.common.counters.TezCounters;
import org.apache.tez.dag.api.DagTypeConverters;
import org.apache.tez.dag.api.EdgeProperty;
-import org.apache.tez.dag.api.EdgeProperty.ConnectionPattern;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
import org.apache.tez.dag.api.ProcessorDescriptor;
import org.apache.tez.dag.api.TezUncheckedException;
import org.apache.tez.dag.api.VertexLocationHint;
@@ -955,7 +955,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
boolean hasBipartite = false;
if (vertex.sourceVertices != null) {
for (EdgeProperty edgeProperty : vertex.sourceVertices.values()) {
- if (edgeProperty.getConnectionPattern() == ConnectionPattern.BIPARTITE) {
+ if (edgeProperty.getDataMovementType() == DataMovementType.SCATTER_GATHER) {
hasBipartite = true;
break;
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGImpl.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGImpl.java
index acf40b1..6e42673 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGImpl.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGImpl.java
@@ -35,8 +35,9 @@ import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.oldrecords.TaskState;
import org.apache.tez.dag.api.records.DAGProtos.DAGPlan;
import org.apache.tez.dag.api.records.DAGProtos.EdgePlan;
-import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeConnectionPattern;
-import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeSourceType;
+import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeDataMovementType;
+import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeDataSourceType;
+import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeSchedulingType;
import org.apache.tez.dag.api.records.DAGProtos.PlanTaskConfiguration;
import org.apache.tez.dag.api.records.DAGProtos.PlanTaskLocationHint;
import org.apache.tez.dag.api.records.DAGProtos.PlanVertexType;
@@ -239,9 +240,10 @@ public class TestDAGImpl {
.setInputVertexName("vertex1")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o1"))
.setOutputVertexName("vertex2")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e1")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.addEdge(
@@ -250,9 +252,10 @@ public class TestDAGImpl {
.setInputVertexName("vertex2")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o2"))
.setOutputVertexName("vertex3")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e2")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.build();
@@ -409,9 +412,10 @@ public class TestDAGImpl {
.setInputVertexName("vertex1")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o1"))
.setOutputVertexName("vertex3")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e1")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.addEdge(
@@ -420,9 +424,10 @@ public class TestDAGImpl {
.setInputVertexName("vertex2")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o2"))
.setOutputVertexName("vertex3")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e2")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.addEdge(
@@ -431,9 +436,10 @@ public class TestDAGImpl {
.setInputVertexName("vertex3")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o3_v4"))
.setOutputVertexName("vertex4")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e3")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.addEdge(
@@ -442,9 +448,10 @@ public class TestDAGImpl {
.setInputVertexName("vertex3")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o3_v5"))
.setOutputVertexName("vertex5")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e4")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.addEdge(
@@ -453,9 +460,10 @@ public class TestDAGImpl {
.setInputVertexName("vertex4")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o4"))
.setOutputVertexName("vertex6")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e5")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.addEdge(
@@ -464,9 +472,10 @@ public class TestDAGImpl {
.setInputVertexName("vertex5")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o5"))
.setOutputVertexName("vertex6")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e6")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.build();
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
index b11aaab..238b2bd 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
@@ -50,8 +50,9 @@ import org.apache.tez.dag.api.committer.VertexOutputCommitter;
import org.apache.tez.dag.api.oldrecords.TaskState;
import org.apache.tez.dag.api.records.DAGProtos.DAGPlan;
import org.apache.tez.dag.api.records.DAGProtos.EdgePlan;
-import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeConnectionPattern;
-import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeSourceType;
+import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeDataMovementType;
+import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeDataSourceType;
+import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeSchedulingType;
import org.apache.tez.dag.api.records.DAGProtos.PlanTaskConfiguration;
import org.apache.tez.dag.api.records.DAGProtos.PlanTaskLocationHint;
import org.apache.tez.dag.api.records.DAGProtos.PlanVertexType;
@@ -381,9 +382,10 @@ public class TestVertexImpl {
.setInputVertexName("vertex1")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o1"))
.setOutputVertexName("vertex3")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e1")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.addEdge(
@@ -392,9 +394,10 @@ public class TestVertexImpl {
.setInputVertexName("vertex2")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o2"))
.setOutputVertexName("vertex3")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e2")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.addEdge(
@@ -403,9 +406,10 @@ public class TestVertexImpl {
.setInputVertexName("vertex3")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o3_v4"))
.setOutputVertexName("vertex4")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e3")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.addEdge(
@@ -414,9 +418,10 @@ public class TestVertexImpl {
.setInputVertexName("vertex3")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o3_v5"))
.setOutputVertexName("vertex5")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e4")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.addEdge(
@@ -425,9 +430,10 @@ public class TestVertexImpl {
.setInputVertexName("vertex4")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o4"))
.setOutputVertexName("vertex6")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e5")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.addEdge(
@@ -436,9 +442,10 @@ public class TestVertexImpl {
.setInputVertexName("vertex5")
.setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o5"))
.setOutputVertexName("vertex6")
- .setConnectionPattern(PlanEdgeConnectionPattern.BIPARTITE)
+ .setDataMovementType(PlanEdgeDataMovementType.SCATTER_GATHER)
.setId("e6")
- .setSourceType(PlanEdgeSourceType.STABLE)
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
.build()
)
.build();
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexScheduler.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexScheduler.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexScheduler.java
index 22cd885..5661c27 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexScheduler.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexScheduler.java
@@ -34,6 +34,7 @@ import org.apache.tez.dag.api.OutputDescriptor;
import org.apache.tez.dag.api.ProcessorDescriptor;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezUncheckedException;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
import org.apache.tez.dag.app.dag.Task;
import org.apache.tez.dag.app.dag.Vertex;
import org.apache.tez.dag.records.TezDAGID;
@@ -66,22 +67,28 @@ public class TestVertexScheduler {
Vertex mockSrcVertex1 = mock(Vertex.class);
TezVertexID mockSrcVertexId1 = new TezVertexID(dagId, 1);
EdgeProperty eProp1 = new EdgeProperty(
- EdgeProperty.ConnectionPattern.BIPARTITE,
- EdgeProperty.SourceType.STABLE, new OutputDescriptor("out"),
+ EdgeProperty.DataMovementType.SCATTER_GATHER,
+ EdgeProperty.DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL,
+ new OutputDescriptor("out"),
new InputDescriptor("in"));
when(mockSrcVertex1.getVertexId()).thenReturn(mockSrcVertexId1);
Vertex mockSrcVertex2 = mock(Vertex.class);
TezVertexID mockSrcVertexId2 = new TezVertexID(dagId, 2);
EdgeProperty eProp2 = new EdgeProperty(
- EdgeProperty.ConnectionPattern.BIPARTITE,
- EdgeProperty.SourceType.STABLE, new OutputDescriptor("out"),
+ EdgeProperty.DataMovementType.SCATTER_GATHER,
+ EdgeProperty.DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL,
+ new OutputDescriptor("out"),
new InputDescriptor("in"));
when(mockSrcVertex2.getVertexId()).thenReturn(mockSrcVertexId2);
Vertex mockSrcVertex3 = mock(Vertex.class);
TezVertexID mockSrcVertexId3 = new TezVertexID(dagId, 3);
EdgeProperty eProp3 = new EdgeProperty(
- EdgeProperty.ConnectionPattern.ONE_TO_ALL,
- EdgeProperty.SourceType.STABLE, new OutputDescriptor("out"),
+ EdgeProperty.DataMovementType.BROADCAST,
+ EdgeProperty.DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL,
+ new OutputDescriptor("out"),
new InputDescriptor("in"));
when(mockSrcVertex3.getVertexId()).thenReturn(mockSrcVertexId3);
@@ -231,22 +238,28 @@ public class TestVertexScheduler {
Vertex mockSrcVertex1 = mock(Vertex.class);
TezVertexID mockSrcVertexId1 = new TezVertexID(dagId, 1);
EdgeProperty eProp1 = new EdgeProperty(
- EdgeProperty.ConnectionPattern.BIPARTITE,
- EdgeProperty.SourceType.STABLE, new OutputDescriptor("out"),
+ EdgeProperty.DataMovementType.SCATTER_GATHER,
+ EdgeProperty.DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL,
+ new OutputDescriptor("out"),
new InputDescriptor("in"));
when(mockSrcVertex1.getVertexId()).thenReturn(mockSrcVertexId1);
Vertex mockSrcVertex2 = mock(Vertex.class);
TezVertexID mockSrcVertexId2 = new TezVertexID(dagId, 2);
EdgeProperty eProp2 = new EdgeProperty(
- EdgeProperty.ConnectionPattern.BIPARTITE,
- EdgeProperty.SourceType.STABLE, new OutputDescriptor("out"),
+ EdgeProperty.DataMovementType.SCATTER_GATHER,
+ EdgeProperty.DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL,
+ new OutputDescriptor("out"),
new InputDescriptor("in"));
when(mockSrcVertex2.getVertexId()).thenReturn(mockSrcVertexId2);
Vertex mockSrcVertex3 = mock(Vertex.class);
TezVertexID mockSrcVertexId3 = new TezVertexID(dagId, 3);
EdgeProperty eProp3 = new EdgeProperty(
- EdgeProperty.ConnectionPattern.ONE_TO_ALL,
- EdgeProperty.SourceType.STABLE, new OutputDescriptor("out"),
+ EdgeProperty.DataMovementType.BROADCAST,
+ EdgeProperty.DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL,
+ new OutputDescriptor("out"),
new InputDescriptor("in"));
when(mockSrcVertex3.getVertexId()).thenReturn(mockSrcVertexId3);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
index 0747f0e..b2f1318 100644
--- a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
+++ b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
@@ -72,8 +72,9 @@ import org.apache.tez.dag.api.ProcessorDescriptor;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezUncheckedException;
import org.apache.tez.dag.api.Vertex;
-import org.apache.tez.dag.api.EdgeProperty.ConnectionPattern;
-import org.apache.tez.dag.api.EdgeProperty.SourceType;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
+import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
import org.apache.tez.dag.api.client.DAGClient;
import org.apache.tez.dag.api.client.DAGStatus;
import org.apache.tez.engine.common.objectregistry.ObjectLifeCycle;
@@ -607,7 +608,8 @@ public class MRRSleepJob extends Configured implements Tool {
if (i != 0) {
dag.addEdge(new Edge(vertices.get(i-1),
vertices.get(i), new EdgeProperty(
- ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL,
new OutputDescriptor(
OnFileSortedOutput.class.getName()),
new InputDescriptor(
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/OrderedWordCount.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/OrderedWordCount.java b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/OrderedWordCount.java
index 0a6c98a..7fae4a3 100644
--- a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/OrderedWordCount.java
+++ b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/OrderedWordCount.java
@@ -65,8 +65,9 @@ import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezException;
import org.apache.tez.dag.api.TezUncheckedException;
import org.apache.tez.dag.api.Vertex;
-import org.apache.tez.dag.api.EdgeProperty.ConnectionPattern;
-import org.apache.tez.dag.api.EdgeProperty.SourceType;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
+import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
import org.apache.tez.dag.api.client.DAGClient;
import org.apache.tez.dag.api.client.DAGStatus;
import org.apache.tez.engine.lib.input.ShuffledMergedInput;
@@ -308,7 +309,8 @@ public class OrderedWordCount {
if (i != 0) {
dag.addEdge(new Edge(vertices.get(i-1),
vertices.get(i), new EdgeProperty(
- ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL,
new OutputDescriptor(
OnFileSortedOutput.class.getName()),
new InputDescriptor(
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-mapreduce-tests/src/test/java/org/apache/tez/mapreduce/TestMRRJobsDAGApi.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce-tests/src/test/java/org/apache/tez/mapreduce/TestMRRJobsDAGApi.java b/tez-mapreduce-tests/src/test/java/org/apache/tez/mapreduce/TestMRRJobsDAGApi.java
index 7eae23d..701ca87 100644
--- a/tez-mapreduce-tests/src/test/java/org/apache/tez/mapreduce/TestMRRJobsDAGApi.java
+++ b/tez-mapreduce-tests/src/test/java/org/apache/tez/mapreduce/TestMRRJobsDAGApi.java
@@ -62,8 +62,9 @@ import org.apache.tez.dag.api.ProcessorDescriptor;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezException;
import org.apache.tez.dag.api.Vertex;
-import org.apache.tez.dag.api.EdgeProperty.ConnectionPattern;
-import org.apache.tez.dag.api.EdgeProperty.SourceType;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
+import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
import org.apache.tez.dag.api.client.DAGClient;
import org.apache.tez.dag.api.client.DAGStatus;
import org.apache.tez.dag.api.client.DAGStatus.State;
@@ -393,19 +394,23 @@ public class TestMRRJobsDAGApi {
dag.addVertex(stage3Vertex);
Edge edge1 = new Edge(stage1Vertex, stage2Vertex, new EdgeProperty(
- ConnectionPattern.BIPARTITE, SourceType.STABLE, new OutputDescriptor(
+ DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL, new OutputDescriptor(
OnFileSortedOutput.class.getName()), new InputDescriptor(
ShuffledMergedInput.class.getName())));
Edge edge11 = new Edge(stage11Vertex, stage22Vertex, new EdgeProperty(
- ConnectionPattern.BIPARTITE, SourceType.STABLE, new OutputDescriptor(
+ DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL, new OutputDescriptor(
OnFileSortedOutput.class.getName()), new InputDescriptor(
ShuffledMergedInput.class.getName())));
Edge edge2 = new Edge(stage2Vertex, stage3Vertex, new EdgeProperty(
- ConnectionPattern.BIPARTITE, SourceType.STABLE, new OutputDescriptor(
+ DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL, new OutputDescriptor(
OnFileSortedOutput.class.getName()), new InputDescriptor(
ShuffledMergedInput.class.getName())));
Edge edge3 = new Edge(stage22Vertex, stage3Vertex, new EdgeProperty(
- ConnectionPattern.BIPARTITE, SourceType.STABLE, new OutputDescriptor(
+ DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL, new OutputDescriptor(
OnFileSortedOutput.class.getName()), new InputDescriptor(
ShuffledMergedInput.class.getName())));
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/33d0dcef/tez-yarn-client/src/main/java/org/apache/tez/mapreduce/YARNRunner.java
----------------------------------------------------------------------
diff --git a/tez-yarn-client/src/main/java/org/apache/tez/mapreduce/YARNRunner.java b/tez-yarn-client/src/main/java/org/apache/tez/mapreduce/YARNRunner.java
index 0799e09..0fb823f 100644
--- a/tez-yarn-client/src/main/java/org/apache/tez/mapreduce/YARNRunner.java
+++ b/tez-yarn-client/src/main/java/org/apache/tez/mapreduce/YARNRunner.java
@@ -82,8 +82,9 @@ import org.apache.tez.client.TezClient;
import org.apache.tez.dag.api.DAG;
import org.apache.tez.dag.api.Edge;
import org.apache.tez.dag.api.EdgeProperty;
-import org.apache.tez.dag.api.EdgeProperty.ConnectionPattern;
-import org.apache.tez.dag.api.EdgeProperty.SourceType;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
+import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
import org.apache.tez.dag.api.InputDescriptor;
import org.apache.tez.dag.api.OutputDescriptor;
import org.apache.tez.dag.api.ProcessorDescriptor;
@@ -460,7 +461,8 @@ public class YARNRunner implements ClientProtocol {
dag.addVertex(vertices[i]);
if (i > 0) {
EdgeProperty edgeProperty = new EdgeProperty(
- ConnectionPattern.BIPARTITE, SourceType.STABLE,
+ DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+ SchedulingType.SEQUENTIAL,
new OutputDescriptor(OnFileSortedOutput.class.getName()),
new InputDescriptor(ShuffledMergedInput.class.getName()));