You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2016/06/10 23:10:09 UTC
hive git commit: HIVE-13731 : LLAP: return LLAP token with the splits
(Sergey Shelukhin, reviewed by Jason Dere)
Repository: hive
Updated Branches:
refs/heads/master 88005d246 -> bde4f8dac
HIVE-13731 : LLAP: return LLAP token with the splits (Sergey Shelukhin, reviewed by Jason Dere)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/bde4f8da
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/bde4f8da
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/bde4f8da
Branch: refs/heads/master
Commit: bde4f8dac94df1812715c08c37e10706689d1f54
Parents: 88005d2
Author: Sergey Shelukhin <se...@apache.org>
Authored: Fri Jun 10 15:49:47 2016 -0700
Committer: Sergey Shelukhin <se...@apache.org>
Committed: Fri Jun 10 15:49:47 2016 -0700
----------------------------------------------------------------------
.../hive/llap/ext/TestLlapInputSplit.java | 17 ++----
.../apache/hadoop/hive/llap/LlapInputSplit.java | 21 +++++++-
.../ext/LlapTaskUmbilicalExternalClient.java | 8 +--
.../hadoop/hive/llap/LlapBaseInputFormat.java | 14 ++++-
.../ql/udf/generic/GenericUDTFGetSplits.java | 57 +++++++++++++-------
5 files changed, 79 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/bde4f8da/itests/hive-unit/src/test/java/org/apache/hadoop/hive/llap/ext/TestLlapInputSplit.java
----------------------------------------------------------------------
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/llap/ext/TestLlapInputSplit.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/llap/ext/TestLlapInputSplit.java
index 7491222..654e92b 100644
--- a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/llap/ext/TestLlapInputSplit.java
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/llap/ext/TestLlapInputSplit.java
@@ -21,12 +21,8 @@ package org.apache.hadoop.hive.llap.ext;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
-import java.io.DataOutput;
import java.io.DataOutputStream;
import java.util.ArrayList;
-import java.util.HashMap;
-
-import org.apache.hadoop.io.Text;
import org.apache.hadoop.hive.llap.LlapInputSplit;
import org.apache.hadoop.hive.llap.Schema;
@@ -34,12 +30,7 @@ import org.apache.hadoop.hive.llap.FieldDesc;
import org.apache.hadoop.hive.llap.TypeDesc;
import org.apache.hadoop.mapred.SplitLocationInfo;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.ExpectedException;
import static org.junit.Assert.*;
public class TestLlapInputSplit {
@@ -59,8 +50,9 @@ public class TestLlapInputSplit {
colDescs.add(new FieldDesc("col2", new TypeDesc(TypeDesc.Type.INT)));
Schema schema = new Schema(colDescs);
+ byte[] tokenBytes = new byte[] { 1 };
LlapInputSplit split1 = new LlapInputSplit(splitNum, planBytes, fragmentBytes, null,
- locations, schema, "hive");
+ locations, schema, "hive", tokenBytes);
ByteArrayOutputStream byteOutStream = new ByteArrayOutputStream();
DataOutputStream dataOut = new DataOutputStream(byteOutStream);
split1.write(dataOut);
@@ -75,13 +67,12 @@ public class TestLlapInputSplit {
checkLlapSplits(split1, split2);
}
- static void checkLlapSplits(
- LlapInputSplit split1,
- LlapInputSplit split2) throws Exception {
+ static void checkLlapSplits(LlapInputSplit split1, LlapInputSplit split2) throws Exception {
assertEquals(split1.getSplitNum(), split2.getSplitNum());
assertArrayEquals(split1.getPlanBytes(), split2.getPlanBytes());
assertArrayEquals(split1.getFragmentBytes(), split2.getFragmentBytes());
+ assertArrayEquals(split1.getTokenBytes(), split2.getTokenBytes());
SplitLocationInfo[] locationInfo1 = split1.getLocationInfo();
SplitLocationInfo[] locationInfo2 = split2.getLocationInfo();
for (int idx = 0; idx < locationInfo1.length; ++idx) {
http://git-wip-us.apache.org/repos/asf/hive/blob/bde4f8da/llap-client/src/java/org/apache/hadoop/hive/llap/LlapInputSplit.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/LlapInputSplit.java b/llap-client/src/java/org/apache/hadoop/hive/llap/LlapInputSplit.java
index ab11926..2896651 100644
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/LlapInputSplit.java
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/LlapInputSplit.java
@@ -21,8 +21,10 @@ import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.hive.llap.Schema;
+import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
import org.apache.hadoop.mapred.InputSplitWithLocationInfo;
import org.apache.hadoop.mapred.SplitLocationInfo;
+import org.apache.hadoop.security.token.Token;
public class LlapInputSplit implements InputSplitWithLocationInfo {
@@ -33,13 +35,14 @@ public class LlapInputSplit implements InputSplitWithLocationInfo {
private Schema schema;
private String llapUser;
private byte[] fragmentBytesSignature;
+ private byte[] tokenBytes;
public LlapInputSplit() {
}
public LlapInputSplit(int splitNum, byte[] planBytes, byte[] fragmentBytes,
byte[] fragmentBytesSignature, SplitLocationInfo[] locations, Schema schema,
- String llapUser) {
+ String llapUser, byte[] tokenBytes) {
this.planBytes = planBytes;
this.fragmentBytes = fragmentBytes;
this.fragmentBytesSignature = fragmentBytesSignature;
@@ -47,6 +50,7 @@ public class LlapInputSplit implements InputSplitWithLocationInfo {
this.schema = schema;
this.splitNum = splitNum;
this.llapUser = llapUser;
+ this.tokenBytes = tokenBytes;
}
public Schema getSchema() {
@@ -83,6 +87,10 @@ public class LlapInputSplit implements InputSplitWithLocationInfo {
return fragmentBytesSignature;
}
+ public byte[] getTokenBytes() {
+ return tokenBytes;
+ }
+
@Override
public void write(DataOutput out) throws IOException {
out.writeInt(splitNum);
@@ -105,6 +113,12 @@ public class LlapInputSplit implements InputSplitWithLocationInfo {
schema.write(out);
out.writeUTF(llapUser);
+ if (tokenBytes != null) {
+ out.writeInt(tokenBytes.length);
+ out.write(tokenBytes);
+ } else {
+ out.writeInt(0);
+ }
}
@Override
@@ -133,6 +147,11 @@ public class LlapInputSplit implements InputSplitWithLocationInfo {
schema = new Schema();
schema.readFields(in);
llapUser = in.readUTF();
+ length = in.readInt();
+ if (length > 0) {
+ tokenBytes = new byte[length];
+ in.readFully(tokenBytes);
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/bde4f8da/llap-client/src/java/org/apache/hadoop/hive/llap/ext/LlapTaskUmbilicalExternalClient.java
----------------------------------------------------------------------
diff --git a/llap-client/src/java/org/apache/hadoop/hive/llap/ext/LlapTaskUmbilicalExternalClient.java b/llap-client/src/java/org/apache/hadoop/hive/llap/ext/LlapTaskUmbilicalExternalClient.java
index 3ebae4a..85943d2 100644
--- a/llap-client/src/java/org/apache/hadoop/hive/llap/ext/LlapTaskUmbilicalExternalClient.java
+++ b/llap-client/src/java/org/apache/hadoop/hive/llap/ext/LlapTaskUmbilicalExternalClient.java
@@ -39,6 +39,7 @@ import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWor
import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.VertexIdentifier;
import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.VertexOrBinary;
import org.apache.hadoop.hive.llap.protocol.LlapTaskUmbilicalProtocol;
+import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
import org.apache.hadoop.hive.llap.tez.Converters;
import org.apache.hadoop.hive.llap.tez.LlapProtocolClientProxy;
import org.apache.hadoop.hive.llap.tezplugins.helpers.LlapTaskUmbilicalServer;
@@ -100,7 +101,8 @@ public class LlapTaskUmbilicalExternalClient extends AbstractService {
}
public LlapTaskUmbilicalExternalClient(Configuration conf, String tokenIdentifier,
- Token<JobTokenIdentifier> sessionToken, LlapTaskUmbilicalExternalResponder responder) {
+ Token<JobTokenIdentifier> sessionToken, LlapTaskUmbilicalExternalResponder responder,
+ Token<LlapTokenIdentifier> llapToken) {
super(LlapTaskUmbilicalExternalClient.class.getName());
this.conf = conf;
this.umbilical = new LlapTaskUmbilicalExternalImpl();
@@ -110,8 +112,8 @@ public class LlapTaskUmbilicalExternalClient extends AbstractService {
this.timer = new ScheduledThreadPoolExecutor(1);
this.connectionTimeout = 3 * HiveConf.getTimeVar(conf,
HiveConf.ConfVars.LLAP_DAEMON_AM_LIVENESS_CONNECTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
- // No support for the LLAP token yet. Add support for configurable threads, however 1 should always be enough.
- this.communicator = new LlapProtocolClientProxy(1, conf, null);
+ // Add support for configurable threads, however 1 should always be enough.
+ this.communicator = new LlapProtocolClientProxy(1, conf, llapToken);
this.communicator.init(conf);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/bde4f8da/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 d1748cb..03b2a54 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
@@ -16,6 +16,7 @@
*/
package org.apache.hadoop.hive.llap;
+import java.io.ByteArrayInputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.IOException;
@@ -44,8 +45,10 @@ import org.apache.hadoop.hive.llap.ext.LlapTaskUmbilicalExternalClient.LlapTaskU
import org.apache.hadoop.hive.llap.registry.ServiceInstance;
import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService;
+import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
import org.apache.hadoop.hive.llap.tez.Converters;
import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.DataInputByteBuffer;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
@@ -127,12 +130,21 @@ public class LlapBaseInputFormat<V extends WritableComparable<?>>
LOG.info("Found service instance for host " + host + " with rpc port " + llapSubmitPort
+ " and outputformat port " + serviceInstance.getOutputFormatPort());
+ byte[] llapTokenBytes = llapSplit.getTokenBytes();
+ Token<LlapTokenIdentifier> llapToken = null;
+ if (llapTokenBytes != null) {
+ DataInputBuffer in = new DataInputBuffer();
+ in.reset(llapTokenBytes, 0, llapTokenBytes.length);
+ llapToken = new Token<LlapTokenIdentifier>();
+ llapToken.readFields(in);
+ }
+
LlapRecordReaderTaskUmbilicalExternalResponder umbilicalResponder =
new LlapRecordReaderTaskUmbilicalExternalResponder();
// TODO: close this
LlapTaskUmbilicalExternalClient llapClient =
new LlapTaskUmbilicalExternalClient(job, submitWorkInfo.getTokenIdentifier(),
- submitWorkInfo.getToken(), umbilicalResponder);
+ submitWorkInfo.getToken(), umbilicalResponder, llapToken);
llapClient.init(job);
llapClient.start();
http://git-wip-us.apache.org/repos/asf/hive/blob/bde4f8da/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java b/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java
index cff5ee1..e2f0e84 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDTFGetSplits.java
@@ -54,6 +54,8 @@ import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SignableV
import org.apache.hadoop.hive.llap.security.LlapSigner;
import org.apache.hadoop.hive.llap.security.LlapSigner.Signable;
import org.apache.hadoop.hive.llap.security.LlapSigner.SignedMessage;
+import org.apache.hadoop.hive.llap.security.LlapTokenIdentifier;
+import org.apache.hadoop.hive.llap.security.LlapTokenLocalClient;
import org.apache.hadoop.hive.llap.tez.Converters;
import org.apache.hadoop.hive.ql.CommandNeedRetryException;
import org.apache.hadoop.hive.ql.Context;
@@ -90,6 +92,7 @@ import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.SplitLocationInfo;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
@@ -251,7 +254,7 @@ public class GenericUDTFGetSplits extends GenericUDTF {
try {
HiveConf.setVar(conf, HiveConf.ConfVars.HIVE_EXECUTION_MODE, originalMode);
cpr = driver.run(ctas, false);
- } catch(CommandNeedRetryException e) {
+ } catch (CommandNeedRetryException e) {
throw new HiveException(e);
}
@@ -337,20 +340,40 @@ public class GenericUDTFGetSplits extends GenericUDTF {
// This assumes LLAP cluster owner is always the HS2 user.
String llapUser = UserGroupInformation.getLoginUser().getShortUserName();
- LlapSigner signer = UserGroupInformation.isSecurityEnabled()
- ? coordinator.getLlapSigner(job) : null;
+
+ String queryUser = null;
+ byte[] tokenBytes = null;
+ LlapSigner signer = null;
+ if (UserGroupInformation.isSecurityEnabled()) {
+ signer = coordinator.getLlapSigner(job);
+
+ // 1. Generate the token for query user (applies to all splits).
+ queryUser = SessionState.getUserFromAuthenticator();
+ if (queryUser == null) {
+ queryUser = UserGroupInformation.getCurrentUser().getUserName();
+ LOG.warn("Cannot determine the session user; using " + queryUser + " instead");
+ }
+ LlapTokenLocalClient tokenClient = coordinator.getLocalTokenClient(job, llapUser);
+ // We put the query user, not LLAP user, into the message and token.
+ Token<LlapTokenIdentifier> token = tokenClient.createToken(
+ fakeApplicationId.toString(), queryUser, true);
+ bos.reset();
+ token.write(dos);
+ tokenBytes = bos.toByteArray();
+ } else {
+ queryUser = UserGroupInformation.getCurrentUser().getUserName();
+ }
LOG.info("Number of splits: " + (eventList.size() - 1));
SignedMessage signedSvs = null;
- DataOutputBuffer dob = new DataOutputBuffer();
for (int i = 0; i < eventList.size() - 1; i++) {
TaskSpec taskSpec = new TaskSpecBuilder().constructTaskSpec(dag, vertexName,
eventList.size() - 1, fakeApplicationId, i);
- // 1. Generate the vertex/submit information.
+ // 2. Generate the vertex/submit information for all events.
if (i == 0) {
// Despite the differences in TaskSpec, the vertex spec should be the same.
- signedSvs = createSignedVertexSpec(signer, taskSpec, fakeApplicationId);
+ signedSvs = createSignedVertexSpec(signer, taskSpec, fakeApplicationId, queryUser);
}
SubmitWorkInfo submitWorkInfo = new SubmitWorkInfo(fakeApplicationId,
@@ -358,15 +381,14 @@ public class GenericUDTFGetSplits extends GenericUDTF {
signedSvs.signature);
byte[] submitWorkBytes = SubmitWorkInfo.toBytes(submitWorkInfo);
- // 2. Generate input event.
- SignedMessage eventBytes = makeEventBytes(
- wx, vertexName, eventList.get(i + 1), dob, signer);
+ // 3. Generate input event.
+ SignedMessage eventBytes = makeEventBytes(wx, vertexName, eventList.get(i + 1), signer);
- // 3. Make location hints.
+ // 4. Make location hints.
SplitLocationInfo[] locations = makeLocationHints(hints.get(i));
result[i] = new LlapInputSplit(i, submitWorkBytes, eventBytes.message,
- eventBytes.signature, locations, schema, llapUser);
+ eventBytes.signature, locations, schema, llapUser, tokenBytes);
}
return result;
} catch (Exception e) {
@@ -388,7 +410,7 @@ public class GenericUDTFGetSplits extends GenericUDTF {
}
private SignedMessage makeEventBytes(Vertex wx, String vertexName,
- Event event, DataOutputBuffer dob, LlapSigner signer) throws IOException {
+ Event event, LlapSigner signer) throws IOException {
assert event instanceof InputDataInformationEvent;
List<RootInputLeafOutput<InputDescriptor, InputInitializerDescriptor>> inputs =
TaskSpecBuilder.getVertexInputs(wx);
@@ -406,15 +428,10 @@ public class GenericUDTFGetSplits extends GenericUDTF {
}
private SignedMessage createSignedVertexSpec(LlapSigner signer, TaskSpec taskSpec,
- ApplicationId fakeApplicationId) throws IOException {
- // We put the query user, not LLAP user, in the message (and the token later?)
- String user = SessionState.getUserFromAuthenticator();
- if (user == null) {
- user = UserGroupInformation.getCurrentUser().getUserName();
- LOG.warn("Cannot determine the session user; using " + user + " instead");
- }
+ ApplicationId fakeApplicationId, String queryUser) throws IOException {
+
final SignableVertexSpec.Builder svsb = Converters.convertTaskSpecToProto(
- taskSpec, 0, fakeApplicationId.toString(), user);
+ taskSpec, 0, fakeApplicationId.toString(), queryUser);
if (signer == null) {
SignedMessage result = new SignedMessage();
result.message = serializeVertexSpec(svsb);