hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jd...@apache.org
Subject hive git commit: HIVE-14078: LLAP input split should get task attempt number from conf if available (Jason Dere, reviewed by Prasanth Jayachandran/Siddharth Seth)
Date Thu, 30 Jun 2016 20:14:33 GMT
Repository: hive
Updated Branches:
  refs/heads/master df51738ad -> 2de64b0b0


HIVE-14078: LLAP input split should get task attempt number from conf if available (Jason
Dere, reviewed by Prasanth Jayachandran/Siddharth Seth)


Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/2de64b0b
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/2de64b0b
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/2de64b0b

Branch: refs/heads/master
Commit: 2de64b0b0575264aa9716bc6fa824cd076884257
Parents: df51738
Author: Jason Dere <jdere@hortonworks.com>
Authored: Thu Jun 30 13:14:03 2016 -0700
Committer: Jason Dere <jdere@hortonworks.com>
Committed: Thu Jun 30 13:14:03 2016 -0700

----------------------------------------------------------------------
 .../hadoop/hive/llap/LlapBaseInputFormat.java   | 22 +++++++++++++++-----
 1 file changed, 17 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/2de64b0b/llap-ext-client/src/java/org/apache/hadoop/hive/llap/LlapBaseInputFormat.java
----------------------------------------------------------------------
diff --git a/llap-ext-client/src/java/org/apache/hadoop/hive/llap/LlapBaseInputFormat.java
b/llap-ext-client/src/java/org/apache/hadoop/hive/llap/LlapBaseInputFormat.java
index aef5762..c2fca54 100644
--- a/llap-ext-client/src/java/org/apache/hadoop/hive/llap/LlapBaseInputFormat.java
+++ b/llap-ext-client/src/java/org/apache/hadoop/hive/llap/LlapBaseInputFormat.java
@@ -59,6 +59,8 @@ import org.apache.hadoop.mapred.InputSplitWithLocationInfo;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.RecordReader;
 import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapreduce.MRJobConfig;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.yarn.api.ApplicationConstants;
@@ -137,15 +139,25 @@ public class LlapBaseInputFormat<V extends WritableComparable<?>>
 
     LlapRecordReaderTaskUmbilicalExternalResponder umbilicalResponder =
         new LlapRecordReaderTaskUmbilicalExternalResponder();
-    // TODO: close this
     LlapTaskUmbilicalExternalClient llapClient =
       new LlapTaskUmbilicalExternalClient(job, submitWorkInfo.getTokenIdentifier(),
           submitWorkInfo.getToken(), umbilicalResponder, llapToken);
     llapClient.init(job);
     llapClient.start();
 
+    int attemptNum = 0;
+    // Use task attempt number from conf if provided
+    TaskAttemptID taskAttemptId = TaskAttemptID.forName(job.get(MRJobConfig.TASK_ATTEMPT_ID));
+    if (taskAttemptId != null) {
+      attemptNum = taskAttemptId.getId();
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Setting attempt number to " + attemptNum + " from task attempt ID in conf:
" +
+            job.get(MRJobConfig.TASK_ATTEMPT_ID));
+      }
+    }
+
     SubmitWorkRequestProto request = constructSubmitWorkRequestProto(
-        submitWorkInfo, llapSplit.getSplitNum(), llapClient.getAddress(),
+        submitWorkInfo, llapSplit.getSplitNum(), attemptNum, llapClient.getAddress(),
         submitWorkInfo.getToken(), llapSplit.getFragmentBytes(),
         llapSplit.getFragmentBytesSignature());
     llapClient.submitWork(request, host, llapSubmitPort);
@@ -275,7 +287,7 @@ public class LlapBaseInputFormat<V extends WritableComparable<?>>
   }
 
   private SubmitWorkRequestProto constructSubmitWorkRequestProto(SubmitWorkInfo submitWorkInfo,
-      int taskNum, InetSocketAddress address, Token<JobTokenIdentifier> token,
+      int taskNum, int attemptNum, InetSocketAddress address, Token<JobTokenIdentifier>
token,
       byte[] fragmentBytes, byte[] fragmentBytesSignature) throws IOException {
     ApplicationId appId = submitWorkInfo.getFakeAppId();
 
@@ -284,7 +296,7 @@ public class LlapBaseInputFormat<V extends WritableComparable<?>>
     LOG.info("Setting user in submitWorkRequest to: " + user);
 
     ContainerId containerId =
-        ContainerId.newInstance(ApplicationAttemptId.newInstance(appId, 0), taskNum);
+        ContainerId.newInstance(ApplicationAttemptId.newInstance(appId, attemptNum), taskNum);
 
     // Credentials can change across DAGs. Ideally construct only once per DAG.
     Credentials credentials = new Credentials();
@@ -309,7 +321,7 @@ public class LlapBaseInputFormat<V extends WritableComparable<?>>
     }
     builder.setWorkSpec(vertexBuilder.build());
     builder.setFragmentNumber(taskNum);
-    builder.setAttemptNumber(0); // TODO: hmm
+    builder.setAttemptNumber(attemptNum);
     builder.setContainerIdString(containerId.toString());
     builder.setAmHost(address.getHostName());
     builder.setAmPort(address.getPort());


Mime
View raw message