You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2017/03/16 15:10:36 UTC
[3/3] flink git commit: [FLINK-5134] [runtime] Aggregate ResourceSpec
for chained operators when generating job graph
[FLINK-5134] [runtime] Aggregate ResourceSpec for chained operators when generating job graph
This closes #3455
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/980d072f
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/980d072f
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/980d072f
Branch: refs/heads/master
Commit: 980d072fa2546dbc10cf878cf29532b2d8bbca8a
Parents: 24408e1
Author: \u6dd8\u6c5f <ta...@alibaba-inc.com>
Authored: Thu Mar 2 13:51:39 2017 +0800
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Mar 16 14:43:27 2017 +0100
----------------------------------------------------------------------
.../flink/api/common/operators/Operator.java | 12 +-
.../api/common/operators/ResourceSpec.java | 2 +-
.../flink/api/java/operators/DataSink.java | 90 ++++----
.../api/java/operators/DeltaIteration.java | 88 ++++----
.../flink/api/java/operators/Operator.java | 96 +++++----
.../api/java/operators/OperatorTranslation.java | 12 +-
.../flink/api/java/operator/OperatorTest.java | 22 +-
.../plantranslate/JobGraphGenerator.java | 14 +-
.../plantranslate/JobGraphGeneratorTest.java | 209 +++++++++++++++++++
.../flink/runtime/jobgraph/JobVertex.java | 38 ++++
.../api/datastream/DataStreamSink.java | 79 +++----
.../datastream/SingleOutputStreamOperator.java | 78 +++----
.../flink/streaming/api/graph/StreamGraph.java | 10 +-
.../api/graph/StreamGraphGenerator.java | 10 +-
.../flink/streaming/api/graph/StreamNode.java | 4 +-
.../api/graph/StreamingJobGraphGenerator.java | 26 +++
.../transformations/StreamTransformation.java | 4 +-
.../flink/streaming/api/DataStreamTest.java | 68 +++---
.../graph/StreamingJobGraphGeneratorTest.java | 159 +++++++++++++-
19 files changed, 739 insertions(+), 282 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-core/src/main/java/org/apache/flink/api/common/operators/Operator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/Operator.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/Operator.java
index 1905555..6d906f2 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/Operator.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/Operator.java
@@ -28,8 +28,6 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Visitable;
-import javax.annotation.Nullable;
-
/**
* Abstract base class for all operators. An operator is a source, sink, or it applies an operation to
* one or more inputs, producing a result.
@@ -47,11 +45,9 @@ public abstract class Operator<OUT> implements Visitable<Operator<?>> {
private int parallelism = ExecutionConfig.PARALLELISM_DEFAULT; // the number of parallel instances to use
- @Nullable
- private ResourceSpec minResources; // the minimum resource of the contract instance.
+ private ResourceSpec minResources = ResourceSpec.DEFAULT; // the minimum resource of the contract instance.
- @Nullable
- private ResourceSpec preferredResources; // the preferred resource of the contract instance.
+ private ResourceSpec preferredResources = ResourceSpec.DEFAULT; // the preferred resource of the contract instance.
/**
* The return type of the user function.
@@ -199,7 +195,6 @@ public abstract class Operator<OUT> implements Visitable<Operator<?>> {
*
* @return The minimum resources of this operator.
*/
- @Nullable
@PublicEvolving
public ResourceSpec getMinResources() {
return this.minResources;
@@ -211,7 +206,6 @@ public abstract class Operator<OUT> implements Visitable<Operator<?>> {
*
* @return The preferred resource of this operator.
*/
- @Nullable
@PublicEvolving
public ResourceSpec getPreferredResources() {
return this.preferredResources;
@@ -225,7 +219,7 @@ public abstract class Operator<OUT> implements Visitable<Operator<?>> {
* @param preferredResources The preferred resource of this operator.
*/
@PublicEvolving
- public void setResource(ResourceSpec minResources, ResourceSpec preferredResources) {
+ public void setResources(ResourceSpec minResources, ResourceSpec preferredResources) {
this.minResources = minResources;
this.preferredResources = preferredResources;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-core/src/main/java/org/apache/flink/api/common/operators/ResourceSpec.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/ResourceSpec.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/ResourceSpec.java
index 0ea289a..cd3e5ad 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/ResourceSpec.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/ResourceSpec.java
@@ -44,7 +44,7 @@ public class ResourceSpec implements Serializable {
private static final long serialVersionUID = 1L;
- public static final ResourceSpec UNKNOWN = new ResourceSpec(0, 0, 0, 0, 0);
+ public static final ResourceSpec DEFAULT = new ResourceSpec(0, 0, 0, 0, 0);
/** How many cpu cores are needed, use double so we can specify cpu like 0.1 */
private final double cpuCores;
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-java/src/main/java/org/apache/flink/api/java/operators/DataSink.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/DataSink.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/DataSink.java
index 369e013..fd8190c 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/DataSink.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/DataSink.java
@@ -52,9 +52,9 @@ public class DataSink<T> {
private int parallelism = ExecutionConfig.PARALLELISM_DEFAULT;
- private ResourceSpec minResources = ResourceSpec.UNKNOWN;
+ private ResourceSpec minResources = ResourceSpec.DEFAULT;
- private ResourceSpec preferredResources = ResourceSpec.UNKNOWN;
+ private ResourceSpec preferredResources = ResourceSpec.DEFAULT;
private Configuration parameters;
@@ -306,49 +306,45 @@ public class DataSink<T> {
return this.preferredResources;
}
-// ---------------------------------------------------------------------------
-// Fine-grained resource profiles are an incomplete work-in-progress feature
-// The setters are hence commented out at this point.
-// ---------------------------------------------------------------------------
-//
-// /**
-// * Sets the minimum and preferred resources for this data sink. This overrides the default empty resource.
-// * The minimum resource must be satisfied and the preferred resource specifies the upper bound
-// * for dynamic resource resize.
-// *
-// * @param minResources The minimum resource for this data sink.
-// * @param preferredResources The preferred resource for this data sink.
-// * @return The data sink with set minimum and preferred resources.
-// */
-// @PublicEvolving
-// public DataSink<T> setResources(ResourceSpec minResources, ResourceSpec preferredResources) {
-// Preconditions.checkNotNull(minResources, "The min resources must be not null.");
-// Preconditions.checkNotNull(preferredResources, "The preferred resources must be not null.");
-// Preconditions.checkArgument(minResources.isValid() &&
-// preferredResources.isValid() && minResources.lessThanOrEqual(preferredResources),
-// "The values in resource must be not less than 0 and the preferred " +
-// "resource must be greater than the min resource.");
-//
-// this.minResources = minResources;
-// this.preferredResources = preferredResources;
-//
-// return this;
-// }
-//
-// /**
-// * Sets the resources for this data sink. This overrides the default resource profile.
-// *
-// * @param resources The resources for this data sink.
-// * @return The data sink with set minimum and preferred resources.
-// */
-// @PublicEvolving
-// public DataSink<T> setResources(ResourceSpec resources) {
-// Preconditions.checkNotNull(resources, "The resources must be not null.");
-// Preconditions.checkArgument(resources.isValid(), "The resource values must be greater than 0.");
-//
-// this.minResources = resources;
-// this.preferredResources = resources;
-//
-// return this;
-// }
+ // ---------------------------------------------------------------------------
+ // Fine-grained resource profiles are an incomplete work-in-progress feature
+ // The setters are hence private at this point.
+ // ---------------------------------------------------------------------------
+
+ /**
+ * Sets the minimum and preferred resources for this data sink. and the lower and upper resource limits
+ * will be considered in resource resize feature for future plan.
+ *
+ * @param minResources The minimum resources for this data sink.
+ * @param preferredResources The preferred resources for this data sink.
+ * @return The data sink with set minimum and preferred resources.
+ */
+ private DataSink<T> setResources(ResourceSpec minResources, ResourceSpec preferredResources) {
+ Preconditions.checkNotNull(minResources, "The min resources must be not null.");
+ Preconditions.checkNotNull(preferredResources, "The preferred resources must be not null.");
+ Preconditions.checkArgument(minResources.isValid() && preferredResources.isValid() && minResources.lessThanOrEqual(preferredResources),
+ "The values in resources must be not less than 0 and the preferred resources must be greater than the min resources.");
+
+ this.minResources = minResources;
+ this.preferredResources = preferredResources;
+
+ return this;
+ }
+
+ /**
+ * Sets the resources for this data sink, and the minimum and preferred resources are the same by default.
+ *
+ * @param resources The resources for this data sink.
+ * @return The data sink with set minimum and preferred resources.
+ */
+ private DataSink<T> setResources(ResourceSpec resources) {
+ Preconditions.checkNotNull(resources, "The resources must be not null.");
+ Preconditions.checkArgument(resources.isValid(), "The values in resources must be not less than 0.");
+
+ this.minResources = resources;
+ this.preferredResources = resources;
+
+ return this;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-java/src/main/java/org/apache/flink/api/java/operators/DeltaIteration.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/DeltaIteration.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/DeltaIteration.java
index 3d327e9..61f83b1 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/DeltaIteration.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/DeltaIteration.java
@@ -64,9 +64,9 @@ public class DeltaIteration<ST, WT> {
private int parallelism = ExecutionConfig.PARALLELISM_DEFAULT;
- private ResourceSpec minResources = ResourceSpec.UNKNOWN;
+ private ResourceSpec minResources = ResourceSpec.DEFAULT;
- private ResourceSpec preferredResources = ResourceSpec.UNKNOWN;
+ private ResourceSpec preferredResources = ResourceSpec.DEFAULT;
private boolean solutionSetUnManaged;
@@ -197,49 +197,47 @@ public class DeltaIteration<ST, WT> {
return parallelism;
}
-// ---------------------------------------------------------------------------
-// Fine-grained resource profiles are an incomplete work-in-progress feature
-// The setters are hence commented out at this point.
-// ---------------------------------------------------------------------------
-//
-// /**
-// * Sets the minimum and preferred resources for the iteration. This overrides the default empty resource.
-// * The lower and upper resource limits will be considered in dynamic resource resize feature for future plan.
-// *
-// * @param minResources The minimum resource for the iteration.
-// * @param preferredResources The preferred resource for the iteration.
-// * @return The iteration with set minimum and preferred resources.
-// */
-// @PublicEvolving
-// public DeltaIteration<ST, WT> setResource(ResourceSpec minResources, ResourceSpec preferredResources) {
-// Preconditions.checkNotNull(minResources, "The min resources must be not null.");
-// Preconditions.checkNotNull(preferredResources, "The preferred resources must be not null.");
-// Preconditions.checkArgument(minResources.isValid() && preferredResources.isValid() && minResources.lessThanOrEqual(preferredResources),
-// "The values in resources must be not less than 0 and the preferred resources must be greater than the min resources.");
-//
-// this.minResources = minResources;
-// this.preferredResources = preferredResources;
-//
-// return this;
-// }
-//
-// /**
-// * Sets the resource for the iteration, and the minimum and preferred resources are the same by default.
-// * The lower and upper resource limits will be considered in dynamic resource resize feature for future plan.
-// *
-// * @param resources The resource for the iteration.
-// * @return The iteration with set minimum and preferred resources.
-// */
-// @PublicEvolving
-// public DeltaIteration<ST, WT> setResource(ResourceSpec resources) {
-// Preconditions.checkNotNull(resources, "The resources must be not null.");
-// Preconditions.checkArgument(resources.isValid(), "The values in resource must be not less than 0.");
-//
-// this.minResources = resources;
-// this.preferredResources = resources;
-//
-// return this;
-// }
+ // ---------------------------------------------------------------------------
+ // Fine-grained resource profiles are an incomplete work-in-progress feature
+ // The setters are hence private at this point.
+ // ---------------------------------------------------------------------------
+
+ /**
+ * Sets the minimum and preferred resources for the iteration. This overrides the default resources.
+ * The lower and upper resource limits will be considered in dynamic resource resize feature for future plan.
+ *
+ * @param minResources The minimum resources for the iteration.
+ * @param preferredResources The preferred resources for the iteration.
+ * @return The iteration with set minimum and preferred resources.
+ */
+ private DeltaIteration<ST, WT> setResources(ResourceSpec minResources, ResourceSpec preferredResources) {
+ Preconditions.checkNotNull(minResources, "The min resources must be not null.");
+ Preconditions.checkNotNull(preferredResources, "The preferred resources must be not null.");
+ Preconditions.checkArgument(minResources.isValid() && preferredResources.isValid() && minResources.lessThanOrEqual(preferredResources),
+ "The values in resources must be not less than 0 and the preferred resources must be greater than the min resources.");
+
+ this.minResources = minResources;
+ this.preferredResources = preferredResources;
+
+ return this;
+ }
+
+ /**
+ * Sets the resources for the iteration, and the minimum and preferred resources are the same by default.
+ * The lower and upper resource limits will be considered in dynamic resource resize feature for future plan.
+ *
+ * @param resources The resources for the iteration.
+ * @return The iteration with set minimum and preferred resources.
+ */
+ private DeltaIteration<ST, WT> setResources(ResourceSpec resources) {
+ Preconditions.checkNotNull(resources, "The resources must be not null.");
+ Preconditions.checkArgument(resources.isValid(), "The values in resources must be not less than 0.");
+
+ this.minResources = resources;
+ this.preferredResources = resources;
+
+ return this;
+ }
/**
* Gets the minimum resources from this iteration. If no minimum resources have been set,
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-java/src/main/java/org/apache/flink/api/java/operators/Operator.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/Operator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/Operator.java
index 6ae59dd..e496c62 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/Operator.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/Operator.java
@@ -39,9 +39,9 @@ public abstract class Operator<OUT, O extends Operator<OUT, O>> extends DataSet<
protected int parallelism = ExecutionConfig.PARALLELISM_DEFAULT;
- protected ResourceSpec minResources = ResourceSpec.UNKNOWN;
+ protected ResourceSpec minResources = ResourceSpec.DEFAULT;
- protected ResourceSpec preferredResources = ResourceSpec.UNKNOWN;
+ protected ResourceSpec preferredResources = ResourceSpec.DEFAULT;
protected Operator(ExecutionEnvironment context, TypeInformation<OUT> resultType) {
@@ -130,51 +130,49 @@ public abstract class Operator<OUT, O extends Operator<OUT, O>> extends DataSet<
return returnType;
}
-// ---------------------------------------------------------------------------
-// Fine-grained resource profiles are an incomplete work-in-progress feature
-// The setters are hence commented out at this point.
-// ---------------------------------------------------------------------------
-//
-// /**
-// * Sets the minimum and preferred resources for this operator. This overrides the default empty resource.
-// * The lower and upper resource limits will be considered in dynamic resource resize feature for future plan.
-// *
-// * @param minResources The minimum resource for this operator.
-// * @param preferredResources The preferred resource for this operator.
-// * @return The operator with set minimum and preferred resources.
-// */
-// @PublicEvolving
-// public O setResources(ResourceSpec minResources, ResourceSpec preferredResources) {
-// Preconditions.checkNotNull(minResources, "The min resources must be not null.");
-// Preconditions.checkNotNull(preferredResources, "The preferred resources must be not null.");
-//
-// Preconditions.checkArgument(minResources.isValid() && preferredResources.isValid() && minResources.lessThanOrEqual(preferredResources),
-// "The values in resource must be not less than 0 and the preferred resource must be greater than the min resource.");
-//
-// this.minResources = minResources;
-// this.preferredResources = preferredResources;
-//
-// @SuppressWarnings("unchecked")
-// O returnType = (O) this;
-// return returnType;
-// }
-//
-// /**
-// * Sets the resources for this operator. This overrides the default minimum and preferred resources.
-// *
-// * @param resources The resource for this operator.
-// * @return The operator with set minimum and preferred resources.
-// */
-// @PublicEvolving
-// public O setResources(ResourceSpec resources) {
-// Preconditions.checkNotNull(resources, "The resource must be not null.");
-// Preconditions.checkArgument(resources.isValid(), "The resource values must be greater than 0.");
-//
-// this.minResources = resources;
-// this.preferredResources = resources;
-//
-// @SuppressWarnings("unchecked")
-// O returnType = (O) this;
-// return returnType;
-// }
+ // ---------------------------------------------------------------------------
+ // Fine-grained resource profiles are an incomplete work-in-progress feature
+ // The setters are hence private at this point.
+ // ---------------------------------------------------------------------------
+
+ /**
+ * Sets the minimum and preferred resources for this operator. This overrides the default resources.
+ * The lower and upper resource limits will be considered in dynamic resource resize feature for future plan.
+ *
+ * @param minResources The minimum resources for this operator.
+ * @param preferredResources The preferred resources for this operator.
+ * @return The operator with set minimum and preferred resources.
+ */
+ private O setResources(ResourceSpec minResources, ResourceSpec preferredResources) {
+ Preconditions.checkNotNull(minResources, "The min resources must be not null.");
+ Preconditions.checkNotNull(preferredResources, "The preferred resources must be not null.");
+
+ Preconditions.checkArgument(minResources.isValid() && preferredResources.isValid() && minResources.lessThanOrEqual(preferredResources),
+ "The values in resources must be not less than 0 and the preferred resources must be greater than the min resources.");
+
+ this.minResources = minResources;
+ this.preferredResources = preferredResources;
+
+ @SuppressWarnings("unchecked")
+ O returnType = (O) this;
+ return returnType;
+ }
+
+ /**
+ * Sets the resources for this operator. This overrides the default minimum and preferred resources.
+ *
+ * @param resources The resources for this operator.
+ * @return The operator with set minimum and preferred resources.
+ */
+ private O setResources(ResourceSpec resources) {
+ Preconditions.checkNotNull(resources, "The resources must be not null.");
+ Preconditions.checkArgument(resources.isValid(), "The values in resources must be not less than 0.");
+
+ this.minResources = resources;
+ this.preferredResources = resources;
+
+ @SuppressWarnings("unchecked")
+ O returnType = (O) this;
+ return returnType;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-java/src/main/java/org/apache/flink/api/java/operators/OperatorTranslation.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/OperatorTranslation.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/OperatorTranslation.java
index 3bffd8b..cdbec71 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/OperatorTranslation.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/OperatorTranslation.java
@@ -64,7 +64,7 @@ public class OperatorTranslation {
// translate the sink itself and connect it to the input
GenericDataSinkBase<T> translatedSink = sink.translateToDataFlow(input);
- translatedSink.setResource(sink.getMinResources(), sink.getPreferredResources());
+ translatedSink.setResources(sink.getMinResources(), sink.getPreferredResources());
return translatedSink;
}
@@ -95,28 +95,28 @@ public class OperatorTranslation {
if (dataSet instanceof DataSource) {
DataSource<T> dataSource = (DataSource<T>) dataSet;
dataFlowOp = dataSource.translateToDataFlow();
- dataFlowOp.setResource(dataSource.getMinResources(), dataSource.getPreferredResources());
+ dataFlowOp.setResources(dataSource.getMinResources(), dataSource.getPreferredResources());
}
else if (dataSet instanceof SingleInputOperator) {
SingleInputOperator<?, ?, ?> singleInputOperator = (SingleInputOperator<?, ?, ?>) dataSet;
dataFlowOp = translateSingleInputOperator(singleInputOperator);
- dataFlowOp.setResource(singleInputOperator.getMinResources(), singleInputOperator.getPreferredResources());
+ dataFlowOp.setResources(singleInputOperator.getMinResources(), singleInputOperator.getPreferredResources());
}
else if (dataSet instanceof TwoInputOperator) {
TwoInputOperator<?, ?, ?, ?> twoInputOperator = (TwoInputOperator<?, ?, ?, ?>) dataSet;
dataFlowOp = translateTwoInputOperator(twoInputOperator);
- dataFlowOp.setResource(twoInputOperator.getMinResources(), twoInputOperator.getPreferredResources());
+ dataFlowOp.setResources(twoInputOperator.getMinResources(), twoInputOperator.getPreferredResources());
}
else if (dataSet instanceof BulkIterationResultSet) {
BulkIterationResultSet<?> bulkIterationResultSet = (BulkIterationResultSet<?>) dataSet;
dataFlowOp = translateBulkIteration(bulkIterationResultSet);
- dataFlowOp.setResource(bulkIterationResultSet.getIterationHead().getMinResources(),
+ dataFlowOp.setResources(bulkIterationResultSet.getIterationHead().getMinResources(),
bulkIterationResultSet.getIterationHead().getPreferredResources());
}
else if (dataSet instanceof DeltaIterationResultSet) {
DeltaIterationResultSet<?, ?> deltaIterationResultSet = (DeltaIterationResultSet<?, ?>) dataSet;
dataFlowOp = translateDeltaIteration(deltaIterationResultSet);
- dataFlowOp.setResource(deltaIterationResultSet.getIterationHead().getMinResources(),
+ dataFlowOp.setResources(deltaIterationResultSet.getIterationHead().getMinResources(),
deltaIterationResultSet.getIterationHead().getPreferredResources());
}
else if (dataSet instanceof DeltaIteration.SolutionSetPlaceHolder || dataSet instanceof DeltaIteration.WorksetPlaceHolder) {
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-java/src/test/java/org/apache/flink/api/java/operator/OperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/operator/OperatorTest.java b/flink-java/src/test/java/org/apache/flink/api/java/operator/OperatorTest.java
index 992acc9..3de254b 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/operator/OperatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/operator/OperatorTest.java
@@ -25,6 +25,8 @@ import org.apache.flink.api.java.operators.Operator;
import org.apache.flink.api.java.typeutils.ValueTypeInfo;
import org.junit.Test;
+import java.lang.reflect.Method;
+
import static org.junit.Assert.assertEquals;
public class OperatorTest {
@@ -46,19 +48,21 @@ public class OperatorTest {
assertEquals(parallelism, operator.getParallelism());
}
- /*
@Test
- public void testConfigurationOfResource() {
+ public void testConfigurationOfResource() throws Exception{
Operator operator = new MockOperator();
- // verify explicit change in resource
- ResourceSpec minResource = new ResourceSpec(1.0, 100, 0, 0, 0);
- ResourceSpec preferredResource = new ResourceSpec(2.0, 200, 0, 0, 0);
- operator.setResource(minResource, preferredResource);
+ Method opMethod = Operator.class.getDeclaredMethod("setResources", ResourceSpec.class, ResourceSpec.class);
+ opMethod.setAccessible(true);
+
+ // verify explicit change in resources
+ ResourceSpec minResources = new ResourceSpec(1.0, 100);
+ ResourceSpec preferredResources = new ResourceSpec(2.0, 200);
+ opMethod.invoke(operator, minResources, preferredResources);
- assertEquals(minResource, operator.getMinResource());
- assertEquals(preferredResource, operator.getPreferredResource());
- }*/
+ assertEquals(minResources, operator.getMinResources());
+ assertEquals(preferredResources, operator.getPreferredResources());
+ }
private class MockOperator extends Operator {
public MockOperator() {
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java
index bbc944e..a407bfe 100644
--- a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java
@@ -538,6 +538,10 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
} else {
container.setName("CHAIN " + containerTaskName + " -> " + chainedTask.getTaskName());
}
+
+ //update resource of container task
+ container.setResources(container.getMinResources().merge(node.getMinResources()),
+ container.getPreferredResources().merge(node.getPreferredResources()));
this.chainedTasksInSequence.add(chainedTask);
return;
@@ -837,6 +841,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
} else {
// create task vertex
vertex = new JobVertex(taskName);
+ vertex.setResources(node.getMinResources(), node.getPreferredResources());
vertex.setInvokableClass((this.currentIteration != null && node.isOnDynamicPath()) ? IterationIntermediateTask.class : BatchTask.class);
config = new TaskConfig(vertex.getConfiguration());
@@ -862,6 +867,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
final DriverStrategy ds = node.getDriverStrategy();
final JobVertex vertex = new JobVertex(taskName);
final TaskConfig config = new TaskConfig(vertex.getConfiguration());
+ vertex.setResources(node.getMinResources(), node.getPreferredResources());
vertex.setInvokableClass( (this.currentIteration != null && node.isOnDynamicPath()) ? IterationIntermediateTask.class : BatchTask.class);
// set user code
@@ -890,6 +896,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
final InputFormatVertex vertex = new InputFormatVertex(node.getNodeName());
final TaskConfig config = new TaskConfig(vertex.getConfiguration());
+ vertex.setResources(node.getMinResources(), node.getPreferredResources());
vertex.setInvokableClass(DataSourceTask.class);
vertex.setFormatDescription(getDescriptionForUserCode(node.getProgramOperator().getUserCodeWrapper()));
@@ -905,6 +912,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
final OutputFormatVertex vertex = new OutputFormatVertex(node.getNodeName());
final TaskConfig config = new TaskConfig(vertex.getConfiguration());
+ vertex.setResources(node.getMinResources(), node.getPreferredResources());
vertex.setInvokableClass(DataSinkTask.class);
vertex.setFormatDescription(getDescriptionForUserCode(node.getProgramOperator().getUserCodeWrapper()));
@@ -967,6 +975,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
// everything else happens in the post visit, after the input (the initial partial solution)
// is connected.
headVertex = new JobVertex("PartialSolution ("+iteration.getNodeName()+")");
+ headVertex.setResources(iteration.getMinResources(), iteration.getPreferredResources());
headVertex.setInvokableClass(IterationHeadTask.class);
headConfig = new TaskConfig(headVertex.getConfiguration());
headConfig.setDriver(NoOpDriver.class);
@@ -1035,6 +1044,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
// everything else happens in the post visit, after the input (the initial partial solution)
// is connected.
headVertex = new JobVertex("IterationHead("+iteration.getNodeName()+")");
+ headVertex.setResources(iteration.getMinResources(), iteration.getPreferredResources());
headVertex.setInvokableClass(IterationHeadTask.class);
headConfig = new TaskConfig(headVertex.getConfiguration());
headConfig.setDriver(NoOpDriver.class);
@@ -1275,6 +1285,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
// --------------------------- create the sync task ---------------------------
final JobVertex sync = new JobVertex("Sync(" + bulkNode.getNodeName() + ")");
+ sync.setResources(bulkNode.getMinResources(), bulkNode.getPreferredResources());
sync.setInvokableClass(IterationSynchronizationSinkTask.class);
sync.setParallelism(1);
sync.setMaxParallelism(1);
@@ -1412,6 +1423,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
final TaskConfig syncConfig;
{
final JobVertex sync = new JobVertex("Sync (" + iterNode.getNodeName() + ")");
+ sync.setResources(iterNode.getMinResources(), iterNode.getPreferredResources());
sync.setInvokableClass(IterationSynchronizationSinkTask.class);
sync.setParallelism(1);
sync.setMaxParallelism(1);
@@ -1468,7 +1480,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
if (hasWorksetTail) {
nextWorksetVertex.setInvokableClass(IterationTailTask.class);
-
+
worksetTailConfig.setOutputSerializer(iterNode.getWorksetSerializer());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-optimizer/src/test/java/org/apache/flink/optimizer/plantranslate/JobGraphGeneratorTest.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/plantranslate/JobGraphGeneratorTest.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/plantranslate/JobGraphGeneratorTest.java
new file mode 100644
index 0000000..d8ee80b
--- /dev/null
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/plantranslate/JobGraphGeneratorTest.java
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.optimizer.plantranslate;
+
+import org.apache.flink.api.common.Plan;
+import org.apache.flink.api.common.aggregators.LongSumAggregator;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.operators.ResourceSpec;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.api.java.operators.DataSink;
+import org.apache.flink.api.java.operators.DeltaIteration;
+import org.apache.flink.api.java.operators.IterativeDataSet;
+import org.apache.flink.api.java.operators.Operator;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.optimizer.Optimizer;
+import org.apache.flink.optimizer.plan.OptimizedPlan;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.junit.Test;
+
+import java.lang.reflect.Method;
+
+import static org.junit.Assert.assertTrue;
+
+public class JobGraphGeneratorTest {
+
+ /**
+ * Verifies that the resources are merged correctly for chained operators when
+ * generating job graph
+ */
+ @Test
+ public void testResourcesForChainedOperators() throws Exception {
+ ResourceSpec resource1 = new ResourceSpec(0.1, 100);
+ ResourceSpec resource2 = new ResourceSpec(0.2, 200);
+ ResourceSpec resource3 = new ResourceSpec(0.3, 300);
+ ResourceSpec resource4 = new ResourceSpec(0.4, 400);
+ ResourceSpec resource5 = new ResourceSpec(0.5, 500);
+ ResourceSpec resource6 = new ResourceSpec(0.6, 600);
+ ResourceSpec resource7 = new ResourceSpec(0.7, 700);
+
+ Method opMethod = Operator.class.getDeclaredMethod("setResources", ResourceSpec.class);
+ opMethod.setAccessible(true);
+
+ Method sinkMethod = DataSink.class.getDeclaredMethod("setResources", ResourceSpec.class);
+ sinkMethod.setAccessible(true);
+
+ MapFunction<Long, Long> mapFunction = new MapFunction<Long, Long>() {
+ @Override
+ public Long map(Long value) throws Exception {
+ return value;
+ }
+ };
+
+ FilterFunction<Long> filterFunction = new FilterFunction<Long>() {
+ @Override
+ public boolean filter(Long value) throws Exception {
+ return false;
+ }
+ };
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Long> input = env.fromElements(1L, 2L, 3L);
+ opMethod.invoke(input, resource1);
+
+ DataSet<Long> map1 = input.map(mapFunction);
+ opMethod.invoke(map1, resource2);
+
+ // CHAIN(Source -> Map -> Filter)
+ DataSet<Long> filter1 = map1.filter(filterFunction);
+ opMethod.invoke(filter1, resource3);
+
+ IterativeDataSet<Long> startOfIteration = filter1.iterate(10);
+ opMethod.invoke(startOfIteration, resource4);
+
+ DataSet<Long> map2 = startOfIteration.map(mapFunction);
+ opMethod.invoke(map2, resource5);
+
+ // CHAIN(Map -> Filter)
+ DataSet<Long> feedback = map2.filter(filterFunction);
+ opMethod.invoke(feedback, resource6);
+
+ DataSink<Long> sink = startOfIteration.closeWith(feedback).output(new DiscardingOutputFormat<Long>());
+ sinkMethod.invoke(sink, resource7);
+
+ Plan plan = env.createProgramPlan();
+ Optimizer pc = new Optimizer(new Configuration());
+ OptimizedPlan op = pc.compile(plan);
+
+ JobGraphGenerator jgg = new JobGraphGenerator();
+ JobGraph jobGraph = jgg.compileJobGraph(op);
+
+ JobVertex sourceMapFilterVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0);
+ JobVertex iterationHeadVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(1);
+ JobVertex feedbackVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(2);
+ JobVertex sinkVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(3);
+ JobVertex iterationSyncVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(4);
+
+ assertTrue(sourceMapFilterVertex.getMinResources().equals(resource1.merge(resource2).merge(resource3)));
+ assertTrue(iterationHeadVertex.getPreferredResources().equals(resource4));
+ assertTrue(feedbackVertex.getMinResources().equals(resource5.merge(resource6)));
+ assertTrue(sinkVertex.getPreferredResources().equals(resource7));
+ assertTrue(iterationSyncVertex.getMinResources().equals(resource4));
+ }
+
+ /**
+ * Verifies that the resources are set onto each job vertex correctly when generating job graph
+ * which covers the delta iteration case
+ */
+ @Test
+ public void testResourcesForDeltaIteration() throws Exception{
+ ResourceSpec resource1 = new ResourceSpec(0.1, 100);
+ ResourceSpec resource2 = new ResourceSpec(0.2, 200);
+ ResourceSpec resource3 = new ResourceSpec(0.3, 300);
+ ResourceSpec resource4 = new ResourceSpec(0.4, 400);
+ ResourceSpec resource5 = new ResourceSpec(0.5, 500);
+ ResourceSpec resource6 = new ResourceSpec(0.6, 600);
+
+ Method opMethod = Operator.class.getDeclaredMethod("setResources", ResourceSpec.class);
+ opMethod.setAccessible(true);
+
+ Method deltaMethod = DeltaIteration.class.getDeclaredMethod("setResources", ResourceSpec.class);
+ deltaMethod.setAccessible(true);
+
+ Method sinkMethod = DataSink.class.getDeclaredMethod("setResources", ResourceSpec.class);
+ sinkMethod.setAccessible(true);
+
+ MapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> mapFunction = new MapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>() {
+ @Override
+ public Tuple2<Long, Long> map(Tuple2<Long, Long> value) throws Exception {
+ return value;
+ }
+ };
+
+ FilterFunction<Tuple2<Long, Long>> filterFunction = new FilterFunction<Tuple2<Long, Long>>() {
+ @Override
+ public boolean filter(Tuple2<Long, Long> value) throws Exception {
+ return false;
+ }
+ };
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ DataSet<Tuple2<Long, Long>> input = env.fromElements(new Tuple2<>(1L, 2L));
+ opMethod.invoke(input, resource1);
+
+ // CHAIN(Map -> Filter)
+ DataSet<Tuple2<Long, Long>> map = input.map(mapFunction);
+ opMethod.invoke(map, resource2);
+
+ DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration = map.iterateDelta(map, 100, 0).
+ registerAggregator("test", new LongSumAggregator());
+ deltaMethod.invoke(iteration, resource3);
+
+ DataSet<Tuple2<Long, Long>> delta = iteration.getWorkset().map(mapFunction);
+ opMethod.invoke(delta, resource4);
+
+ DataSet<Tuple2<Long, Long>> feedback = delta.filter(filterFunction);
+ opMethod.invoke(feedback, resource5);
+
+ DataSink<Tuple2<Long, Long>> sink = iteration.closeWith(delta, feedback).
+ output(new DiscardingOutputFormat<Tuple2<Long, Long>>());
+ sinkMethod.invoke(sink, resource6);
+
+ Plan plan = env.createProgramPlan();
+ Optimizer pc = new Optimizer(new Configuration());
+ OptimizedPlan op = pc.compile(plan);
+
+ JobGraphGenerator jgg = new JobGraphGenerator();
+ JobGraph jobGraph = jgg.compileJobGraph(op);
+
+ JobVertex sourceMapVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0);
+ JobVertex iterationHeadVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(1);
+ JobVertex deltaVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(2);
+ JobVertex iterationTailVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(3);
+ JobVertex feedbackVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(4);
+ JobVertex sinkVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(5);
+ JobVertex iterationSyncVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(6);
+
+ assertTrue(sourceMapVertex.getMinResources().equals(resource1.merge(resource2)));
+ assertTrue(iterationHeadVertex.getPreferredResources().equals(resource3));
+ assertTrue(deltaVertex.getMinResources().equals(resource4));
+ // the iteration tail task will be scheduled in the same instance with iteration head, and currently not set resources.
+ assertTrue(iterationTailVertex.getPreferredResources().equals(ResourceSpec.DEFAULT));
+ assertTrue(feedbackVertex.getMinResources().equals(resource5));
+ assertTrue(sinkVertex.getPreferredResources().equals(resource6));
+ assertTrue(iterationSyncVertex.getMinResources().equals(resource3));
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
index 260bd74..1180db4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.jobgraph;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.operators.ResourceSpec;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.InputSplitSource;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
@@ -31,6 +32,8 @@ import org.apache.flink.util.Preconditions;
import java.util.ArrayList;
import java.util.List;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
/**
* The base class for job vertexes.
*/
@@ -61,6 +64,12 @@ public class JobVertex implements java.io.Serializable {
/** Maximum number of subtasks to split this taks into a runtime. */
private int maxParallelism = -1;
+ /** The minimum resource of the vertex */
+ private ResourceSpec minResources = ResourceSpec.DEFAULT;
+
+ /** The preferred resource of the vertex */
+ private ResourceSpec preferredResources = ResourceSpec.DEFAULT;
+
/** Custom configuration passed to the assigned task at runtime. */
private Configuration configuration;
@@ -278,6 +287,35 @@ public class JobVertex implements java.io.Serializable {
this.maxParallelism = maxParallelism;
}
+ /**
+ * Gets the minimum resource for the task.
+ *
+ * @return The minimum resource for the task.
+ */
+ public ResourceSpec getMinResources() {
+ return minResources;
+ }
+
+ /**
+ * Gets the preferred resource for the task.
+ *
+ * @return The preferred resource for the task.
+ */
+ public ResourceSpec getPreferredResources() {
+ return preferredResources;
+ }
+
+ /**
+ * Sets the minimum and preferred resources for the task.
+ *
+ * @param minResources The minimum resource for the task.
+ * @param preferredResources The preferred resource for the task.
+ */
+ public void setResources(ResourceSpec minResources, ResourceSpec preferredResources) {
+ this.minResources = checkNotNull(minResources);
+ this.preferredResources = checkNotNull(preferredResources);
+ }
+
public InputSplitSource<?> getInputSplitSource() {
return inputSplitSource;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamSink.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamSink.java
index 39d81c6..b0e78d7 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamSink.java
@@ -20,9 +20,11 @@ package org.apache.flink.streaming.api.datastream;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.Public;
+import org.apache.flink.api.common.operators.ResourceSpec;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.StreamSink;
import org.apache.flink.streaming.api.transformations.SinkTransformation;
+import org.apache.flink.util.Preconditions;
/**
* A Stream Sink. This is used for emitting elements from a streaming topology.
@@ -113,45 +115,44 @@ public class DataStreamSink<T> {
return this;
}
-// ---------------------------------------------------------------------------
-// Fine-grained resource profiles are an incomplete work-in-progress feature
-// The setters are hence commented out at this point.
-// ---------------------------------------------------------------------------
-// /**
-// * Sets the minimum and preferred resources for this sink, and the lower and upper resource limits will
-// * be considered in resource resize feature for future plan.
-// *
-// * @param minResources The minimum resources for this sink.
-// * @param preferredResources The preferred resources for this sink
-// * @return The sink with set minimum and preferred resources.
-// */
-// @PublicEvolving
-// public DataStreamSink<T> setResources(ResourceSpec minResources, ResourceSpec preferredResources) {
-// Preconditions.checkNotNull(minResources, "The min resources must be not null.");
-// Preconditions.checkNotNull(preferredResources, "The preferred resources must be not null.");
-// Preconditions.checkArgument(minResources.isValid() && preferredResources.isValid() && minResources.lessThanOrEqual(preferredResources),
-// "The values in resource must be not less than 0 and the preferred resource must be greater than the min resource.");
-//
-// transformation.setResources(minResources, preferredResources);
-//
-// return this;
-// }
-//
-// /**
-// * Sets the resource for this sink, the minimum and preferred resources are the same by default.
-// *
-// * @param resources The resource for this sink.
-// * @return The sink with set minimum and preferred resources.
-// */
-// @PublicEvolving
-// public DataStreamSink<T> setResources(ResourceSpec resources) {
-// Preconditions.checkNotNull(resources, "The resource must be not null.");
-// Preconditions.checkArgument(resources.isValid(), "The resource values must be greater than 0.");
-//
-// transformation.setResources(resources, resources);
-//
-// return this;
-// }
+ // ---------------------------------------------------------------------------
+ // Fine-grained resource profiles are an incomplete work-in-progress feature
+ // The setters are hence private at this point.
+ // ---------------------------------------------------------------------------
+
+ /**
+ * Sets the minimum and preferred resources for this sink, and the lower and upper resource limits will
+ * be considered in resource resize feature for future plan.
+ *
+ * @param minResources The minimum resources for this sink.
+ * @param preferredResources The preferred resources for this sink
+ * @return The sink with set minimum and preferred resources.
+ */
+ private DataStreamSink<T> setResources(ResourceSpec minResources, ResourceSpec preferredResources) {
+ Preconditions.checkNotNull(minResources, "The min resources must be not null.");
+ Preconditions.checkNotNull(preferredResources, "The preferred resources must be not null.");
+ Preconditions.checkArgument(minResources.isValid() && preferredResources.isValid() && minResources.lessThanOrEqual(preferredResources),
+ "The values in resources must be not less than 0 and the preferred resources must be greater than the min resources.");
+
+ transformation.setResources(minResources, preferredResources);
+
+ return this;
+ }
+
+ /**
+ * Sets the resources for this sink, the minimum and preferred resources are the same by default.
+ *
+ * @param resources The resources for this sink.
+ * @return The sink with set minimum and preferred resources.
+ */
+ private DataStreamSink<T> setResources(ResourceSpec resources) {
+ Preconditions.checkNotNull(resources, "The resources must be not null.");
+ Preconditions.checkArgument(resources.isValid(), "The values in resources must be not less than 0.");
+
+ transformation.setResources(resources, resources);
+
+ return this;
+ }
/**
* Turns off chaining for this operator so thread co-location will not be
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java
index 859c6d5..f1d5e3a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.api.datastream;
import org.apache.flink.annotation.Public;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.functions.InvalidTypesException;
+import org.apache.flink.api.common.operators.ResourceSpec;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
@@ -154,45 +155,44 @@ public class SingleOutputStreamOperator<T> extends DataStream<T> {
return this;
}
-// ---------------------------------------------------------------------------
-// Fine-grained resource profiles are an incomplete work-in-progress feature
-// The setters are hence commented out at this point.
-// ---------------------------------------------------------------------------
-// /**
-// * Sets the minimum and preferred resources for this operator, and the lower and upper resource limits will
-// * be considered in dynamic resource resize feature for future plan.
-// *
-// * @param minResources The minimum resources for this operator.
-// * @param preferredResources The preferred resources for this operator.
-// * @return The operator with set minimum and preferred resources.
-// */
-// @PublicEvolving
-// public SingleOutputStreamOperator<T> setResources(ResourceSpec minResources, ResourceSpec preferredResources) {
-// Preconditions.checkNotNull(minResources, "The min resources must be not null.");
-// Preconditions.checkNotNull(preferredResources, "The preferred resources must be not null.");
-// Preconditions.checkArgument(minResources.isValid() && preferredResources.isValid() && minResource.lessThanOrEqual(preferredResources),
-// "The values in resource must be not less than 0 and the preferred resource must be greater than the min resource.");
-//
-// transformation.setResources(minResources, preferredResources);
-//
-// return this;
-// }
-//
-// /**
-// * Sets the resources for this operator, the minimum and preferred resources are the same by default.
-// *
-// * @param resources The resources for this operator.
-// * @return The operator with set minimum and preferred resources.
-// */
-// @PublicEvolving
-// public SingleOutputStreamOperator<T> setResources(ResourceSpec resources) {
-// Preconditions.checkNotNull(resources, "The resource must be not null.");
-// Preconditions.checkArgument(resources.isValid(), "The resource values must be greater than 0.");
-//
-// transformation.setResources(resources, resources);
-//
-// return this;
-// }
+ // ---------------------------------------------------------------------------
+ // Fine-grained resource profiles are an incomplete work-in-progress feature
+ // The setters are hence private at this point.
+ // ---------------------------------------------------------------------------
+
+ /**
+ * Sets the minimum and preferred resources for this operator, and the lower and upper resource limits will
+ * be considered in dynamic resource resize feature for future plan.
+ *
+ * @param minResources The minimum resources for this operator.
+ * @param preferredResources The preferred resources for this operator.
+ * @return The operator with set minimum and preferred resources.
+ */
+ private SingleOutputStreamOperator<T> setResources(ResourceSpec minResources, ResourceSpec preferredResources) {
+ Preconditions.checkNotNull(minResources, "The min resources must be not null.");
+ Preconditions.checkNotNull(preferredResources, "The preferred resources must be not null.");
+ Preconditions.checkArgument(minResources.isValid() && preferredResources.isValid() && minResources.lessThanOrEqual(preferredResources),
+ "The values in resources must be not less than 0 and the preferred resources must be greater than the min resources.");
+
+ transformation.setResources(minResources, preferredResources);
+
+ return this;
+ }
+
+ /**
+ * Sets the resources for this operator, the minimum and preferred resources are the same by default.
+ *
+ * @param resources The resources for this operator.
+ * @return The operator with set minimum and preferred resources.
+ */
+ private SingleOutputStreamOperator<T> setResources(ResourceSpec resources) {
+ Preconditions.checkNotNull(resources, "The resources must be not null.");
+ Preconditions.checkArgument(resources.isValid(), "The values in resources must be not less than 0.");
+
+ transformation.setResources(resources, resources);
+
+ return this;
+ }
private boolean canBeParallel() {
return !nonParallel;
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
index a87e63d..c1775e4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
@@ -414,9 +414,9 @@ public class StreamGraph extends StreamingPlan {
}
}
- public void setResource(int vertexID, ResourceSpec minResource, ResourceSpec preferredResource) {
+ public void setResources(int vertexID, ResourceSpec minResources, ResourceSpec preferredResources) {
if (getStreamNode(vertexID) != null) {
- getStreamNode(vertexID).setResources(minResource, preferredResource);
+ getStreamNode(vertexID).setResources(minResources, preferredResources);
}
}
@@ -477,6 +477,7 @@ public class StreamGraph extends StreamingPlan {
StreamNode node = streamNodes.get(nodeId);
if (node != null) {
node.setUserHash(nodeHash);
+
}
}
@@ -540,7 +541,9 @@ public class StreamGraph extends StreamingPlan {
int sinkId,
long timeout,
int parallelism,
- int maxParallelism) {
+ int maxParallelism,
+ ResourceSpec minResources,
+ ResourceSpec preferredResources) {
StreamNode source = this.addNode(sourceId,
null,
StreamIterationHead.class,
@@ -549,6 +552,7 @@ public class StreamGraph extends StreamingPlan {
sources.add(source.getId());
setParallelism(source.getId(), parallelism);
setMaxParallelism(source.getId(), maxParallelism);
+ setResources(source.getId(), minResources, preferredResources);
StreamNode sink = this.addNode(sinkId,
null,
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
index de87a66..f4d4071 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
@@ -203,7 +203,7 @@ public class StreamGraphGenerator {
}
if (transform.getMinResources() != null && transform.getPreferredResources() != null) {
- streamGraph.setResource(transform.getId(), transform.getMinResources(), transform.getPreferredResources());
+ streamGraph.setResources(transform.getId(), transform.getMinResources(), transform.getPreferredResources());
}
return transformedIds;
@@ -335,7 +335,9 @@ public class StreamGraphGenerator {
getNewIterationNodeId(),
iterate.getWaitTime(),
iterate.getParallelism(),
- iterate.getMaxParallelism());
+ iterate.getMaxParallelism(),
+ iterate.getMinResources(),
+ iterate.getPreferredResources() );
StreamNode itSource = itSourceAndSink.f0;
StreamNode itSink = itSourceAndSink.f1;
@@ -400,7 +402,9 @@ public class StreamGraphGenerator {
getNewIterationNodeId(),
coIterate.getWaitTime(),
coIterate.getParallelism(),
- coIterate.getMaxParallelism());
+ coIterate.getMaxParallelism(),
+ coIterate.getMinResources(),
+ coIterate.getPreferredResources());
StreamNode itSource = itSourceAndSink.f0;
StreamNode itSink = itSourceAndSink.f1;
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
index 2d2e1e75..e4164ba 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
@@ -49,8 +49,8 @@ public class StreamNode implements Serializable {
* dynamic scaling and the number of key groups used for partitioned state.
*/
private int maxParallelism;
- private ResourceSpec minResources;
- private ResourceSpec preferredResources;
+ private ResourceSpec minResources = ResourceSpec.DEFAULT;
+ private ResourceSpec preferredResources = ResourceSpec.DEFAULT;
private Long bufferTimeout = null;
private final String operatorName;
private String slotSharingGroup;
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
index c18b527..5c1e1ac 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -19,6 +19,7 @@ package org.apache.flink.streaming.api.graph;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.operators.ResourceSpec;
import org.apache.flink.api.common.operators.util.UserCodeObjectWrapper;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -85,6 +86,9 @@ public class StreamingJobGraphGenerator {
private Map<Integer, StreamConfig> vertexConfigs;
private Map<Integer, String> chainedNames;
+ private Map<Integer, ResourceSpec> chainedMinResources;
+ private Map<Integer, ResourceSpec> chainedPreferredResources;
+
private final StreamGraphHasher defaultStreamGraphHasher;
private final List<StreamGraphHasher> legacyStreamGraphHashers;
@@ -100,6 +104,8 @@ public class StreamingJobGraphGenerator {
this.chainedConfigs = new HashMap<>();
this.vertexConfigs = new HashMap<>();
this.chainedNames = new HashMap<>();
+ this.chainedMinResources = new HashMap<>();
+ this.chainedPreferredResources = new HashMap<>();
this.physicalEdgesInOrder = new ArrayList<>();
}
@@ -211,6 +217,8 @@ public class StreamingJobGraphGenerator {
}
chainedNames.put(currentNodeId, createChainedName(currentNodeId, chainableOutputs));
+ chainedMinResources.put(currentNodeId, createChainedMinResources(currentNodeId, chainableOutputs));
+ chainedPreferredResources.put(currentNodeId, createChainedPreferredResources(currentNodeId, chainableOutputs));
StreamConfig config = currentNodeId.equals(startNodeId)
? createJobVertex(startNodeId, hashes, legacyHashes)
@@ -269,6 +277,22 @@ public class StreamingJobGraphGenerator {
}
}
+ private ResourceSpec createChainedMinResources(Integer vertexID, List<StreamEdge> chainedOutputs) {
+ ResourceSpec minResources = streamGraph.getStreamNode(vertexID).getMinResources();
+ for (StreamEdge chainable : chainedOutputs) {
+ minResources = minResources.merge(chainedMinResources.get(chainable.getTargetId()));
+ }
+ return minResources;
+ }
+
+ private ResourceSpec createChainedPreferredResources(Integer vertexID, List<StreamEdge> chainedOutputs) {
+ ResourceSpec preferredResources = streamGraph.getStreamNode(vertexID).getPreferredResources();
+ for (StreamEdge chainable : chainedOutputs) {
+ preferredResources = preferredResources.merge(chainedPreferredResources.get(chainable.getTargetId()));
+ }
+ return preferredResources;
+ }
+
private StreamConfig createJobVertex(
Integer streamNodeId,
Map<Integer, byte[]> hashes,
@@ -308,6 +332,8 @@ public class StreamingJobGraphGenerator {
legacyJobVertexIds);
}
+ jobVertex.setResources(chainedMinResources.get(streamNodeId), chainedPreferredResources.get(streamNodeId));
+
jobVertex.setInvokableClass(streamNode.getJobVertexClass());
int parallelism = streamNode.getParallelism();
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/StreamTransformation.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/StreamTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/StreamTransformation.java
index 24b5736..e86b3e8 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/StreamTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/StreamTransformation.java
@@ -132,13 +132,13 @@ public abstract class StreamTransformation<T> {
* The minimum resources for this stream transformation. It defines the lower limit for
* dynamic resources resize in future plan.
*/
- private ResourceSpec minResources = ResourceSpec.UNKNOWN;
+ private ResourceSpec minResources = ResourceSpec.DEFAULT;
/**
* The preferred resources for this stream transformation. It defines the upper limit for
* dynamic resource resize in future plan.
*/
- private ResourceSpec preferredResources = ResourceSpec.UNKNOWN;
+ private ResourceSpec preferredResources = ResourceSpec.DEFAULT;
/**
* User-specified ID for this transformation. This is used to assign the
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java
index b4d2421..5660655 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.streaming.api;
+import java.lang.reflect.Method;
import java.util.List;
import org.apache.flink.api.common.InvalidProgramException;
@@ -26,6 +27,7 @@ import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.Partitioner;
+import org.apache.flink.api.common.operators.ResourceSpec;
import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
@@ -516,11 +518,10 @@ public class DataStreamTest {
}
/**
- * Tests whether resource gets set.
+ * Tests whether resources get set.
*/
- /*
@Test
- public void testResource() {
+ public void testResources() throws Exception{
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
ResourceSpec minResource1 = new ResourceSpec(1.0, 100);
@@ -542,23 +543,35 @@ public class DataStreamTest {
ResourceSpec preferredResource6 = new ResourceSpec(2.0, 700);
ResourceSpec minResource7 = new ResourceSpec(1.0, 700);
- ResourceSpec maxResource7 = new ResourceSpec(2.0, 800);
+ ResourceSpec preferredResource7 = new ResourceSpec(2.0, 800);
+
+ Method opMethod = SingleOutputStreamOperator.class.getDeclaredMethod("setResources", ResourceSpec.class, ResourceSpec.class);
+ opMethod.setAccessible(true);
+
+ Method sinkMethod = DataStreamSink.class.getDeclaredMethod("setResources", ResourceSpec.class, ResourceSpec.class);
+ sinkMethod.setAccessible(true);
+
+ DataStream<Long> source1 = env.generateSequence(0, 0);
+ opMethod.invoke(source1, minResource1, preferredResource1);
- DataStream<Long> source1 = env.generateSequence(0, 0).setResource(minResource1, preferredResource1);
DataStream<Long> map1 = source1.map(new MapFunction<Long, Long>() {
@Override
public Long map(Long value) throws Exception {
return null;
}
- }).setResource(minResource2, preferredResource2);
+ });
+ opMethod.invoke(map1, minResource2, preferredResource2);
+
+ DataStream<Long> source2 = env.generateSequence(0, 0);
+ opMethod.invoke(source2, minResource3, preferredResource3);
- DataStream<Long> source2 = env.generateSequence(0, 0).setResource(minResource3, preferredResource3);
DataStream<Long> map2 = source2.map(new MapFunction<Long, Long>() {
@Override
public Long map(Long value) throws Exception {
return null;
}
- }).setResource(minResource4, preferredResource4);
+ });
+ opMethod.invoke(map2, minResource4, preferredResource4);
DataStream<Long> connected = map1.connect(map2)
.flatMap(new CoFlatMapFunction<Long, Long, Long>() {
@@ -568,7 +581,8 @@ public class DataStreamTest {
@Override
public void flatMap2(Long value, Collector<Long> out) throws Exception {
}
- }).setResource(minResource5, preferredResource5);
+ });
+ opMethod.invoke(connected, minResource5, preferredResource5);
DataStream<Long> windowed = connected
.windowAll(GlobalWindows.create())
@@ -580,31 +594,33 @@ public class DataStreamTest {
public Long fold(Long accumulator, Long value) throws Exception {
return null;
}
- }).setResource(minResource6, preferredResource6);
+ });
+ opMethod.invoke(windowed, minResource6, preferredResource6);
- DataStreamSink<Long> sink = windowed.print().setResource(minResource7, maxResource7);
+ DataStreamSink<Long> sink = windowed.print();
+ sinkMethod.invoke(sink, minResource7, preferredResource7);
- assertEquals(minResource1, env.getStreamGraph().getStreamNode(source1.getId()).getMinResource());
- assertEquals(preferredResource1, env.getStreamGraph().getStreamNode(source1.getId()).getPreferredResource());
+ assertEquals(minResource1, env.getStreamGraph().getStreamNode(source1.getId()).getMinResources());
+ assertEquals(preferredResource1, env.getStreamGraph().getStreamNode(source1.getId()).getPreferredResources());
- assertEquals(minResource2, env.getStreamGraph().getStreamNode(map1.getId()).getMinResource());
- assertEquals(preferredResource2, env.getStreamGraph().getStreamNode(map1.getId()).getPreferredResource());
+ assertEquals(minResource2, env.getStreamGraph().getStreamNode(map1.getId()).getMinResources());
+ assertEquals(preferredResource2, env.getStreamGraph().getStreamNode(map1.getId()).getPreferredResources());
- assertEquals(minResource3, env.getStreamGraph().getStreamNode(source2.getId()).getMinResource());
- assertEquals(preferredResource3, env.getStreamGraph().getStreamNode(source2.getId()).getPreferredResource());
+ assertEquals(minResource3, env.getStreamGraph().getStreamNode(source2.getId()).getMinResources());
+ assertEquals(preferredResource3, env.getStreamGraph().getStreamNode(source2.getId()).getPreferredResources());
- assertEquals(minResource4, env.getStreamGraph().getStreamNode(map2.getId()).getMinResource());
- assertEquals(preferredResource4, env.getStreamGraph().getStreamNode(map2.getId()).getPreferredResource());
+ assertEquals(minResource4, env.getStreamGraph().getStreamNode(map2.getId()).getMinResources());
+ assertEquals(preferredResource4, env.getStreamGraph().getStreamNode(map2.getId()).getPreferredResources());
- assertEquals(minResource5, env.getStreamGraph().getStreamNode(connected.getId()).getMinResource());
- assertEquals(preferredResource5, env.getStreamGraph().getStreamNode(connected.getId()).getPreferredResource());
+ assertEquals(minResource5, env.getStreamGraph().getStreamNode(connected.getId()).getMinResources());
+ assertEquals(preferredResource5, env.getStreamGraph().getStreamNode(connected.getId()).getPreferredResources());
- assertEquals(minResource6, env.getStreamGraph().getStreamNode(windowed.getId()).getMinResource());
- assertEquals(preferredResource6, env.getStreamGraph().getStreamNode(windowed.getId()).getPreferredResource());
+ assertEquals(minResource6, env.getStreamGraph().getStreamNode(windowed.getId()).getMinResources());
+ assertEquals(preferredResource6, env.getStreamGraph().getStreamNode(windowed.getId()).getPreferredResources());
- assertEquals(minResource7, env.getStreamGraph().getStreamNode(sink.getTransformation().getId()).getMinResource());
- assertEquals(maxResource7, env.getStreamGraph().getStreamNode(sink.getTransformation().getId()).getPreferredResource());
- }*/
+ assertEquals(minResource7, env.getStreamGraph().getStreamNode(sink.getTransformation().getId()).getMinResources());
+ assertEquals(preferredResource7, env.getStreamGraph().getStreamNode(sink.getTransformation().getId()).getPreferredResources());
+ }
@Test
public void testTypeInfo() {
http://git-wip-us.apache.org/repos/asf/flink/blob/980d072f/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
index 5f1973c..6d2fcaa 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
@@ -17,19 +17,29 @@
package org.apache.flink.streaming.api.graph;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.operators.ResourceSpec;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.tasks.JobSnapshottingSettings;
import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.datastream.IterativeStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
+import org.apache.flink.util.Collector;
import org.apache.flink.util.TestLogger;
import org.junit.Test;
+import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;
@@ -139,7 +149,7 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
StreamConfig sourceConfig = new StreamConfig(sourceVertex.getConfiguration());
StreamConfig mapConfig = new StreamConfig(mapPrintVertex.getConfiguration());
Map<Integer, StreamConfig> chainedConfigs = mapConfig.getTransitiveChainedTaskConfigs(getClass().getClassLoader());
- StreamConfig printConfig = chainedConfigs.get(3);
+ StreamConfig printConfig = chainedConfigs.values().iterator().next();
assertTrue(sourceConfig.isChainStart());
assertTrue(sourceConfig.isChainEnd());
@@ -150,4 +160,151 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
assertFalse(printConfig.isChainStart());
assertTrue(printConfig.isChainEnd());
}
+
+ /**
+ * Verifies that the resources are merged correctly for chained operators (covers source and sink cases)
+ * when generating job graph
+ */
+ @Test
+ public void testResourcesForChainedSourceSink() throws Exception {
+ ResourceSpec resource1 = new ResourceSpec(0.1, 100);
+ ResourceSpec resource2 = new ResourceSpec(0.2, 200);
+ ResourceSpec resource3 = new ResourceSpec(0.3, 300);
+ ResourceSpec resource4 = new ResourceSpec(0.4, 400);
+ ResourceSpec resource5 = new ResourceSpec(0.5, 500);
+
+ Method opMethod = SingleOutputStreamOperator.class.getDeclaredMethod("setResources", ResourceSpec.class);
+ opMethod.setAccessible(true);
+
+ Method sinkMethod = DataStreamSink.class.getDeclaredMethod("setResources", ResourceSpec.class);
+ sinkMethod.setAccessible(true);
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Tuple2<Integer, Integer>> source = env.addSource(new ParallelSourceFunction<Tuple2<Integer, Integer>>() {
+ @Override
+ public void run(SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception {
+ }
+
+ @Override
+ public void cancel() {
+ }
+ });
+ opMethod.invoke(source, resource1);
+
+ DataStream<Tuple2<Integer, Integer>> map = source.map(new MapFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {
+ @Override
+ public Tuple2<Integer, Integer> map(Tuple2<Integer, Integer> value) throws Exception {
+ return value;
+ }
+ });
+ opMethod.invoke(map, resource2);
+
+ // CHAIN(Source -> Map -> Filter)
+ DataStream<Tuple2<Integer, Integer>> filter = map.filter(new FilterFunction<Tuple2<Integer, Integer>>() {
+ @Override
+ public boolean filter(Tuple2<Integer, Integer> value) throws Exception {
+ return false;
+ }
+ });
+ opMethod.invoke(filter, resource3);
+
+ DataStream<Tuple2<Integer, Integer>> reduce = filter.keyBy(0).reduce(new ReduceFunction<Tuple2<Integer, Integer>>() {
+ @Override
+ public Tuple2<Integer, Integer> reduce(Tuple2<Integer, Integer> value1, Tuple2<Integer, Integer> value2) throws Exception {
+ return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
+ }
+ });
+ opMethod.invoke(reduce, resource4);
+
+ DataStreamSink<Tuple2<Integer, Integer>> sink = reduce.addSink(new SinkFunction<Tuple2<Integer, Integer>>() {
+ @Override
+ public void invoke(Tuple2<Integer, Integer> value) throws Exception {
+ }
+ });
+ sinkMethod.invoke(sink, resource5);
+
+ JobGraph jobGraph = new StreamingJobGraphGenerator(env.getStreamGraph()).createJobGraph();
+
+ JobVertex sourceMapFilterVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0);
+ JobVertex reduceSinkVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(1);
+
+ assertTrue(sourceMapFilterVertex.getMinResources().equals(resource1.merge(resource2).merge(resource3)));
+ assertTrue(reduceSinkVertex.getPreferredResources().equals(resource4.merge(resource5)));
+ }
+
+ /**
+ * Verifies that the resources are merged correctly for chained operators (covers middle chaining and iteration cases)
+ * when generating job graph
+ */
+ @Test
+ public void testResourcesForIteration() throws Exception {
+ ResourceSpec resource1 = new ResourceSpec(0.1, 100);
+ ResourceSpec resource2 = new ResourceSpec(0.2, 200);
+ ResourceSpec resource3 = new ResourceSpec(0.3, 300);
+ ResourceSpec resource4 = new ResourceSpec(0.4, 400);
+ ResourceSpec resource5 = new ResourceSpec(0.5, 500);
+
+ Method opMethod = SingleOutputStreamOperator.class.getDeclaredMethod("setResources", ResourceSpec.class);
+ opMethod.setAccessible(true);
+
+ Method sinkMethod = DataStreamSink.class.getDeclaredMethod("setResources", ResourceSpec.class);
+ sinkMethod.setAccessible(true);
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataStream<Integer> source = env.addSource(new ParallelSourceFunction<Integer>() {
+ @Override
+ public void run(SourceContext<Integer> ctx) throws Exception {
+ }
+
+ @Override
+ public void cancel() {
+ }
+ }).name("test_source");
+ opMethod.invoke(source, resource1);
+
+ IterativeStream<Integer> iteration = source.iterate(3000);
+ opMethod.invoke(iteration, resource2);
+
+ DataStream<Integer> flatMap = iteration.flatMap(new FlatMapFunction<Integer, Integer>() {
+ @Override
+ public void flatMap(Integer value, Collector<Integer> out) throws Exception {
+ out.collect(value);
+ }
+ }).name("test_flatMap");
+ opMethod.invoke(flatMap, resource3);
+
+ // CHAIN(flatMap -> Filter)
+ DataStream<Integer> increment = flatMap.filter(new FilterFunction<Integer>() {
+ @Override
+ public boolean filter(Integer value) throws Exception {
+ return false;
+ }
+ }).name("test_filter");
+ opMethod.invoke(increment, resource4);
+
+ DataStreamSink<Integer> sink = iteration.closeWith(increment).addSink(new SinkFunction<Integer>() {
+ @Override
+ public void invoke(Integer value) throws Exception {
+ }
+ }).disableChaining().name("test_sink");
+ sinkMethod.invoke(sink, resource5);
+
+ JobGraph jobGraph = new StreamingJobGraphGenerator(env.getStreamGraph()).createJobGraph();
+
+ for (JobVertex jobVertex : jobGraph.getVertices()) {
+ if (jobVertex.getName().contains("test_source")) {
+ assertTrue(jobVertex.getMinResources().equals(resource1));
+ } else if (jobVertex.getName().contains("Iteration_Source")) {
+ assertTrue(jobVertex.getPreferredResources().equals(resource2));
+ } else if (jobVertex.getName().contains("test_flatMap")) {
+ assertTrue(jobVertex.getMinResources().equals(resource3.merge(resource4)));
+ } else if (jobVertex.getName().contains("Iteration_Tail")) {
+ assertTrue(jobVertex.getPreferredResources().equals(ResourceSpec.DEFAULT));
+ } else if (jobVertex.getName().contains("test_sink")) {
+ assertTrue(jobVertex.getMinResources().equals(resource5));
+ }
+ }
+ }
}