You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by ss...@apache.org on 2013/08/03 00:55:51 UTC
git commit: TEZ-335. Start tracking locality counters for tasks.
(sseth)
Updated Branches:
refs/heads/master bfab4ca67 -> 3b3630cde
TEZ-335. Start tracking locality counters for tasks. (sseth)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/3b3630cd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/3b3630cd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/3b3630cd
Branch: refs/heads/master
Commit: 3b3630cde6a84b0613c8eeef441691a67e759b40
Parents: bfab4ca
Author: Siddharth Seth <ss...@apache.org>
Authored: Fri Aug 2 15:55:24 2013 -0700
Committer: Siddharth Seth <ss...@apache.org>
Committed: Fri Aug 2 15:55:24 2013 -0700
----------------------------------------------------------------------
.../dag/event/TaskAttemptEventStatusUpdate.java | 19 ++++++
.../tez/dag/app/dag/impl/TaskAttemptImpl.java | 70 ++++++++++++++------
.../tez/mapreduce/hadoop/TestMRHelpers.java | 13 ++--
3 files changed, 76 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/3b3630cd/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/TaskAttemptEventStatusUpdate.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/TaskAttemptEventStatusUpdate.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/TaskAttemptEventStatusUpdate.java
index 2659bba..b968fc6 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/TaskAttemptEventStatusUpdate.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/TaskAttemptEventStatusUpdate.java
@@ -19,7 +19,9 @@
package org.apache.tez.dag.app.dag.event;
import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.tez.common.counters.DAGCounter;
import org.apache.tez.common.counters.TezCounters;
import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
import org.apache.tez.dag.records.TezTaskAttemptID;
@@ -43,6 +45,9 @@ public class TaskAttemptEventStatusUpdate extends TaskAttemptEvent {
*
*/
public static class TaskAttemptStatus {
+
+ private AtomicBoolean localitySet = new AtomicBoolean(false);
+
public TezTaskAttemptID id;
public float progress;
public TezCounters counters;
@@ -54,5 +59,19 @@ public class TaskAttemptEventStatusUpdate extends TaskAttemptEvent {
public long shuffleFinishTime;
public long sortFinishTime;
public TaskAttemptState taskState;
+
+ public void setLocalityCounter(DAGCounter localityCounter) {
+ if (!localitySet.get()) {
+ localitySet.set(true);
+ if (counters == null) {
+ counters = new TezCounters();
+ }
+ if (localityCounter != null) {
+ counters.findCounter(localityCounter).increment(1);
+ // TODO Maybe validate that the correct value is being set.
+ }
+ }
+ }
+
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/3b3630cd/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
index ce388db..6f897ba 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
@@ -20,6 +20,7 @@ package org.apache.tez.dag.app.dag.impl;
import java.net.InetSocketAddress;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
@@ -138,6 +139,11 @@ public class TaskAttemptImpl implements TaskAttempt,
private String nodeRackName;
private TaskAttemptStatus reportedStatus;
+ private DAGCounter localityCounter;
+
+ // Used to store locality information when
+ Set<String> taskHosts = new HashSet<String>();
+ Set<String> taskRacks = new HashSet<String>();
protected final TaskLocationHint locationHint;
protected final Resource taskResource;
@@ -365,6 +371,7 @@ public class TaskAttemptImpl implements TaskAttempt,
public TezCounters getCounters() {
readLock.lock();
try {
+ reportedStatus.setLocalityCounter(localityCounter);
TezCounters counters = reportedStatus.counters;
if (counters == null) {
counters = EMPTY_COUNTERS;
@@ -874,25 +881,32 @@ public class TaskAttemptImpl implements TaskAttempt,
TezTaskContext remoteTaskContext = ta.createRemoteTask();
// Create startTaskRequest
- String[] hostArray = new String[0];
- String[] rackArray = new String[0];
- if (!ta.isRescheduled) {
- // Ask for node / rack locality.
- Set<String> racks = new HashSet<String>();
- if (ta.locationHint != null) {
- if (ta.locationHint.getRacks() != null) {
- racks.addAll(ta.locationHint.getRacks());
- }
- if (ta.locationHint.getDataLocalHosts() != null) {
- for (String host : ta.locationHint.getDataLocalHosts()) {
- racks.add(RackResolver.resolve(host).getNetworkLocation());
- }
- hostArray = ta.resolveHosts(
- ta.locationHint.getDataLocalHosts().toArray(
- new String[ta.locationHint.getDataLocalHosts().size()]));
+ String[] requestHosts = new String[0];
+ String[] requestRacks = new String[0];
+
+ // Compute node/rack location request even if re-scheduled.
+ Set<String> racks = new HashSet<String>();
+ if (ta.locationHint != null) {
+ if (ta.locationHint.getRacks() != null) {
+ racks.addAll(ta.locationHint.getRacks());
+ }
+ if (ta.locationHint.getDataLocalHosts() != null) {
+ for (String host : ta.locationHint.getDataLocalHosts()) {
+ racks.add(RackResolver.resolve(host).getNetworkLocation());
}
+ requestHosts = ta.resolveHosts(ta.locationHint.getDataLocalHosts()
+ .toArray(new String[ta.locationHint.getDataLocalHosts().size()]));
}
- rackArray = racks.toArray(new String[racks.size()]);
+ }
+ requestRacks = racks.toArray(new String[racks.size()]);
+
+ ta.taskHosts.addAll(Arrays.asList(requestHosts));
+ ta.taskRacks = racks;
+
+ // Ask for hosts / racks only if not a re-scheduled task.
+ if (ta.isRescheduled) {
+ requestHosts = new String[0];
+ requestRacks = new String[0];
}
if (LOG.isDebugEnabled()) {
@@ -904,8 +918,8 @@ public class TaskAttemptImpl implements TaskAttempt,
new AMSchedulerEventTALaunchRequest(ta.attemptId,
ta.taskResource,
ta.localResources, remoteTaskContext, ta,
- ta.credentials, ta.jobToken, hostArray,
- rackArray,
+ ta.credentials, ta.jobToken, requestHosts,
+ requestRacks,
scheduleEvent.getPriority(), ta.environment, //ta.javaOpts,
ta.conf);
ta.sendEvent(launchRequestEvent);
@@ -993,6 +1007,19 @@ public class TaskAttemptImpl implements TaskAttempt,
// JobHistoryEvent
ta.logJobHistoryAttemptStarted();
+ // Compute LOLCAITY counter for this task.
+ if (ta.taskHosts.contains(ta.containerNodeId.getHost())) {
+ ta.localityCounter = DAGCounter.DATA_LOCAL_TASKS;
+ } else if (ta.taskRacks.contains(ta.nodeRackName)) {
+ ta.localityCounter = DAGCounter.RACK_LOCAL_TASKS;
+ } else {
+ // Not computing this if the task does not have locality information.
+ if (ta.locationHint != null) {
+ ta.localityCounter = DAGCounter.OTHER_LOCAL_TASKS;
+ }
+ }
+
+
// Inform the speculator about the container assignment.
//ta.maybeSendSpeculatorContainerNoLongerRequired();
// Inform speculator about startTime
@@ -1219,9 +1246,8 @@ public class TaskAttemptImpl implements TaskAttempt,
// result.phase = Phase.STARTING;
result.stateString = "NEW";
result.taskState = TaskAttemptState.NEW;
- TezCounters counters = EMPTY_COUNTERS;
- // counters.groups = new HashMap<String, CounterGroup>();
- result.counters = counters;
+ //TezCounters counters = EMPTY_COUNTERS;
+ //result.counters = counters;
}
private void addDiagnosticInfo(String diag) {
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/3b3630cd/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/hadoop/TestMRHelpers.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/hadoop/TestMRHelpers.java b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/hadoop/TestMRHelpers.java
index cd3ef88..91a8524 100644
--- a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/hadoop/TestMRHelpers.java
+++ b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/hadoop/TestMRHelpers.java
@@ -50,6 +50,7 @@ import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.VertexLocationHint.TaskLocationHint;
import org.apache.zookeeper.Environment.Entry;
import org.junit.Assert;
@@ -271,8 +272,10 @@ public class TestMRHelpers {
Assert.assertFalse(opts.contains(" fooReduceJavaOpts "));
Assert.assertTrue(opts.indexOf("fooMapAdminOpts")
< opts.indexOf("fooMapJavaOpts"));
- Assert.assertTrue(opts.contains(" -Dhadoop.root.logger=FATAL"));
- Assert.assertFalse(opts.contains(" -Dhadoop.root.logger=TRACE"));
+ Assert.assertTrue(opts.contains(" -D"
+ + TezConfiguration.TEZ_ROOT_LOGGER_NAME + "=FATAL"));
+ Assert.assertFalse(opts.contains(" -D"
+ + TezConfiguration.TEZ_ROOT_LOGGER_NAME + "=TRACE"));
}
@Test
@@ -286,8 +289,10 @@ public class TestMRHelpers {
Assert.assertTrue(opts.contains(" fooReduceJavaOpts "));
Assert.assertTrue(opts.indexOf("fooReduceAdminOpts")
< opts.indexOf("fooReduceJavaOpts"));
- Assert.assertFalse(opts.contains(" -Dhadoop.root.logger=FATAL"));
- Assert.assertTrue(opts.contains(" -Dhadoop.root.logger=TRACE"));
+ Assert.assertFalse(opts.contains(" -D"
+ + TezConfiguration.TEZ_ROOT_LOGGER_NAME + "=FATAL"));
+ Assert.assertTrue(opts.contains(" -D"
+ + TezConfiguration.TEZ_ROOT_LOGGER_NAME + "=TRACE"));
}
@Test