You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by uc...@apache.org on 2014/06/25 11:05:18 UTC
[2/3] git commit: LocalExecutor automatically picks up the maximum
DOP for plans as its number of slots.
LocalExecutor automatically picks up the maximum DOP for plans as its number of slots.
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/d55ba70e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/d55ba70e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/d55ba70e
Branch: refs/heads/master
Commit: d55ba70eb0d09bf4f62da390f19f1b10a0a4d94d
Parents: fe91d4b
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Jun 24 17:52:26 2014 +0200
Committer: uce <u....@fu-berlin.de>
Committed: Wed Jun 25 10:57:57 2014 +0200
----------------------------------------------------------------------
.../eu/stratosphere/client/LocalExecutor.java | 11 ++++-
.../java/eu/stratosphere/api/common/Plan.java | 49 +++++++++-----------
.../jobmanager/scheduler/DefaultScheduler.java | 2 +-
3 files changed, 33 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d55ba70e/stratosphere-clients/src/main/java/eu/stratosphere/client/LocalExecutor.java
----------------------------------------------------------------------
diff --git a/stratosphere-clients/src/main/java/eu/stratosphere/client/LocalExecutor.java b/stratosphere-clients/src/main/java/eu/stratosphere/client/LocalExecutor.java
index b017220..d73f893 100644
--- a/stratosphere-clients/src/main/java/eu/stratosphere/client/LocalExecutor.java
+++ b/stratosphere-clients/src/main/java/eu/stratosphere/client/LocalExecutor.java
@@ -42,7 +42,7 @@ public class LocalExecutor extends PlanExecutor {
private static boolean DEFAULT_OVERWRITE = false;
- private static final int DEFAULT_TASK_MANAGER_NUM_SLOTS = 1;
+ private static final int DEFAULT_TASK_MANAGER_NUM_SLOTS = -1;
private final Object lock = new Object(); // we lock to ensure singleton execution
@@ -214,6 +214,15 @@ public class LocalExecutor extends PlanExecutor {
if (this.nephele == null) {
// we start a session just for us now
shutDownAtEnd = true;
+
+ // configure the number of local slots equal to the parallelism of the local plan
+ if (this.taskManagerNumSlots == DEFAULT_TASK_MANAGER_NUM_SLOTS) {
+ int maxParallelism = plan.getMaximumParallelism();
+ if (maxParallelism > 0) {
+ this.taskManagerNumSlots = maxParallelism;
+ }
+ }
+
start();
} else {
// we use the existing session
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d55ba70e/stratosphere-core/src/main/java/eu/stratosphere/api/common/Plan.java
----------------------------------------------------------------------
diff --git a/stratosphere-core/src/main/java/eu/stratosphere/api/common/Plan.java b/stratosphere-core/src/main/java/eu/stratosphere/api/common/Plan.java
index b6f5385..8824107 100644
--- a/stratosphere-core/src/main/java/eu/stratosphere/api/common/Plan.java
+++ b/stratosphere-core/src/main/java/eu/stratosphere/api/common/Plan.java
@@ -62,11 +62,6 @@ public class Plan implements Visitable<Operator<?>> {
* The default parallelism to use for nodes that have no explicitly specified parallelism.
*/
protected int defaultParallelism = DEFAULT_PARALELLISM;
-
- /**
- * The maximal number of machines to use in the job.
- */
- protected int maxNumberMachines;
/**
* Hash map for files in the distributed cache: registered name to cache entry.
@@ -234,28 +229,6 @@ public class Plan implements Visitable<Operator<?>> {
checkNotNull(jobName, "The job name must not be null.");
this.jobName = jobName;
}
-
- /**
- * Gets the maximum number of machines to be used for this job.
- *
- * @return The maximum number of machines to be used for this job.
- */
- public int getMaxNumberMachines() {
- return this.maxNumberMachines;
- }
-
- /**
- * Sets the maximum number of machines to be used for this job.
- *
- * @param maxNumberMachines The the maximum number to set.
- */
- public void setMaxNumberMachines(int maxNumberMachines) {
- if (maxNumberMachines == 0 || maxNumberMachines < -1) {
- throw new IllegalArgumentException("The maximum number of machines must be positive, or -1 if no limit is imposed.");
- }
-
- this.maxNumberMachines = maxNumberMachines;
- }
/**
* Gets the default degree of parallelism for this job. That degree is always used when an operator
@@ -338,4 +311,26 @@ public class Plan implements Visitable<Operator<?>> {
public Set<Entry<String,DistributedCacheEntry>> getCachedFiles() {
return this.cacheFile.entrySet();
}
+
+ public int getMaximumParallelism() {
+ MaxDopVisitor visitor = new MaxDopVisitor();
+ accept(visitor);
+ return Math.max(visitor.maxDop, this.defaultParallelism);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private static final class MaxDopVisitor implements Visitor<Operator<?>> {
+
+ private int maxDop = -1;
+
+ @Override
+ public boolean preVisit(Operator<?> visitable) {
+ this.maxDop = Math.max(this.maxDop, visitable.getDegreeOfParallelism());
+ return true;
+ }
+
+ @Override
+ public void postVisit(Operator<?> visitable) {}
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/d55ba70e/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/scheduler/DefaultScheduler.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/scheduler/DefaultScheduler.java b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/scheduler/DefaultScheduler.java
index 745b199..48c9143 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/scheduler/DefaultScheduler.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/jobmanager/scheduler/DefaultScheduler.java
@@ -269,7 +269,7 @@ public class DefaultScheduler implements InstanceListener, JobStatusListener, Ex
final int requiredSlots = executionStage.getRequiredSlots();
- LOG.info("Requesting " + requiredSlots + " for job " + executionGraph.getJobID());
+ LOG.info("Requesting " + requiredSlots + " slots for job " + executionGraph.getJobID());
this.instanceManager.requestInstance(executionGraph.getJobID(), executionGraph.getJobConfiguration(),
requiredSlots);