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/09 18:57:06 UTC
[2/3] hive git commit: HIVE-13443 : LLAP: signing for the second
state of submit (the event) (Sergey Shelukhin, reviewed by Siddharth Seth)
http://git-wip-us.apache.org/repos/asf/hive/blob/b895d9d6/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorTestHelpers.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorTestHelpers.java b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorTestHelpers.java
index 1df5253..fe2ced5 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorTestHelpers.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TaskExecutorTestHelpers.java
@@ -39,6 +39,7 @@ import org.apache.tez.dag.records.TezTaskID;
import org.apache.tez.dag.records.TezVertexID;
import org.apache.tez.hadoop.shim.DefaultHadoopShim;
import org.apache.tez.runtime.api.impl.ExecutionContextImpl;
+import org.apache.tez.runtime.api.impl.TezEvent;
import org.apache.tez.runtime.task.EndReason;
import org.apache.tez.runtime.task.TaskRunner2Result;
import org.slf4j.Logger;
@@ -51,22 +52,21 @@ public class TaskExecutorTestHelpers {
public static MockRequest createMockRequest(int fragmentNum, int parallelism, long startTime,
boolean canFinish, long workTime) {
SubmitWorkRequestProto
- requestProto = createSubmitWorkRequestProto(fragmentNum, parallelism,
- startTime);
+ request = createSubmitWorkRequestProto(fragmentNum, parallelism, startTime);
+ return createMockRequest(canFinish, workTime, request);
+ }
+
+ private static MockRequest createMockRequest(boolean canFinish,
+ long workTime, SubmitWorkRequestProto request) {
QueryFragmentInfo queryFragmentInfo = createQueryFragmentInfo(
- requestProto.getWorkSpec().getVertex(), requestProto.getFragmentNumber());
- MockRequest mockRequest = new MockRequest(requestProto, queryFragmentInfo, canFinish, workTime);
- return mockRequest;
+ request.getWorkSpec().getVertex(), request.getFragmentNumber());
+ return new MockRequest(request, queryFragmentInfo, canFinish, workTime, null);
}
public static TaskExecutorService.TaskWrapper createTaskWrapper(
SubmitWorkRequestProto request, boolean canFinish, int workTime) {
- QueryFragmentInfo queryFragmentInfo = createQueryFragmentInfo(
- request.getWorkSpec().getVertex(), request.getFragmentNumber());
- MockRequest mockRequest = new MockRequest(request, queryFragmentInfo, canFinish, workTime);
- TaskExecutorService.TaskWrapper
- taskWrapper = new TaskExecutorService.TaskWrapper(mockRequest, null);
- return taskWrapper;
+ return new TaskExecutorService.TaskWrapper(
+ createMockRequest(canFinish, workTime, request), null);
}
public static QueryFragmentInfo createQueryFragmentInfo(
@@ -145,13 +145,13 @@ public class TaskExecutorTestHelpers {
private final Condition finishedCondition = lock.newCondition();
public MockRequest(SubmitWorkRequestProto requestProto, QueryFragmentInfo fragmentInfo,
- boolean canFinish, long workTime) {
+ boolean canFinish, long workTime, TezEvent initialEvent) {
super(requestProto, fragmentInfo, new Configuration(),
new ExecutionContextImpl("localhost"), null, new Credentials(), 0, mock(AMReporter.class), null, mock(
LlapDaemonExecutorMetrics.class),
mock(KilledTaskHandler.class), mock(
FragmentCompletionHandler.class), new DefaultHadoopShim(), null,
- requestProto.getWorkSpec().getVertex(), null);
+ requestProto.getWorkSpec().getVertex(), initialEvent, null);
this.workTime = workTime;
this.canFinish = canFinish;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/b895d9d6/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HiveSplitGenerator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HiveSplitGenerator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HiveSplitGenerator.java
index 54fc3af..5dd5e80 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HiveSplitGenerator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HiveSplitGenerator.java
@@ -33,9 +33,6 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hive.common.JavaUtils;
import org.apache.hadoop.hive.conf.HiveConf;
-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.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.plan.MapWork;
import org.apache.hadoop.hive.serde2.SerDeException;
@@ -47,7 +44,6 @@ import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.split.SplitLocationProvider;
import org.apache.hadoop.mapreduce.split.TezMapReduceSplitsGrouper;
import org.apache.hadoop.util.ReflectionUtils;
-import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.tez.common.TezUtils;
import org.apache.tez.dag.api.TaskLocationHint;
import org.apache.tez.dag.api.VertexLocationHint;
http://git-wip-us.apache.org/repos/asf/hive/blob/b895d9d6/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 7be6f7b..cff5ee1 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
@@ -43,6 +43,7 @@ import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.llap.LlapInputSplit;
import org.apache.hadoop.hive.llap.LlapOutputFormat;
+import org.apache.hadoop.hive.llap.NotTezEventHelper;
import org.apache.hadoop.hive.llap.SubmitWorkInfo;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.llap.Schema;
@@ -93,15 +94,17 @@ import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.InputDescriptor;
+import org.apache.tez.dag.api.InputInitializerDescriptor;
+import org.apache.tez.dag.api.RootInputLeafOutput;
import org.apache.tez.dag.api.TaskLocationHint;
import org.apache.tez.dag.api.TaskSpecBuilder;
import org.apache.tez.dag.api.Vertex;
import org.apache.tez.mapreduce.grouper.TezSplitGrouper;
import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.events.InputConfigureVertexTasksEvent;
-import org.apache.tez.runtime.api.impl.EventMetaData;
+import org.apache.tez.runtime.api.events.InputDataInformationEvent;
import org.apache.tez.runtime.api.impl.TaskSpec;
-import org.apache.tez.runtime.api.impl.TezEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -289,6 +292,7 @@ public class GenericUDTFGetSplits extends GenericUDTF {
// bunch of things get setup in the context based on conf but we need only the MR tmp directory
// for the following method.
JobConf wxConf = utils.initializeVertexConf(job, ctx, mapWork);
+ // TODO: should we also whitelist input formats here? from mapred.input.format.class
Path scratchDir = utils.createTezDir(ctx.getMRScratchDir(), job);
FileSystem fs = scratchDir.getFileSystem(job);
try {
@@ -307,7 +311,6 @@ public class GenericUDTFGetSplits extends GenericUDTF {
Preconditions.checkState(HiveConf.getBoolVar(wxConf,
HiveConf.ConfVars.LLAP_CLIENT_CONSISTENT_SPLITS));
HiveSplitGenerator splitGenerator = new HiveSplitGenerator(wxConf, mapWork);
- // TODO: these need to be signed, or at least the splits inside.
List<Event> eventList = splitGenerator.initialize();
InputSplit[] result = new InputSplit[eventList.size() - 1];
@@ -320,8 +323,7 @@ public class GenericUDTFGetSplits extends GenericUDTF {
Preconditions.checkState(hints.size() == eventList.size() - 1);
if (LOG.isDebugEnabled()) {
- LOG.debug("NumEvents=" + eventList.size());
- LOG.debug("NumSplits=" + result.length);
+ LOG.debug("NumEvents=" + eventList.size() + ", NumSplits=" + result.length);
}
LlapCoordinator coordinator = LlapCoordinator.getInstance();
@@ -345,6 +347,7 @@ public class GenericUDTFGetSplits extends GenericUDTF {
TaskSpec taskSpec = new TaskSpecBuilder().constructTaskSpec(dag, vertexName,
eventList.size() - 1, fakeApplicationId, i);
+ // 1. Generate the vertex/submit information.
if (i == 0) {
// Despite the differences in TaskSpec, the vertex spec should be the same.
signedSvs = createSignedVertexSpec(signer, taskSpec, fakeApplicationId);
@@ -353,41 +356,55 @@ public class GenericUDTFGetSplits extends GenericUDTF {
SubmitWorkInfo submitWorkInfo = new SubmitWorkInfo(fakeApplicationId,
System.currentTimeMillis(), taskSpec.getVertexParallelism(), signedSvs.message,
signedSvs.signature);
- EventMetaData sourceMetaData =
- new EventMetaData(EventMetaData.EventProducerConsumerType.INPUT, vertexName,
- "NULL_VERTEX", null);
- EventMetaData destinationMetaInfo = new TaskSpecBuilder().getDestingationMetaData(wx);
-
- // Creating the TezEvent here itself, since it's easy to serialize.
- Event event = eventList.get(i + 1);
- TaskLocationHint hint = hints.get(i);
- Set<String> hosts = hint.getHosts();
- if (hosts.size() != 1) {
- LOG.warn("Bad # of locations: " + hosts.size());
- }
- SplitLocationInfo[] locations = new SplitLocationInfo[hosts.size()];
+ byte[] submitWorkBytes = SubmitWorkInfo.toBytes(submitWorkInfo);
- int j = 0;
- for (String host : hosts) {
- locations[j++] = new SplitLocationInfo(host, false);
- }
- TezEvent tezEvent = new TezEvent(event, sourceMetaData, System.currentTimeMillis());
- tezEvent.setDestinationInfo(destinationMetaInfo);
+ // 2. Generate input event.
+ SignedMessage eventBytes = makeEventBytes(
+ wx, vertexName, eventList.get(i + 1), dob, signer);
- bos.reset();
- dob.reset();
- tezEvent.write(dob);
+ // 3. Make location hints.
+ SplitLocationInfo[] locations = makeLocationHints(hints.get(i));
- byte[] submitWorkBytes = SubmitWorkInfo.toBytes(submitWorkInfo);
-
- result[i] = new LlapInputSplit(i, submitWorkBytes, dob.getData(), locations, schema, llapUser);
- }
+ result[i] = new LlapInputSplit(i, submitWorkBytes, eventBytes.message,
+ eventBytes.signature, locations, schema, llapUser);
+ }
return result;
} catch (Exception e) {
throw new IOException(e);
}
}
+ private SplitLocationInfo[] makeLocationHints(TaskLocationHint hint) {
+ Set<String> hosts = hint.getHosts();
+ if (hosts.size() != 1) {
+ LOG.warn("Bad # of locations: " + hosts.size());
+ }
+ SplitLocationInfo[] locations = new SplitLocationInfo[hosts.size()];
+ int j = 0;
+ for (String host : hosts) {
+ locations[j++] = new SplitLocationInfo(host, false);
+ }
+ return locations;
+ }
+
+ private SignedMessage makeEventBytes(Vertex wx, String vertexName,
+ Event event, DataOutputBuffer dob, LlapSigner signer) throws IOException {
+ assert event instanceof InputDataInformationEvent;
+ List<RootInputLeafOutput<InputDescriptor, InputInitializerDescriptor>> inputs =
+ TaskSpecBuilder.getVertexInputs(wx);
+ Preconditions.checkState(inputs.size() == 1);
+
+ Signable signableNte = NotTezEventHelper.createSignableNotTezEvent(
+ (InputDataInformationEvent)event, vertexName, inputs.get(0).getName());
+ if (signer != null) {
+ return signer.serializeAndSign(signableNte);
+ } else {
+ SignedMessage sm = new SignedMessage();
+ sm.message = signableNte.serialize();
+ return sm;
+ }
+ }
+
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?)
http://git-wip-us.apache.org/repos/asf/hive/blob/b895d9d6/ql/src/java/org/apache/tez/dag/api/TaskSpecBuilder.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/tez/dag/api/TaskSpecBuilder.java b/ql/src/java/org/apache/tez/dag/api/TaskSpecBuilder.java
index 6d31802..2c719ee 100644
--- a/ql/src/java/org/apache/tez/dag/api/TaskSpecBuilder.java
+++ b/ql/src/java/org/apache/tez/dag/api/TaskSpecBuilder.java
@@ -63,15 +63,8 @@ public class TaskSpecBuilder {
return new TaskSpec(taskAttemptId, dag.getName(), vertexName, numSplits, processorDescriptor, inputSpecs, outputSpecs, null);
}
- public EventMetaData getDestingationMetaData(Vertex vertex) {
- List<RootInputLeafOutput<InputDescriptor, InputInitializerDescriptor>> inputs =
- vertex.getInputs();
- Preconditions.checkState(inputs.size() == 1);
- String inputName = inputs.get(0).getName();
- EventMetaData destMeta =
- new EventMetaData(EventMetaData.EventProducerConsumerType.INPUT, vertex.getName(),
- inputName, null);
- return destMeta;
+ public static List<RootInputLeafOutput<
+ InputDescriptor, InputInitializerDescriptor>> getVertexInputs(Vertex wx) {
+ return wx.getInputs();
}
-
}