You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ss...@apache.org on 2015/04/24 02:28:00 UTC
hive git commit: HIVE-10475. LLAP: Minor fixes after tez api
enhancements for dag completion. (Siddharth Seth)
Repository: hive
Updated Branches:
refs/heads/llap e36f1fc4c -> d6b3a1cfd
HIVE-10475. LLAP: Minor fixes after tez api enhancements for dag completion. (Siddharth Seth)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/d6b3a1cf
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/d6b3a1cf
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/d6b3a1cf
Branch: refs/heads/llap
Commit: d6b3a1cfd9653bc0127af246376c0bd6bd335858
Parents: e36f1fc
Author: Siddharth Seth <ss...@apache.org>
Authored: Thu Apr 23 17:02:05 2015 -0700
Committer: Siddharth Seth <ss...@apache.org>
Committed: Thu Apr 23 17:02:05 2015 -0700
----------------------------------------------------------------------
.../hive/llap/daemon/impl/LlapTaskReporter.java | 24 +++++++++++---------
.../dag/app/rm/LlapTaskSchedulerService.java | 2 +-
.../hive/llap/daemon/MiniLlapCluster.java | 5 ++++
3 files changed, 19 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/d6b3a1cf/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapTaskReporter.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapTaskReporter.java b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapTaskReporter.java
index 1f25f76..ef596ac 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapTaskReporter.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapTaskReporter.java
@@ -231,7 +231,7 @@ public class LlapTaskReporter implements TaskReporterInterface {
eventsToSend.drainTo(events);
if (!task.isTaskDone() && !task.hadFatalError()) {
- TezCounters counters = null;
+ boolean sendCounters = false;
/**
* Increasing the heartbeat interval can delay the delivery of events. Sending just updated
* records would save CPU in DAG AM, but certain counters are updated very frequently. Until
@@ -239,11 +239,10 @@ public class LlapTaskReporter implements TaskReporterInterface {
*/
// Not completely accurate, since OOB heartbeats could go out.
if ((nonOobHeartbeatCounter.get() - prevCounterSendHeartbeatNum) * pollInterval >= sendCounterInterval) {
- counters = task.getCounters();
+ sendCounters = true;
prevCounterSendHeartbeatNum = nonOobHeartbeatCounter.get();
}
- updateEvent = new TezEvent(new TaskStatusUpdateEvent(counters, task.getProgress()),
- updateEventMetadata);
+ updateEvent = new TezEvent(getStatusUpdateEvent(sendCounters), updateEventMetadata);
events.add(updateEvent);
}
@@ -321,13 +320,17 @@ public class LlapTaskReporter implements TaskReporterInterface {
* indicates an exception somewhere in the AM.
*/
private boolean taskSucceeded(TezTaskAttemptID taskAttemptID) throws IOException, TezException {
- TezEvent statusUpdateEvent = new TezEvent(new TaskStatusUpdateEvent(task.getCounters(),
- task.getProgress()), updateEventMetadata);
+ TezEvent statusUpdateEvent = new TezEvent(getStatusUpdateEvent(true), updateEventMetadata);
TezEvent taskCompletedEvent = new TezEvent(new TaskAttemptCompletedEvent(),
updateEventMetadata);
return !heartbeat(Lists.newArrayList(statusUpdateEvent, taskCompletedEvent)).shouldDie;
}
+ private TaskStatusUpdateEvent getStatusUpdateEvent(boolean sendCounters) {
+ return new TaskStatusUpdateEvent((sendCounters ? task.getCounters() : null),
+ task.getProgress(), task.getTaskStatistics());
+ }
+
/**
* Sends out final events for task failure.
* @param taskAttemptID
@@ -342,8 +345,7 @@ public class LlapTaskReporter implements TaskReporterInterface {
*/
private boolean taskFailed(TezTaskAttemptID taskAttemptID, Throwable t, String diagnostics,
EventMetaData srcMeta) throws IOException, TezException {
- TezEvent statusUpdateEvent = new TezEvent(new TaskStatusUpdateEvent(task.getCounters(),
- task.getProgress()), updateEventMetadata);
+ TezEvent statusUpdateEvent = new TezEvent(getStatusUpdateEvent(true), updateEventMetadata);
if (diagnostics == null) {
diagnostics = ExceptionUtils.getStackTrace(t);
} else {
@@ -382,18 +384,18 @@ public class LlapTaskReporter implements TaskReporterInterface {
}
}
- public boolean taskSucceeded(TezTaskAttemptID taskAttemptID) throws IOException, TezException {
+ public synchronized boolean taskSucceeded(TezTaskAttemptID taskAttemptID) throws IOException, TezException {
return currentCallable.taskSucceeded(taskAttemptID);
}
@Override
- public boolean taskFailed(TezTaskAttemptID taskAttemptID, Throwable t, String diagnostics,
+ public synchronized boolean taskFailed(TezTaskAttemptID taskAttemptID, Throwable t, String diagnostics,
EventMetaData srcMeta) throws IOException, TezException {
return currentCallable.taskFailed(taskAttemptID, t, diagnostics, srcMeta);
}
@Override
- public void addEvents(TezTaskAttemptID taskAttemptID, Collection<TezEvent> events) {
+ public synchronized void addEvents(TezTaskAttemptID taskAttemptID, Collection<TezEvent> events) {
currentCallable.addEvents(taskAttemptID, events);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/d6b3a1cf/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java b/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java
index 3a827c3..8100ece 100644
--- a/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java
+++ b/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java
@@ -312,7 +312,7 @@ public class LlapTaskSchedulerService extends TaskSchedulerService {
}
@Override
- public void resetMatchLocalityForAllHeldContainers() {
+ public void dagComplete() {
// This is effectively DAG completed, and can be used to reset statistics being tracked.
LOG.info("DAG: " + dagCounter.get() + " completed. Scheduling stats: " + dagStats);
dagCounter.incrementAndGet();
http://git-wip-us.apache.org/repos/asf/hive/blob/d6b3a1cf/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/MiniLlapCluster.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/MiniLlapCluster.java b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/MiniLlapCluster.java
index 9fd4ba6..ac8dcba 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/MiniLlapCluster.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/MiniLlapCluster.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.service.Service;
import org.apache.hadoop.util.Shell;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
public class MiniLlapCluster extends AbstractService {
private static final Log LOG = LogFactory.getLog(MiniLlapCluster.class);
@@ -138,6 +139,10 @@ public class MiniLlapCluster extends AbstractService {
numExecutorsPerService);
clusterSpecificConfiguration.setLong(
LlapConfiguration.LLAP_DAEMON_MEMORY_PER_INSTANCE_MB, execBytesPerService);
+ // Optimize local fetch does not work with LLAP due to different local directories
+ // used by containers and LLAP
+ clusterSpecificConfiguration
+ .setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH, false);
}
@Override