You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by ss...@apache.org on 2014/01/30 07:18:15 UTC
git commit: TEZ-646. Introduce a CompositeDataMovementEvent to avoid
multiple copies of the same payload in the AM. (sseth)
Updated Branches:
refs/heads/master 9db81d999 -> f55dbfb81
TEZ-646. Introduce a CompositeDataMovementEvent to avoid multiple copies
of the same payload in the AM. (sseth)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/f55dbfb8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/f55dbfb8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/f55dbfb8
Branch: refs/heads/master
Commit: f55dbfb81fff6b6612890d378824c2404b118272
Parents: 9db81d9
Author: Siddharth Seth <ss...@apache.org>
Authored: Wed Jan 29 22:16:45 2014 -0800
Committer: Siddharth Seth <ss...@apache.org>
Committed: Wed Jan 29 22:16:45 2014 -0800
----------------------------------------------------------------------
pom.xml | 5 +
.../org/apache/tez/client/TezClientUtils.java | 4 +-
.../api/events/CompositeDataMovementEvent.java | 98 +++++++++++
tez-api/src/main/proto/Events.proto | 6 +
tez-dag/pom.xml | 4 +
.../org/apache/tez/dag/app/dag/impl/Edge.java | 31 ++--
.../apache/tez/dag/app/dag/impl/VertexImpl.java | 9 +-
.../apache/tez/dag/app/dag/impl/TestEdge.java | 175 +++++++++++++++++++
tez-runtime-internals/pom.xml | 4 +
.../tez/runtime/api/impl/EventMetaData.java | 18 +-
.../apache/tez/runtime/api/impl/EventType.java | 3 +-
.../apache/tez/runtime/api/impl/TezEvent.java | 19 ++
.../library/output/OnFileSortedOutput.java | 11 +-
13 files changed, 347 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 458d721..ba5a3b9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -257,6 +257,11 @@
<artifactId>netty</artifactId>
<version>3.9.0.Final</version>
</dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>15.0</version>
+ </dependency>
</dependencies>
</dependencyManagement>
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java b/tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java
index bd97de1..26af839 100644
--- a/tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java
+++ b/tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java
@@ -34,8 +34,6 @@ import java.util.TreeMap;
import java.util.Vector;
import java.util.Map.Entry;
-import javax.annotation.Nullable;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
@@ -261,7 +259,7 @@ public class TezClientUtils {
if (uris != null && !uris.isEmpty()) {
Iterator<Path> pathIter = Iterators.transform(uris.iterator(), new Function<URI, Path>() {
@Override
- public Path apply(@Nullable URI input) {
+ public Path apply(URI input) {
return new Path(input);
}
});
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/tez-api/src/main/java/org/apache/tez/runtime/api/events/CompositeDataMovementEvent.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/runtime/api/events/CompositeDataMovementEvent.java b/tez-api/src/main/java/org/apache/tez/runtime/api/events/CompositeDataMovementEvent.java
new file mode 100644
index 0000000..f7d7878
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/runtime/api/events/CompositeDataMovementEvent.java
@@ -0,0 +1,98 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tez.runtime.api.events;
+
+import org.apache.tez.runtime.api.Event;
+
+import com.google.common.collect.ContiguousSet;
+import com.google.common.collect.DiscreteDomain;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Range;
+import com.google.common.base.Function;
+
+/**
+ * A convenience class to specify multiple DataMovementEvents which share the
+ * same payload. A contiguous range of srcIndices can be specified.
+ *
+ * This event will NOT be seen by downstream Inputs - instead they will see
+ * {@link DataMovementEvent}s which are generated based on the range specified
+ * in this event.
+ *
+ * This event should be used by an output which has the same payload for all of
+ * the Physical Outputs that it generates.
+ *
+ */
+public class CompositeDataMovementEvent extends Event {
+
+ protected final int sourceIndexStart;
+ protected final int sourceIndexEnd;
+ protected int version;
+
+ protected final byte[] userPayload;
+
+ /**
+ * @param srcIndexStart
+ * the startIndex of the physical source which generated the event
+ * (inclusive)
+ * @param srcIndexEnd
+ * the endIndex of the physical source which generated the event
+ * (non-inclusive)
+ * @param userPayload
+ * the common payload between all the events.
+ */
+ public CompositeDataMovementEvent(int srcIndexStart, int srcIndexEnd, byte[] userPayload) {
+ this.sourceIndexStart = srcIndexStart;
+ this.sourceIndexEnd = srcIndexEnd;
+ this.userPayload = userPayload;
+ }
+
+ public int getSourceIndexStart() {
+ return sourceIndexStart;
+ }
+
+ public int getSourceIndexEnd() {
+ return sourceIndexEnd;
+ }
+
+ public byte[] getUserPayload() {
+ return userPayload;
+ }
+
+ public void setVersion(int version) {
+ this.version = version;
+ }
+
+ public int getVersion() {
+ return this.version;
+ }
+
+ public Iterable<DataMovementEvent> getEvents() {
+ // Overkill. For now is enough to just run a loop over the ints.
+ Range<Integer> range = Range.closedOpen(sourceIndexStart, sourceIndexEnd);
+ ContiguousSet<Integer> intRange = ContiguousSet.create(range, DiscreteDomain.integers());
+ return Iterables.transform(intRange, new Function<Integer, DataMovementEvent>() {
+ public DataMovementEvent apply(Integer integer) {
+ DataMovementEvent dmEvent = new DataMovementEvent(integer, userPayload);
+ dmEvent.setVersion(version);
+ return dmEvent;
+ }
+ });
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/tez-api/src/main/proto/Events.proto
----------------------------------------------------------------------
diff --git a/tez-api/src/main/proto/Events.proto b/tez-api/src/main/proto/Events.proto
index a738972..c5cfabf 100644
--- a/tez-api/src/main/proto/Events.proto
+++ b/tez-api/src/main/proto/Events.proto
@@ -48,3 +48,9 @@ message RootInputDataInformationEventProto {
optional int32 index = 1;
optional bytes user_payload = 2;
}
+
+message CompositeEventProto {
+ optional int32 start_index = 1;
+ optional int32 end_index = 2;
+ optional bytes user_payload = 3;
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/tez-dag/pom.xml
----------------------------------------------------------------------
diff --git a/tez-dag/pom.xml b/tez-dag/pom.xml
index 67ae2e0..2342150 100644
--- a/tez-dag/pom.xml
+++ b/tez-dag/pom.xml
@@ -96,6 +96,10 @@
<groupId>com.google.inject</groupId>
<artifactId>guice</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
</dependencies>
<build>
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java
index e8ec81a..288a417 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java
@@ -36,6 +36,7 @@ import org.apache.tez.dag.records.TezTaskAttemptID;
import org.apache.tez.dag.records.TezTaskID;
import org.apache.tez.runtime.RuntimeUtils;
import org.apache.tez.runtime.api.Event;
+import org.apache.tez.runtime.api.events.CompositeDataMovementEvent;
import org.apache.tez.runtime.api.events.DataMovementEvent;
import org.apache.tez.runtime.api.events.InputFailedEvent;
import org.apache.tez.runtime.api.events.InputReadErrorEvent;
@@ -85,6 +86,7 @@ public class Edge {
private List<TezEvent> sourceEventBuffer = new ArrayList<TezEvent>();
private Vertex sourceVertex;
private Vertex destinationVertex; // this may end up being a list for shared edge
+ private EventMetaData destinationMetaInfo;
@SuppressWarnings("rawtypes")
public Edge(EdgeProperty edgeProperty, EventHandler eventHandler) {
@@ -119,6 +121,10 @@ public class Edge {
edgeManagerContext = new EdgeManagerContextImpl(sourceVertex.getName(),
destinationVertex.getName(), bb);
edgeManager.initialize(edgeManagerContext);
+ destinationMetaInfo = new EventMetaData(EventProducerConsumerType.INPUT,
+ destinationVertex.getName(),
+ sourceVertex.getName(),
+ null);
}
public EdgeProperty getEdgeProperty() {
@@ -222,11 +228,25 @@ public class Edge {
}
}
+
+ private void handleCompositeDataMovementEvent(TezEvent tezEvent) {
+ CompositeDataMovementEvent compEvent = (CompositeDataMovementEvent) tezEvent.getEvent();
+ EventMetaData srcInfo = tezEvent.getSourceInfo();
+
+ for (DataMovementEvent dmEvent : compEvent.getEvents()) {
+ TezEvent newEvent = new TezEvent(dmEvent, srcInfo);
+ sendTezEventToDestinationTasks(newEvent);
+ }
+ }
+
public void sendTezEventToDestinationTasks(TezEvent tezEvent) {
if (!bufferEvents.get()) {
List<Integer> destTaskIndices = new ArrayList<Integer>();
boolean isDataMovementEvent = true;
switch (tezEvent.getEventType()) {
+ case COMPOSITE_DATA_MOVEMENT_EVENT:
+ handleCompositeDataMovementEvent(tezEvent);
+ break;
case INPUT_FAILED_EVENT:
isDataMovementEvent = false;
case DATA_MOVEMENT_EVENT:
@@ -242,16 +262,7 @@ public class Edge {
sourceTaskIndex, destinationVertex.getTotalTasks(),
destTaskIndices);
}
- EventMetaData destMeta = new EventMetaData(EventProducerConsumerType.INPUT,
- destinationVertex.getName(),
- sourceVertex.getName(),
- null);
- if (isDataMovementEvent) {
- destMeta.setIndex(((DataMovementEvent)event).getTargetIndex());
- } else {
- destMeta.setIndex(((InputFailedEvent)event).getTargetIndex());
- }
- tezEvent.setDestinationInfo(destMeta);
+ tezEvent.setDestinationInfo(destinationMetaInfo);
for(Integer destTaskIndex : destTaskIndices) {
Task destTask = destinationVertex.getTask(destTaskIndex);
if (destTask == null) {
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
index 8d1826d..87c54bf 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
@@ -116,12 +116,14 @@ import org.apache.tez.dag.records.TezVertexID;
import org.apache.tez.runtime.RuntimeUtils;
import org.apache.tez.runtime.api.OutputCommitter;
import org.apache.tez.runtime.api.OutputCommitterContext;
+import org.apache.tez.runtime.api.events.CompositeDataMovementEvent;
import org.apache.tez.runtime.api.events.DataMovementEvent;
import org.apache.tez.runtime.api.events.InputFailedEvent;
import org.apache.tez.runtime.api.events.TaskAttemptFailedEvent;
import org.apache.tez.runtime.api.events.TaskStatusUpdateEvent;
import org.apache.tez.runtime.api.events.VertexManagerEvent;
import org.apache.tez.runtime.api.impl.EventMetaData;
+import org.apache.tez.runtime.api.impl.EventType;
import org.apache.tez.runtime.api.impl.InputSpec;
import org.apache.tez.runtime.api.impl.OutputSpec;
import org.apache.tez.runtime.api.impl.TezEvent;
@@ -1983,17 +1985,18 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
+ tezEvent.getEventType());
}
EventMetaData sourceMeta = tezEvent.getSourceInfo();
- boolean isDataMovementEvent = true;
switch(tezEvent.getEventType()) {
case INPUT_FAILED_EVENT:
- isDataMovementEvent = false;
case DATA_MOVEMENT_EVENT:
+ case COMPOSITE_DATA_MOVEMENT_EVENT:
{
if (isEventFromVertex(vertex, sourceMeta)) {
// event from this vertex. send to destination vertex
TezTaskAttemptID srcTaId = sourceMeta.getTaskAttemptID();
- if (isDataMovementEvent) {
+ if (tezEvent.getEventType() == EventType.DATA_MOVEMENT_EVENT) {
((DataMovementEvent) tezEvent.getEvent()).setVersion(srcTaId.getId());
+ } else if (tezEvent.getEventType() == EventType.COMPOSITE_DATA_MOVEMENT_EVENT) {
+ ((CompositeDataMovementEvent) tezEvent.getEvent()).setVersion(srcTaId.getId());
} else {
((InputFailedEvent) tezEvent.getEvent()).setVersion(srcTaId.getId());
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestEdge.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestEdge.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestEdge.java
new file mode 100644
index 0000000..36a5a04
--- /dev/null
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestEdge.java
@@ -0,0 +1,175 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tez.dag.app.dag.impl;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map.Entry;
+
+import org.apache.hadoop.yarn.event.Event;
+import org.apache.hadoop.yarn.event.EventHandler;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
+import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
+import org.apache.tez.dag.api.InputDescriptor;
+import org.apache.tez.dag.api.OutputDescriptor;
+import org.apache.tez.dag.app.dag.Task;
+import org.apache.tez.dag.app.dag.Vertex;
+import org.apache.tez.dag.app.dag.event.TaskEventAddTezEvent;
+import org.apache.tez.dag.records.TezDAGID;
+import org.apache.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.dag.records.TezTaskID;
+import org.apache.tez.dag.records.TezVertexID;
+import org.apache.tez.runtime.api.events.CompositeDataMovementEvent;
+import org.apache.tez.runtime.api.events.DataMovementEvent;
+import org.apache.tez.runtime.api.impl.EventMetaData;
+import org.apache.tez.runtime.api.impl.EventMetaData.EventProducerConsumerType;
+import org.apache.tez.runtime.api.impl.TezEvent;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+public class TestEdge {
+
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ @Test (timeout = 5000)
+ public void testCompositeEventHandling() {
+ EventHandler eventHandler = mock(EventHandler.class);
+ EdgeProperty edgeProp = new EdgeProperty(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL, mock(OutputDescriptor.class),
+ mock(InputDescriptor.class));
+ Edge edge = new Edge(edgeProp, eventHandler);
+
+ TezVertexID srcVertexID = createVertexID(1);
+ TezVertexID destVertexID = createVertexID(2);
+ LinkedHashMap<TezTaskID, Task> srcTasks = mockTasks(srcVertexID, 1);
+ LinkedHashMap<TezTaskID, Task> destTasks = mockTasks(destVertexID, 5);
+
+ TezTaskID srcTaskID = srcTasks.keySet().iterator().next();
+
+ Vertex srcVertex = mockVertex("src", srcVertexID, srcTasks);
+ Vertex destVertex = mockVertex("dest", destVertexID, destTasks);
+
+ edge.setSourceVertex(srcVertex);
+ edge.setDestinationVertex(destVertex);
+ edge.initialize();
+
+ TezTaskAttemptID srcTAID = createTAIDForTest(srcTaskID, 2); // Task0, Attempt 0
+
+ EventMetaData srcMeta = new EventMetaData(EventProducerConsumerType.OUTPUT, "consumerVertex", "producerVertex", srcTAID);
+
+ // Verification via a CompositeEvent
+ CompositeDataMovementEvent cdmEvent = new CompositeDataMovementEvent(0, destTasks.size(), "bytes".getBytes());
+ cdmEvent.setVersion(2); // AttemptNum
+ TezEvent tezEvent = new TezEvent(cdmEvent, srcMeta);
+ // Event setup to look like it would after the Vertex is done with it.
+
+ edge.sendTezEventToDestinationTasks(tezEvent);
+
+ ArgumentCaptor<Event> args = ArgumentCaptor.forClass(Event.class);
+ verify(eventHandler, times(destTasks.size())).handle(args.capture());
+
+ verifyEvents(args.getAllValues(), srcTAID, destTasks);
+
+
+ // Same Verification via regular DataMovementEvents
+ reset(eventHandler);
+ for (int i = 0 ; i < destTasks.size() ; i++) {
+ DataMovementEvent dmEvent = new DataMovementEvent(i, "bytes".getBytes());
+ dmEvent.setVersion(2);
+ tezEvent = new TezEvent(dmEvent, srcMeta);
+ edge.sendTezEventToDestinationTasks(tezEvent);
+ }
+ args = ArgumentCaptor.forClass(Event.class);
+ verify(eventHandler, times(destTasks.size())).handle(args.capture());
+
+ verifyEvents(args.getAllValues(), srcTAID, destTasks);
+
+ }
+
+ @SuppressWarnings("rawtypes")
+ private void verifyEvents(List<Event> events, TezTaskAttemptID srcTAID, LinkedHashMap<TezTaskID, Task> destTasks) {
+ int count = 0;
+
+ Iterator<Entry<TezTaskID, Task>> taskIter = destTasks.entrySet().iterator();
+
+ for (Event event : events) {
+ Entry<TezTaskID, Task> expEntry = taskIter.next();
+ assertTrue(event instanceof TaskEventAddTezEvent);
+ TaskEventAddTezEvent taEvent = (TaskEventAddTezEvent) event;
+ assertEquals(expEntry.getKey(), taEvent.getTaskID());
+ TezEvent tezEvent = taEvent.getTezEvent();
+
+ DataMovementEvent dmEvent = (DataMovementEvent)tezEvent.getEvent();
+ assertEquals(srcTAID.getId(), dmEvent.getVersion());
+ assertEquals(count, dmEvent.getSourceIndex());
+ assertEquals(srcTAID.getTaskID().getId(), dmEvent.getTargetIndex());
+ assertTrue(Arrays.equals("bytes".getBytes(), dmEvent.getUserPayload()));
+
+ count++;
+ }
+ }
+
+ private LinkedHashMap<TezTaskID, Task> mockTasks(TezVertexID vertexID, int numTasks) {
+ LinkedHashMap<TezTaskID, Task> tasks = new LinkedHashMap<TezTaskID, Task>();
+ for (int i = 0 ; i < numTasks ; i++) {
+ Task task = mock(Task.class);
+ TezTaskID taskID = TezTaskID.getInstance(vertexID, i);
+ doReturn(taskID).when(task).getTaskId();
+ tasks.put(taskID, task);
+ }
+ return tasks;
+ }
+
+ private Vertex mockVertex(String name, TezVertexID vertexID, LinkedHashMap<TezTaskID, Task> tasks) {
+ Vertex vertex = mock(Vertex.class);
+ doReturn(vertexID).when(vertex).getVertexId();
+ doReturn(name).when(vertex).getName();
+ doReturn(tasks).when(vertex).getTasks();
+ doReturn(tasks.size()).when(vertex).getTotalTasks();
+ for (Entry<TezTaskID, Task> entry : tasks.entrySet()) {
+ doReturn(entry.getValue()).when(vertex).getTask(eq(entry.getKey()));
+ doReturn(entry.getValue()).when(vertex).getTask(eq(entry.getKey().getId()));
+ }
+ return vertex;
+ }
+
+ private TezVertexID createVertexID(int id) {
+ TezDAGID dagID = TezDAGID.getInstance("1000", 1, 1);
+ TezVertexID vertexID = TezVertexID.getInstance(dagID, id);
+ return vertexID;
+ }
+
+ private TezTaskAttemptID createTAIDForTest(TezTaskID taskID, int taId) {
+ TezTaskAttemptID taskAttemptID = TezTaskAttemptID.getInstance(taskID, taId);
+ return taskAttemptID;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/tez-runtime-internals/pom.xml
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/pom.xml b/tez-runtime-internals/pom.xml
index 271c2ef..f00e2ac 100644
--- a/tez-runtime-internals/pom.xml
+++ b/tez-runtime-internals/pom.xml
@@ -49,6 +49,10 @@
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/EventMetaData.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/EventMetaData.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/EventMetaData.java
index b745987..77d266a 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/EventMetaData.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/EventMetaData.java
@@ -56,11 +56,6 @@ public class EventMetaData implements Writable {
private String edgeVertexName;
/**
- * i'th physical input/output that this event maps to.
- */
- private int index;
-
- /**
* Task Attempt ID
*/
private TezTaskAttemptID taskAttemptID;
@@ -114,8 +109,6 @@ public class EventMetaData implements Writable {
} else {
out.writeBoolean(false);
}
-
- out.writeInt(index);
}
@Override
@@ -130,15 +123,6 @@ public class EventMetaData implements Writable {
if (in.readBoolean()) {
taskAttemptID = TezTaskAttemptID.readTezTaskAttemptID(in);
}
- index = in.readInt();
- }
-
- public int getIndex() {
- return index;
- }
-
- public void setIndex(int index) {
- this.index = index;
}
@Override
@@ -147,6 +131,6 @@ public class EventMetaData implements Writable {
+ ", taskVertexName=" + taskVertexName
+ ", edgeVertexName=" + edgeVertexName
+ ", taskAttemptId=" + taskAttemptID
- + ", index=" + index + " }";
+ + " }";
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/EventType.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/EventType.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/EventType.java
index 6281b10..adefbd6 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/EventType.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/EventType.java
@@ -26,5 +26,6 @@ public enum EventType {
INPUT_FAILED_EVENT,
TASK_STATUS_UPDATE_EVENT,
VERTEX_MANAGER_EVENT,
- ROOT_INPUT_DATA_INFORMATION_EVENT
+ ROOT_INPUT_DATA_INFORMATION_EVENT,
+ COMPOSITE_DATA_MOVEMENT_EVENT
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezEvent.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezEvent.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezEvent.java
index af43d9a..51042c2 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezEvent.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezEvent.java
@@ -25,7 +25,9 @@ import java.io.IOException;
import org.apache.hadoop.io.Writable;
import org.apache.tez.dag.api.TezUncheckedException;
import org.apache.tez.runtime.api.Event;
+import org.apache.tez.runtime.api.events.CompositeDataMovementEvent;
import org.apache.tez.runtime.api.events.DataMovementEvent;
+import org.apache.tez.runtime.api.events.EventProtos.CompositeEventProto;
import org.apache.tez.runtime.api.events.EventProtos.DataMovementEventProto;
import org.apache.tez.runtime.api.events.EventProtos.InputFailedEventProto;
import org.apache.tez.runtime.api.events.EventProtos.InputReadErrorEventProto;
@@ -61,6 +63,8 @@ public class TezEvent implements Writable {
this.setSourceInfo(sourceInfo);
if (event instanceof DataMovementEvent) {
eventType = EventType.DATA_MOVEMENT_EVENT;
+ } else if (event instanceof CompositeDataMovementEvent) {
+ eventType = EventType.COMPOSITE_DATA_MOVEMENT_EVENT;
} else if (event instanceof VertexManagerEvent) {
eventType = EventType.VERTEX_MANAGER_EVENT;
} else if (event instanceof InputReadErrorEvent) {
@@ -129,6 +133,16 @@ public class TezEvent implements Writable {
}
eventBytes = dmBuilder.build().toByteArray();
break;
+ case COMPOSITE_DATA_MOVEMENT_EVENT:
+ CompositeDataMovementEvent cEvent = (CompositeDataMovementEvent) event;
+ CompositeEventProto.Builder cBuilder = CompositeEventProto.newBuilder();
+ cBuilder.setStartIndex(cEvent.getSourceIndexStart());
+ cBuilder.setEndIndex(cEvent.getSourceIndexEnd());
+ if (cEvent.getUserPayload() != null) {
+ cBuilder.setUserPayload(ByteString.copyFrom(cEvent.getUserPayload()));
+ }
+ eventBytes = cBuilder.build().toByteArray();
+ break;
case VERTEX_MANAGER_EVENT:
VertexManagerEvent vmEvt = (VertexManagerEvent) event;
VertexManagerEventProto.Builder vmBuilder = VertexManagerEventProto.newBuilder();
@@ -205,6 +219,11 @@ public class TezEvent implements Writable {
dmProto.getVersion(),
dmProto.getUserPayload() != null ? dmProto.getUserPayload().toByteArray() : null);
break;
+ case COMPOSITE_DATA_MOVEMENT_EVENT:
+ CompositeEventProto cProto = CompositeEventProto.parseFrom(eventBytes);
+ event = new CompositeDataMovementEvent(cProto.getStartIndex(), cProto.getEndIndex(),
+ cProto.hasUserPayload() ? cProto.getUserPayload().toByteArray() : null);
+ break;
case VERTEX_MANAGER_EVENT:
VertexManagerEventProto vmProto =
VertexManagerEventProto.parseFrom(eventBytes);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f55dbfb8/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileSortedOutput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileSortedOutput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileSortedOutput.java
index 92e5d81..05d37ed 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileSortedOutput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OnFileSortedOutput.java
@@ -30,7 +30,7 @@ import org.apache.tez.common.counters.TaskCounter;
import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.LogicalOutput;
import org.apache.tez.runtime.api.TezOutputContext;
-import org.apache.tez.runtime.api.events.DataMovementEvent;
+import org.apache.tez.runtime.api.events.CompositeDataMovementEvent;
import org.apache.tez.runtime.api.events.VertexManagerEvent;
import org.apache.tez.runtime.library.api.KeyValueWriter;
import org.apache.tez.runtime.library.common.sort.impl.ExternalSorter;
@@ -134,10 +134,9 @@ public class OnFileSortedOutput implements LogicalOutput {
List<Event> events = Lists.newArrayListWithCapacity(numOutputs+1);
events.add(vmEvent);
- for (int i = 0; i < numOutputs; i++) {
- DataMovementEvent event = new DataMovementEvent(i, payloadBytes);
- events.add(event);
- }
+ CompositeDataMovementEvent csdme = new CompositeDataMovementEvent(0, numOutputs, payloadBytes);
+ events.add(csdme);
+
return events;
}
-}
+}
\ No newline at end of file