You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by bi...@apache.org on 2014/04/15 04:31:27 UTC
git commit: TEZ-700. Helper API's to monitor a DAG to completion
(Mohammad Kamrul Islam via bikas)
Repository: incubator-tez
Updated Branches:
refs/heads/master dbbb7a217 -> de87def6b
TEZ-700. Helper API's to monitor a DAG to completion (Mohammad Kamrul Islam via bikas)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/de87def6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/de87def6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/de87def6
Branch: refs/heads/master
Commit: de87def6b5f577b49c3fa2e1aaddeb29dece777c
Parents: dbbb7a2
Author: Bikas Saha <bi...@apache.org>
Authored: Mon Apr 14 19:31:21 2014 -0700
Committer: Bikas Saha <bi...@apache.org>
Committed: Mon Apr 14 19:31:21 2014 -0700
----------------------------------------------------------------------
.../java/org/apache/tez/client/TezSession.java | 26 ++++
.../apache/tez/dag/api/client/DAGClient.java | 40 ++++++
.../dag/api/client/rpc/DAGClientRPCImpl.java | 134 +++++++++++++++++++
.../tez/mapreduce/examples/UnionExample.java | 58 +-------
.../tez/mapreduce/examples/WordCount.java | 58 +-------
5 files changed, 204 insertions(+), 112 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/de87def6/tez-api/src/main/java/org/apache/tez/client/TezSession.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/client/TezSession.java b/tez-api/src/main/java/org/apache/tez/client/TezSession.java
index f131005..222e77c 100644
--- a/tez-api/src/main/java/org/apache/tez/client/TezSession.java
+++ b/tez-api/src/main/java/org/apache/tez/client/TezSession.java
@@ -71,6 +71,7 @@ public class TezSession {
/** Tokens which will be required for all DAGs submitted to this session. */
private Credentials sessionCredentials = new Credentials();
private long clientTimeout;
+ private static final long SLEEP_FOR_READY = 500;
private JobTokenSecretManager jobTokenSecretManager =
new JobTokenSecretManager();
@@ -382,6 +383,31 @@ public class TezSession {
throw new TezException(e);
}
}
+ /**
+ * Wait until TEZ session is ready.
+ *
+ * @throws IOException
+ * @throws TezException
+ */
+ @InterfaceStability.Evolving
+ public void waitTillReady() throws IOException, TezException {
+ verifySessionStateForSubmission();
+ while (true) {
+ TezSessionStatus status = getSessionStatus();
+ if (status.equals(TezSessionStatus.SHUTDOWN)) {
+ throw new SessionNotRunning("TezSession has already shutdown");
+ }
+ if (status.equals(TezSessionStatus.READY)) {
+ return;
+ }
+ try {
+ Thread.sleep(SLEEP_FOR_READY);
+ } catch (InterruptedException e) {
+ LOG.info("Sleep interrupted", e);
+ continue;
+ }
+ }
+ }
private DAGClientAMProtocolBlockingPB waitForProxy()
throws IOException, TezException, InterruptedException {
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/de87def6/tez-api/src/main/java/org/apache/tez/dag/api/client/DAGClient.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/client/DAGClient.java b/tez-api/src/main/java/org/apache/tez/dag/api/client/DAGClient.java
index bbb225c..32f2712 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/client/DAGClient.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/client/DAGClient.java
@@ -20,12 +20,15 @@ package org.apache.tez.dag.api.client;
import java.io.Closeable;
import java.io.IOException;
+import java.util.EnumSet;
import java.util.Set;
+import javax.annotation.Nullable;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.tez.dag.api.TezException;
+import org.apache.tez.dag.api.Vertex;
/*
* Interface class for monitoring the <code>DAG</code> running in a Tez DAG
@@ -71,4 +74,41 @@ public interface DAGClient extends Closeable {
*/
public void tryKillDAG() throws IOException, TezException;
+ /**
+ * Wait for DAG to complete without printing any vertex statuses
+ *
+ * @return Final DAG Status
+ * @throws IOException
+ * @throws TezException
+ */
+ public DAGStatus waitForCompletion() throws IOException, TezException;
+
+ /**
+ * Wait for DAG to complete and print the selected vertex status periodically.
+ *
+ * @param vertexNames
+ * which vertex details to print; null mean no vertex status and it
+ * is equivalent to call <code>waitForCompletion()</code>
+ * @param statusGetOpts
+ * : status get options. For example, to get counter pass
+ * <code>EnumSet.of(StatusGetOpts.GET_COUNTERS)</code>
+ * @return Final DAG Status
+ * @throws IOException
+ * @throws TezException
+ */
+ public DAGStatus waitForCompletionWithStatusUpdates(@Nullable Set<Vertex> vertices,
+ @Nullable Set<StatusGetOpts> statusGetOpts) throws IOException, TezException;
+
+ /**
+ * Wait for DAG to complete and periodically print *all* vertices' status.
+ *
+ * @param statusGetOpts
+ * : status get options. For example, to get counter pass
+ * <code>EnumSet.of(StatusGetOpts.GET_COUNTERS)</code>
+ * @return Final DAG Status
+ * @throws IOException
+ * @throws TezException
+ */
+ DAGStatus waitForCompletionWithAllStatusUpdates(@Nullable Set<StatusGetOpts> statusGetOpts)
+ throws IOException, TezException;
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/de87def6/tez-api/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientRPCImpl.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientRPCImpl.java b/tez-api/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientRPCImpl.java
index 8afdf05..711ffb6 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientRPCImpl.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientRPCImpl.java
@@ -19,9 +19,14 @@
package org.apache.tez.dag.api.client.rpc;
import java.io.IOException;
+import java.text.DecimalFormat;
import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashSet;
import java.util.Set;
+import javax.annotation.Nullable;
+import static com.google.common.base.Preconditions.checkNotNull;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.ipc.RPC;
@@ -33,12 +38,15 @@ import org.apache.hadoop.yarn.client.api.impl.YarnClientImpl;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.tez.client.TezClientUtils;
+import org.apache.tez.common.counters.TezCounters;
import org.apache.tez.dag.api.DagTypeConverters;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezException;
import org.apache.tez.dag.api.TezUncheckedException;
+import org.apache.tez.dag.api.Vertex;
import org.apache.tez.dag.api.client.DAGClient;
import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.api.client.Progress;
import org.apache.tez.dag.api.client.StatusGetOpts;
import org.apache.tez.dag.api.client.VertexStatus;
import org.apache.tez.dag.api.client.rpc.DAGClientAMProtocolRPC.GetDAGStatusRequestProto;
@@ -52,6 +60,8 @@ import com.google.protobuf.ServiceException;
public class DAGClientRPCImpl implements DAGClient {
private static final Log LOG = LogFactory.getLog(DAGClientRPCImpl.class);
+ private static final long SLEEP_FOR_COMPLETION = 500;
+ private final DecimalFormat formatter = new DecimalFormat("###.##%");
private final ApplicationId appId;
private final String dagId;
private final TezConfiguration conf;
@@ -305,4 +315,128 @@ public class DAGClientRPCImpl implements DAGClient {
return true;
}
+ @Override
+ public DAGStatus waitForCompletion() throws IOException, TezException {
+ return waitForCompletionWithStatusUpdates(null, EnumSet.noneOf(StatusGetOpts.class));
+ }
+
+ @Override
+ public DAGStatus waitForCompletionWithAllStatusUpdates(@Nullable Set<StatusGetOpts> statusGetOpts)
+ throws IOException, TezException {
+ Set<String> vertexSet = getDAGStatus(statusGetOpts).getVertexProgress().keySet();
+ return _waitForCompletionWithStatusUpdates(vertexSet, statusGetOpts);
+ }
+
+ @Override
+ public DAGStatus waitForCompletionWithStatusUpdates(@Nullable Set<Vertex> vertices,
+ @Nullable Set<StatusGetOpts> statusGetOpts) throws IOException, TezException {
+ Set<String> vertexNames = new HashSet<String>();
+ if (vertices != null) {
+ for (Vertex vertex : vertices) {
+ vertexNames.add(vertex.getVertexName());
+ }
+ }
+ return _waitForCompletionWithStatusUpdates(vertexNames, statusGetOpts);
+ }
+
+ private DAGStatus _waitForCompletionWithStatusUpdates(@Nullable Set<String> vertexNames,
+ @Nullable Set<StatusGetOpts> statusGetOpts) throws IOException, TezException {
+ DAGStatus dagStatus;
+ boolean initPrinted = false;
+ double dagProgress = -1.0; // Print the first one
+ // monitoring
+ while (true) {
+ dagStatus = getDAGStatus(statusGetOpts);
+ if (!initPrinted
+ && (dagStatus.getState() == DAGStatus.State.INITING || dagStatus.getState() == DAGStatus.State.SUBMITTED)) {
+ initPrinted = true; // Print once
+ log("Waiting for DAG to start running");
+ }
+ if (dagStatus.getState() == DAGStatus.State.RUNNING
+ || dagStatus.getState() == DAGStatus.State.SUCCEEDED
+ || dagStatus.getState() == DAGStatus.State.FAILED
+ || dagStatus.getState() == DAGStatus.State.KILLED
+ || dagStatus.getState() == DAGStatus.State.ERROR) {
+ break;
+ }
+ try {
+ Thread.sleep(SLEEP_FOR_COMPLETION);
+ } catch (InterruptedException e) {
+ // continue;
+ }
+ }// End of while(true)
+ while (dagStatus.getState() == DAGStatus.State.RUNNING) {
+ if (vertexNames != null) {
+ dagProgress = monitorProgress(vertexNames, dagProgress, null, dagStatus);
+ }
+ try {
+ Thread.sleep(SLEEP_FOR_COMPLETION);
+ } catch (InterruptedException e) {
+ }
+ dagStatus = getDAGStatus(statusGetOpts);
+ }// end of while
+ if (vertexNames != null) {
+ // Always print the last status irrespective of progress change
+ monitorProgress(vertexNames, -1.0, statusGetOpts, dagStatus);
+ }
+ log("DAG completed. " + "FinalState=" + dagStatus.getState());
+ return dagStatus;
+ }
+
+ private double monitorProgress(Set<String> vertexNamess, double prevDagProgress,
+ Set<StatusGetOpts> opts, DAGStatus dagStatus) throws IOException, TezException {
+ Progress progress = dagStatus.getDAGProgress();
+ double dagProgress = 0.0;
+ if (progress != null && (dagProgress = getProgress(progress)) > prevDagProgress) {
+ printDAGStatus(vertexNamess, opts, dagStatus, progress);
+ }
+ return dagProgress;
+ }
+
+ private void printDAGStatus(Set<String> vertexNamess, Set<StatusGetOpts> opts,
+ DAGStatus dagStatus, Progress dagProgress) throws IOException, TezException {
+ double vProgressFloat = 0.0f;
+ log("DAG: State: " + dagStatus.getState() + " Progress: "
+ + formatter.format(getProgress(dagProgress)) + " " + dagProgress);
+ boolean displayCounter = opts != null ? opts.contains(StatusGetOpts.GET_COUNTERS) : false;
+ if (displayCounter) {
+ TezCounters counters = dagStatus.getDAGCounters();
+ if (counters != null) {
+ log("DAG Counters:\n" + counters);
+ }
+ }
+ for (String vertex : vertexNamess) {
+ VertexStatus vStatus = getVertexStatus(vertex, opts);
+ if (vStatus == null) {
+ log("Could not retrieve status for vertex: " + vertex);
+ continue;
+ }
+ Progress vProgress = vStatus.getProgress();
+ if (vProgress != null) {
+ vProgressFloat = 0.0f;
+ if (vProgress.getTotalTaskCount() == 0) {
+ vProgressFloat = 1.0f;
+ } else if (vProgress.getTotalTaskCount() > 0) {
+ vProgressFloat = getProgress(vProgress);
+ }
+ log("VertexStatus:" + " VertexName: " + vertex + " Progress: "
+ + formatter.format(vProgressFloat) + " " + vProgress);
+ }
+ if (displayCounter) {
+ TezCounters counters = vStatus.getVertexCounters();
+ if (counters != null) {
+ log("Vertex Counters for " + vertex + ":\n" + counters);
+ }
+ }
+ } // end of for loop
+ }
+
+ private double getProgress(Progress progress) {
+ return (progress.getTotalTaskCount() == 0 ? 0.0 : (double) (progress.getSucceededTaskCount())
+ / progress.getTotalTaskCount());
+ }
+
+ private void log(String message) {
+ LOG.info(message);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/de87def6/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java
index 58443e8..14a9648 100644
--- a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java
+++ b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/UnionExample.java
@@ -365,24 +365,6 @@ public class UnionExample {
return dag;
}
- private static void waitForTezSessionReady(TezSession tezSession)
- throws IOException, TezException {
- while (true) {
- TezSessionStatus status = tezSession.getSessionStatus();
- if (status.equals(TezSessionStatus.SHUTDOWN)) {
- throw new RuntimeException("TezSession has already shutdown");
- }
- if (status.equals(TezSessionStatus.READY)) {
- return;
- }
- try {
- Thread.sleep(100);
- } catch (InterruptedException e) {
- return;
- }
- }
- }
-
private static void printUsage() {
System.err.println("Usage: " + " unionexample <in1> <out1>");
}
@@ -436,11 +418,8 @@ public class UnionExample {
sessionConfig);
tezSession.start();
- DAGStatus dagStatus = null;
DAGClient dagClient = null;
- String[] vNames = { "map1", "map2", "map3", "checker" };
- Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
try {
if (fs.exists(new Path(outputPath))) {
throw new FileAlreadyExistsException("Output directory "
@@ -453,44 +432,11 @@ public class UnionExample {
DAG dag = createDAG(fs, tezConf, localResources,
stagingDir, inputPath, outputPath);
- waitForTezSessionReady(tezSession);
+ tezSession.waitTillReady();
dagClient = tezSession.submitDAG(dag);
- //dagClient = tezClient.submitDAGApplication(dag, amConfig);
// monitoring
- while (true) {
- dagStatus = dagClient.getDAGStatus(statusGetOpts);
- if(dagStatus.getState() == DAGStatus.State.RUNNING ||
- dagStatus.getState() == DAGStatus.State.SUCCEEDED ||
- dagStatus.getState() == DAGStatus.State.FAILED ||
- dagStatus.getState() == DAGStatus.State.KILLED ||
- dagStatus.getState() == DAGStatus.State.ERROR) {
- break;
- }
- try {
- Thread.sleep(500);
- } catch (InterruptedException e) {
- // continue;
- }
- }
-
-
- while (dagStatus.getState() == DAGStatus.State.RUNNING) {
- try {
- ExampleDriver.printDAGStatus(dagClient, vNames);
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- // continue;
- }
- dagStatus = dagClient.getDAGStatus(statusGetOpts);
- } catch (TezException e) {
- System.exit(-1);
- }
- }
- ExampleDriver.printDAGStatus(dagClient, vNames,
- true, true);
- System.out.println("DAG completed. " + "FinalState=" + dagStatus.getState());
+ DAGStatus dagStatus = dagClient.waitForCompletionWithStatusUpdates(dag.getVertices(), EnumSet.of(StatusGetOpts.GET_COUNTERS));
if (dagStatus.getState() != DAGStatus.State.SUCCEEDED) {
System.out.println("DAG diagnostics: " + dagStatus.getDiagnostics());
return false;
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/de87def6/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/WordCount.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/WordCount.java b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/WordCount.java
index b1cede3..20d9229 100644
--- a/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/WordCount.java
+++ b/tez-mapreduce-examples/src/main/java/org/apache/tez/mapreduce/examples/WordCount.java
@@ -259,24 +259,6 @@ public class WordCount {
return dag;
}
- private static void waitForTezSessionReady(TezSession tezSession)
- throws IOException, TezException {
- while (true) {
- TezSessionStatus status = tezSession.getSessionStatus();
- if (status.equals(TezSessionStatus.SHUTDOWN)) {
- throw new RuntimeException("TezSession has already shutdown");
- }
- if (status.equals(TezSessionStatus.READY)) {
- return;
- }
- try {
- Thread.sleep(100);
- } catch (InterruptedException e) {
- return;
- }
- }
- }
-
private static void printUsage() {
System.err.println("Usage: " + " wordcount <in1> <out1>");
}
@@ -330,11 +312,8 @@ public class WordCount {
sessionConfig);
tezSession.start();
- DAGStatus dagStatus = null;
DAGClient dagClient = null;
- String[] vNames = { "tokenizer", "summer" };
- Set<StatusGetOpts> statusGetOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS);
try {
if (fs.exists(new Path(outputPath))) {
throw new FileAlreadyExistsException("Output directory "
@@ -347,44 +326,11 @@ public class WordCount {
DAG dag = createDAG(fs, tezConf, localResources,
stagingDir, inputPath, outputPath);
- waitForTezSessionReady(tezSession);
+ tezSession.waitTillReady();
dagClient = tezSession.submitDAG(dag);
- //dagClient = tezClient.submitDAGApplication(dag, amConfig);
// monitoring
- while (true) {
- dagStatus = dagClient.getDAGStatus(statusGetOpts);
- if(dagStatus.getState() == DAGStatus.State.RUNNING ||
- dagStatus.getState() == DAGStatus.State.SUCCEEDED ||
- dagStatus.getState() == DAGStatus.State.FAILED ||
- dagStatus.getState() == DAGStatus.State.KILLED ||
- dagStatus.getState() == DAGStatus.State.ERROR) {
- break;
- }
- try {
- Thread.sleep(500);
- } catch (InterruptedException e) {
- // continue;
- }
- }
-
-
- while (dagStatus.getState() == DAGStatus.State.RUNNING) {
- try {
- ExampleDriver.printDAGStatus(dagClient, vNames);
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- // continue;
- }
- dagStatus = dagClient.getDAGStatus(statusGetOpts);
- } catch (TezException e) {
- System.exit(-1);
- }
- }
- ExampleDriver.printDAGStatus(dagClient, vNames,
- true, true);
- System.out.println("DAG completed. " + "FinalState=" + dagStatus.getState());
+ DAGStatus dagStatus = dagClient.waitForCompletionWithAllStatusUpdates(null);
if (dagStatus.getState() != DAGStatus.State.SUCCEEDED) {
System.out.println("DAG diagnostics: " + dagStatus.getDiagnostics());
return false;