Repository: tez Updated Branches: refs/heads/master 12fc2c708 -> 11aa17e01
http://git-wip-us.apache.org/repos/asf/tez/blob/11aa17e0/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/TaskAttemptInfo.java ---------------------------------------------------------------------- diff --git a/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/TaskAttemptInfo.java b/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/TaskAttemptInfo.java new file mode 100644 index 0000000..4c3fa97 --- /dev/null +++ b/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/TaskAttemptInfo.java @@ -0,0 +1,252 @@ +/** + * 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.history.parser.datamodel; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Maps; +import org.apache.tez.common.counters.DAGCounter; +import org.apache.tez.common.counters.TaskCounter; +import org.apache.tez.common.counters.TezCounter; +import org.codehaus.jettison.json.JSONException; +import org.codehaus.jettison.json.JSONObject; + +import java.util.Map; + +import static org.apache.hadoop.classification.InterfaceStability.Evolving; +import static org.apache.hadoop.classification.InterfaceAudience.Public; + +@Public +@Evolving +public class TaskAttemptInfo extends BaseInfo { + + private final String taskAttemptId; + private final long startTime; + private final long endTime; + private final String diagnostics; + private final String successfulAttemptId; + private final long scheduledTime; + private final String containerId; + private final String nodeId; + private final String status; + private final String logUrl; + + private TaskInfo taskInfo; + + private Container container; + + TaskAttemptInfo(JSONObject jsonObject) throws JSONException { + super(jsonObject); + + Preconditions.checkArgument( + jsonObject.getString(Constants.ENTITY_TYPE).equalsIgnoreCase + (Constants.TEZ_TASK_ATTEMPT_ID)); + + taskAttemptId = jsonObject.optString(Constants.ENTITY); + + //Parse additional Info + final JSONObject otherInfoNode = jsonObject.getJSONObject(Constants.OTHER_INFO); + startTime = otherInfoNode.optLong(Constants.START_TIME); + endTime = otherInfoNode.optLong(Constants.FINISH_TIME); + diagnostics = otherInfoNode.optString(Constants.DIAGNOSTICS); + successfulAttemptId = otherInfoNode.optString(Constants.SUCCESSFUL_ATTEMPT_ID); + scheduledTime = otherInfoNode.optLong(Constants.SCHEDULED_TIME); + + containerId = otherInfoNode.optString(Constants.CONTAINER_ID); + String id = otherInfoNode.optString(Constants.NODE_ID); + nodeId = (id != null) ? (id.split(":")[0]) : ""; + logUrl = otherInfoNode.optString(Constants.COMPLETED_LOGS_URL); + + status = otherInfoNode.optString(Constants.STATUS); + container = new Container(containerId, nodeId); + } + + void setTaskInfo(TaskInfo taskInfo) { + Preconditions.checkArgument(taskInfo != null, "Provide valid taskInfo"); + this.taskInfo = taskInfo; + taskInfo.addTaskAttemptInfo(this); + } + + @Override + public final long getStartTime() { + return startTime - (getTaskInfo().getVertexInfo().getDagInfo().getAbsStartTime()); + } + + @Override + public final long getFinishTime() { + return endTime - (getTaskInfo().getVertexInfo().getDagInfo().getAbsStartTime()); + } + + public final long getAbsStartTime() { + return startTime; + } + + public final long getAbsFinishTime() { + return endTime; + } + + public final long getAbsoluteScheduledTime() { + return scheduledTime; + } + + public final long getTimeTaken() { + return getFinishTime() - getStartTime(); + } + + public final long getScheduledTime() { + return scheduledTime - (getTaskInfo().getVertexInfo().getDagInfo().getAbsStartTime()); + } + + @Override + public final String getDiagnostics() { + return diagnostics; + } + + public static TaskAttemptInfo create(JSONObject taskInfoObject) throws JSONException { + return new TaskAttemptInfo(taskInfoObject); + } + + public final boolean isLocalityInfoAvailable() { + Map<String, TezCounter> dataLocalTask = getCounter(DAGCounter.class.getName(), + DAGCounter.DATA_LOCAL_TASKS.toString()); + Map<String, TezCounter> rackLocalTask = getCounter(DAGCounter.class.getName(), + DAGCounter.RACK_LOCAL_TASKS.toString()); + + Map<String, TezCounter> otherLocalTask = getCounter(DAGCounter.class.getName(), + DAGCounter.OTHER_LOCAL_TASKS.toString()); + + if (!dataLocalTask.isEmpty() || !rackLocalTask.isEmpty() || !otherLocalTask.isEmpty()) { + return true; + } + return false; + } + + public final TezCounter getLocalityInfo() { + Map<String, TezCounter> dataLocalTask = getCounter(DAGCounter.class.getName(), + DAGCounter.DATA_LOCAL_TASKS.toString()); + Map<String, TezCounter> rackLocalTask = getCounter(DAGCounter.class.getName(), + DAGCounter.RACK_LOCAL_TASKS.toString()); + Map<String, TezCounter> otherLocalTask = getCounter(DAGCounter.class.getName(), + DAGCounter.OTHER_LOCAL_TASKS.toString()); + + if (!dataLocalTask.isEmpty()) { + return dataLocalTask.get(DAGCounter.class.getName()); + } + + if (!rackLocalTask.isEmpty()) { + return rackLocalTask.get(DAGCounter.class.getName()); + } + + if (!otherLocalTask.isEmpty()) { + return otherLocalTask.get(DAGCounter.class.getName()); + } + return null; + } + + public final TaskInfo getTaskInfo() { + return taskInfo; + } + + public final String getTaskAttemptId() { + return taskAttemptId; + } + + public final String getSuccessfulAttemptId() { + return successfulAttemptId; + } + + public final String getNodeId() { + return nodeId; + } + + public final String getStatus() { + return status; + } + + public final Container getContainer() { + return container; + } + + public final String getLogURL() { + return logUrl; + } + + /** + * Get merge counter per source. Available in case of reducer task + * + * @return Map<String, TezCounter> merge phase time at every counter group level + */ + public final Map<String, TezCounter> getMergePhaseTime() { + return getCounter(null, TaskCounter.MERGE_PHASE_TIME.name()); + } + + /** + * Get shuffle counter per source. Available in case of shuffle + * + * @return Map<String, TezCounter> shuffle phase time at every counter group level + */ + public final Map<String, TezCounter> getShufflePhaseTime() { + return getCounter(null, TaskCounter.SHUFFLE_PHASE_TIME.name()); + } + + /** + * Get OUTPUT_BYTES counter per source. Available in case of map outputs + * + * @return Map<String, TezCounter> output bytes counter at every counter group + */ + public final Map<String, TezCounter> getTaskOutputBytes() { + return getCounter(null, TaskCounter.OUTPUT_BYTES.name()); + } + + /** + * Get number of spills per source. (SPILLED_RECORDS / OUTPUT_RECORDS) + * + * @return Map<String, Long> spill count details + */ + public final Map<String, Float> getSpillCount() { + Map<String, TezCounter> outputRecords = getCounter(null, "OUTPUT_RECORDS"); + Map<String, TezCounter> spilledRecords = getCounter(null, "SPILLED_RECORDS"); + Map<String, Float> result = Maps.newHashMap(); + for (Map.Entry<String, TezCounter> entry : spilledRecords.entrySet()) { + String source = entry.getKey(); + long spilledVal = entry.getValue().getValue(); + long outputVal = outputRecords.get(source).getValue(); + result.put(source, (spilledVal * 1.0f) / (outputVal * 1.0f)); + } + return result; + } + + public String toString() { + StringBuilder sb = new StringBuilder(); + sb.append("["); + sb.append("taskAttemptId=").append(getTaskAttemptId()).append(", "); + sb.append("scheduledTime=").append(getScheduledTime()).append(", "); + sb.append("startTime=").append(getStartTime()).append(", "); + sb.append("finishTime=").append(getFinishTime()).append(", "); + sb.append("timeTaken=").append(getTimeTaken()).append(", "); + sb.append("events=").append(getEvents()).append(", "); + sb.append("diagnostics=").append(getDiagnostics()).append(", "); + sb.append("successfulAttempId=").append(getSuccessfulAttemptId()).append(", "); + sb.append("container=").append(getContainer()).append(", "); + sb.append("nodeId=").append(getNodeId()).append(", "); + sb.append("logURL=").append(getLogURL()).append(", "); + sb.append("status=").append(getStatus()); + sb.append("]"); + return sb.toString(); + } +} http://git-wip-us.apache.org/repos/asf/tez/blob/11aa17e0/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/TaskInfo.java ---------------------------------------------------------------------- diff --git a/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/TaskInfo.java b/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/TaskInfo.java new file mode 100644 index 0000000..cb966a4 --- /dev/null +++ b/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/TaskInfo.java @@ -0,0 +1,341 @@ +/** + * 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.history.parser.datamodel; + +import com.google.common.base.Preconditions; +import com.google.common.base.Predicate; +import com.google.common.collect.Iterables; +import com.google.common.collect.LinkedHashMultimap; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.google.common.collect.Multimap; +import com.google.common.collect.Multimaps; +import com.google.common.collect.Ordering; +import org.apache.tez.dag.api.oldrecords.TaskAttemptState; +import org.codehaus.jettison.json.JSONException; +import org.codehaus.jettison.json.JSONObject; + +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Map; + +import static org.apache.hadoop.classification.InterfaceAudience.Public; +import static org.apache.hadoop.classification.InterfaceStability.Evolving; + +@Public +@Evolving +public class TaskInfo extends BaseInfo { + + private final long startTime; + private final long endTime; + private final String diagnostics; + private final String successfulAttemptId; + private final long scheduledTime; + private final String status; + private final String taskId; + + private VertexInfo vertexInfo; + + private Map<String, TaskAttemptInfo> attemptInfoMap = Maps + .newHashMap(); + + TaskInfo(JSONObject jsonObject) throws JSONException { + super(jsonObject); + + Preconditions.checkArgument( + jsonObject.getString(Constants.ENTITY_TYPE).equalsIgnoreCase + (Constants.TEZ_TASK_ID)); + + taskId = jsonObject.optString(Constants.ENTITY); + + //Parse additional Info + final JSONObject otherInfoNode = jsonObject.getJSONObject(Constants.OTHER_INFO); + startTime = otherInfoNode.optLong(Constants.START_TIME); + endTime = otherInfoNode.optLong(Constants.FINISH_TIME); + diagnostics = otherInfoNode.optString(Constants.DIAGNOSTICS); + successfulAttemptId = otherInfoNode.optString(Constants.SUCCESSFUL_ATTEMPT_ID); + scheduledTime = otherInfoNode.optLong(Constants.SCHEDULED_TIME); + status = otherInfoNode.optString(Constants.STATUS); + } + + @Override + public final long getStartTime() { + return startTime - (vertexInfo.getDagInfo().getAbsStartTime()); + } + + public final long getAbsStartTime() { + return startTime; + } + + public final long getAbsFinishTime() { + return endTime; + } + + @Override + public final long getFinishTime() { + long taskFinishTime = endTime - (vertexInfo.getDagInfo().getAbsStartTime()); + if (taskFinishTime < 0) { + //probably vertex is not complete or failed in middle. get the last task attempt time + for (TaskAttemptInfo attemptInfo : getTaskAttempts()) { + taskFinishTime = (attemptInfo.getFinishTime() > taskFinishTime) + ? attemptInfo.getFinishTime() : taskFinishTime; + } + } + return taskFinishTime; + } + + @Override + public final String getDiagnostics() { + return diagnostics; + } + + public static TaskInfo create(JSONObject taskInfoObject) throws + JSONException { + return new TaskInfo(taskInfoObject); + } + + void addTaskAttemptInfo(TaskAttemptInfo taskAttemptInfo) { + attemptInfoMap.put(taskAttemptInfo.getTaskAttemptId(), taskAttemptInfo); + } + + void setVertexInfo(VertexInfo vertexInfo) { + Preconditions.checkArgument(vertexInfo != null, "Provide valid vertexInfo"); + this.vertexInfo = vertexInfo; + //link it to vertex + vertexInfo.addTaskInfo(this); + } + + public final VertexInfo getVertexInfo() { + return vertexInfo; + } + + /** + * Get all task attempts + * + * @return list of task attempt info + */ + public final List<TaskAttemptInfo> getTaskAttempts() { + List<TaskAttemptInfo> attemptsList = Lists.newLinkedList(attemptInfoMap.values()); + Collections.sort(attemptsList, orderingOnAttemptStartTime()); + return Collections.unmodifiableList(attemptsList); + } + + /** + * Get list of failed tasks + * + * @return List<TaskAttemptInfo> + */ + public final List<TaskAttemptInfo> getFailedTaskAttempts() { + return getTaskAttempts(TaskAttemptState.FAILED); + } + + /** + * Get list of killed tasks + * + * @return List<TaskAttemptInfo> + */ + public final List<TaskAttemptInfo> getKilledTaskAttempts() { + return getTaskAttempts(TaskAttemptState.KILLED); + } + + /** + * Get list of failed tasks + * + * @return List<TaskAttemptInfo> + */ + public final List<TaskAttemptInfo> getSuccessfulTaskAttempts() { + return getTaskAttempts(TaskAttemptState.SUCCEEDED); + } + + /** + * Get list of tasks belonging to a specific state + * + * @param state + * @return Collection<TaskAttemptInfo> + */ + public final List<TaskAttemptInfo> getTaskAttempts(final TaskAttemptState state) { + return Collections.unmodifiableList(Lists.newLinkedList(Iterables.filter(Lists.newLinkedList + (attemptInfoMap.values()), new Predicate<TaskAttemptInfo>() { + @Override public boolean apply(TaskAttemptInfo input) { + return input.getStatus() != null && input.getStatus().equals(state.toString()); + } + } + ) + ) + ); + } + + /** + * Get the set of containers on which the task attempts ran for this task + * + * @return Multimap<Container, TaskAttemptInfo> task attempt details at container level + */ + public final Multimap<Container, TaskAttemptInfo> getContainersMapping() { + Multimap<Container, TaskAttemptInfo> containerMapping = LinkedHashMultimap.create(); + for (TaskAttemptInfo attemptInfo : getTaskAttempts()) { + containerMapping.put(attemptInfo.getContainer(), attemptInfo); + } + return Multimaps.unmodifiableMultimap(containerMapping); + } + + /** + * Get the successful task attempt + * + * @return TaskAttemptInfo + */ + public final TaskAttemptInfo getSuccessfulTaskAttempt() { + for (TaskAttemptInfo attemptInfo : getTaskAttempts()) { + if (attemptInfo.getStatus().equalsIgnoreCase(TaskAttemptState.SUCCEEDED.toString())) { + return attemptInfo; + } + } + return null; + } + + /** + * Get last task attempt to finish + * + * @return TaskAttemptInfo + */ + public final TaskAttemptInfo getLastTaskAttemptToFinish() { + List<TaskAttemptInfo> attemptsList = getTaskAttempts(); + if (attemptsList.isEmpty()) { + return null; + } + + return Ordering.from(new Comparator<TaskAttemptInfo>() { + @Override public int compare(TaskAttemptInfo o1, TaskAttemptInfo o2) { + return (o1.getFinishTime() < o2.getFinishTime()) ? -1 : + ((o1.getFinishTime() == o2.getFinishTime()) ? + 0 : 1); + } + }).max(attemptsList); + } + + /** + * Get average task attempt duration. Includes succesful and failed tasks + * + * @return float + */ + public final float getAvgTaskAttemptDuration() { + float totalTaskDuration = 0; + List<TaskAttemptInfo> attemptsList = getTaskAttempts(); + if (attemptsList.size() == 0) { + return 0; + } + for (TaskAttemptInfo attemptInfo : attemptsList) { + totalTaskDuration += attemptInfo.getTimeTaken(); + } + return ((totalTaskDuration * 1.0f) / attemptsList.size()); + } + + private Ordering<TaskAttemptInfo> orderingOnTimeTaken() { + return Ordering.from(new Comparator<TaskAttemptInfo>() { + @Override public int compare(TaskAttemptInfo o1, TaskAttemptInfo o2) { + return (o1.getTimeTaken() < o2.getTimeTaken()) ? -1 : + ((o1.getTimeTaken() == o2.getTimeTaken()) ? + 0 : 1); + } + }); + } + + private Ordering<TaskAttemptInfo> orderingOnAttemptStartTime() { + return Ordering.from(new Comparator<TaskAttemptInfo>() { + @Override public int compare(TaskAttemptInfo o1, TaskAttemptInfo o2) { + return (o1.getStartTime() < o2.getStartTime()) ? -1 : + ((o1.getStartTime() == o2.getStartTime()) ? 0 : 1); + } + }); + } + + /** + * Get min task attempt duration. This includes successful/failed task attempts as well + * + * @return long + */ + public final long getMinTaskAttemptDuration() { + List<TaskAttemptInfo> attemptsList = getTaskAttempts(); + if (attemptsList.isEmpty()) { + return 0; + } + + return orderingOnTimeTaken().min(attemptsList).getTimeTaken(); + } + + /** + * Get max task attempt duration. This includes successful/failed task attempts as well + * + * @return long + */ + public final long getMaxTaskAttemptDuration() { + List<TaskAttemptInfo> attemptsList = getTaskAttempts(); + if (attemptsList.isEmpty()) { + return 0; + } + + return orderingOnTimeTaken().max(attemptsList).getTimeTaken(); + } + + public final int getNumberOfTaskAttempts() { + return getTaskAttempts().size(); + } + + public final String getStatus() { + return status; + } + + public final String getTaskId() { + return taskId; + } + + public final long getTimeTaken() { + return getFinishTime() - getStartTime(); + } + + public final String getSuccessfulAttemptId() { + return successfulAttemptId; + } + + public final long getAbsoluteScheduleTime() { + return scheduledTime; + } + + public final long getScheduledTime() { + return scheduledTime - this.getVertexInfo().getDagInfo().getAbsStartTime(); + } + + @Override + public String toString() { + StringBuilder sb = new StringBuilder(); + sb.append("["); + sb.append("taskId=").append(getTaskId()).append(", "); + sb.append("scheduledTime=").append(getAbsoluteScheduleTime()).append(", "); + sb.append("startTime=").append(getStartTime()).append(", "); + sb.append("finishTime=").append(getFinishTime()).append(", "); + sb.append("timeTaken=").append(getTimeTaken()).append(", "); + sb.append("events=").append(getEvents()).append(", "); + sb.append("diagnostics=").append(getDiagnostics()).append(", "); + sb.append("successfulAttempId=").append(getSuccessfulAttemptId()).append(", "); + sb.append("status=").append(getStatus()).append(", "); + sb.append("vertexName=").append(getVertexInfo().getVertexName()); + sb.append("]"); + return sb.toString(); + } +} http://git-wip-us.apache.org/repos/asf/tez/blob/11aa17e0/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/VersionInfo.java ---------------------------------------------------------------------- diff --git a/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/VersionInfo.java b/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/VersionInfo.java new file mode 100644 index 0000000..97d18cd --- /dev/null +++ b/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/VersionInfo.java @@ -0,0 +1,45 @@ +/** + * 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 + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * 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.history.parser.datamodel; + +public class VersionInfo { + + private final String buildTime; + private final String revision; + private final String version; + + public VersionInfo(String buildTime, String revision, String version) { + this.buildTime = buildTime; + this.revision = revision; + this.version = version; + } + + public String getBuildTime() { + return buildTime; + } + + public String getRevision() { + return revision; + } + + public String getVersion() { + return version; + } + +} http://git-wip-us.apache.org/repos/asf/tez/blob/11aa17e0/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/VertexInfo.java ---------------------------------------------------------------------- diff --git a/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/VertexInfo.java b/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/VertexInfo.java new file mode 100644 index 0000000..3445adb --- /dev/null +++ b/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/datamodel/VertexInfo.java @@ -0,0 +1,536 @@ +/** + * 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.history.parser.datamodel; + +import com.google.common.base.Preconditions; +import com.google.common.base.Predicate; +import com.google.common.collect.Iterables; +import com.google.common.collect.LinkedHashMultimap; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.google.common.collect.Multimap; +import com.google.common.collect.Multimaps; +import com.google.common.collect.Ordering; +import org.apache.tez.dag.api.oldrecords.TaskState; +import org.codehaus.jettison.json.JSONException; +import org.codehaus.jettison.json.JSONObject; + +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Map; + +import static org.apache.hadoop.classification.InterfaceAudience.Public; +import static org.apache.hadoop.classification.InterfaceStability.Evolving; + +@Public +@Evolving +public class VertexInfo extends BaseInfo { + + private final String vertexName; + private final long endTime; + private final long initTime; + private final String diagnostics; + private final String processorClass; + + private final int numTasks; + private final int failedTasks; + private final int completedTasks; + private final int succeededTasks; + private final int killedTasks; + private final int numFailedTaskAttempts; + + private final String status; + + private final long startTime; + + //TaskID --> TaskInfo for internal reference + private Map<String, TaskInfo> taskInfoMap; + + private final List<EdgeInfo> inEdgeList; + private final List<EdgeInfo> outEdgeList; + + private final List<AdditionalInputOutputDetails> additionalInputInfoList; + private final List<AdditionalInputOutputDetails> additionalOutputInfoList; + + private DagInfo dagInfo; + + VertexInfo(JSONObject jsonObject) throws JSONException { + super(jsonObject); + + Preconditions.checkArgument( + jsonObject.getString(Constants.ENTITY_TYPE).equalsIgnoreCase + (Constants.TEZ_VERTEX_ID)); + + taskInfoMap = Maps.newHashMap(); + + inEdgeList = Lists.newLinkedList(); + outEdgeList = Lists.newLinkedList(); + additionalInputInfoList = Lists.newLinkedList(); + additionalOutputInfoList = Lists.newLinkedList(); + + //Parse additional Info + JSONObject otherInfoNode = jsonObject.getJSONObject(Constants.OTHER_INFO); + startTime = otherInfoNode.optLong(Constants.START_TIME); + initTime = otherInfoNode.optLong(Constants.INIT_TIME); + endTime = otherInfoNode.optLong(Constants.FINISH_TIME); + diagnostics = otherInfoNode.optString(Constants.DIAGNOSTICS); + numTasks = otherInfoNode.optInt(Constants.NUM_TASKS); + failedTasks = otherInfoNode.optInt(Constants.NUM_FAILED_TASKS); + succeededTasks = + otherInfoNode.optInt(Constants.NUM_SUCCEEDED_TASKS); + completedTasks = + otherInfoNode.optInt(Constants.NUM_COMPLETED_TASKS); + killedTasks = otherInfoNode.optInt(Constants.NUM_KILLED_TASKS); + numFailedTaskAttempts = + otherInfoNode.optInt(Constants.NUM_FAILED_TASKS_ATTEMPTS); + vertexName = otherInfoNode.optString(Constants.VERTEX_NAME); + processorClass = otherInfoNode.optString(Constants.PROCESSOR_CLASS_NAME); + status = otherInfoNode.optString(Constants.STATUS); + } + + public static VertexInfo create(JSONObject vertexInfoObject) throws + JSONException { + return new VertexInfo(vertexInfoObject); + } + + /** + * Update edge details with source and destination vertex objects. + */ + private void updateEdgeInfo() { + if (dagInfo.getNumVertices() == dagInfo.getVertices().size()) { + //We can update EdgeInfo when all vertices are parsed + Map<String, VertexInfo> vertexMapping = dagInfo.getVertexMapping(); + for (EdgeInfo edge : dagInfo.getEdges()) { + VertexInfo sourceVertex = vertexMapping.get(edge.getInputVertexName()); + VertexInfo destinationVertex = vertexMapping.get(edge.getOutputVertexName()); + edge.setSourceVertex(sourceVertex); + edge.setDestinationVertex(destinationVertex); + } + } + } + + void addTaskInfo(TaskInfo taskInfo) { + this.taskInfoMap.put(taskInfo.getTaskId(), taskInfo); + } + + void setAdditionalInputInfoList(List<AdditionalInputOutputDetails> additionalInputInfoList) { + this.additionalInputInfoList.clear(); + this.additionalInputInfoList.addAll(additionalInputInfoList); + } + + void setAdditionalOutputInfoList(List<AdditionalInputOutputDetails> additionalOutputInfoList) { + this.additionalOutputInfoList.clear(); + this.additionalOutputInfoList.addAll(additionalOutputInfoList); + } + + void addInEdge(EdgeInfo edgeInfo) { + this.inEdgeList.add(edgeInfo); + } + + void addOutEdge(EdgeInfo edgeInfo) { + this.outEdgeList.add(edgeInfo); + } + + void setDagInfo(DagInfo dagInfo) { + Preconditions.checkArgument(dagInfo != null, "Provide valid dagInfo"); + this.dagInfo = dagInfo; + //link vertex to dagInfo + dagInfo.addVertexInfo(this); + updateEdgeInfo(); + } + + @Override + public final long getStartTime() { + return startTime - (dagInfo.getAbsStartTime()); + } + + public final long getFirstTaskStartTime() { + return getFirstTaskToStart().getStartTime(); + } + + public final long getLastTaskFinishTime() { + if (getLastTaskToFinish() == null || getLastTaskToFinish().getFinishTime() < 0) { + return dagInfo.getFinishTime(); + } + return getLastTaskToFinish().getFinishTime(); + } + + public final long getAbsStartTime() { + return startTime; + } + + public final long getAbsFinishTime() { + return endTime; + } + + public final long getAbsoluteInitTime() { + return initTime; + } + + @Override + public final long getFinishTime() { + long vertexEndTime = endTime - (dagInfo.getAbsStartTime()); + if (vertexEndTime < 0) { + //probably vertex is not complete or failed in middle. get the last task attempt time + for (TaskInfo taskInfo : getTasks()) { + vertexEndTime = (taskInfo.getFinishTime() > vertexEndTime) + ? taskInfo.getFinishTime() : vertexEndTime; + } + } + return vertexEndTime; + } + + @Override + public final String getDiagnostics() { + return diagnostics; + } + + public final String getVertexName() { + return vertexName; + } + + //Quite possible that getFinishTime is not yet recorded for failed vertices (or killed vertices) + //Start time of vertex infers that the dependencies are done and AM has inited it. + public final long getTimeTaken() { + return (getFinishTime() - getStartTime()); + } + + //Time taken for last task to finish - time taken for first task to start + public final long getTimeTakenForTasks() { + return (getLastTaskFinishTime() - getFirstTaskStartTime()); + } + + public final long getInitTime() { + return initTime - dagInfo.getAbsStartTime(); + } + + public final int getNumTasks() { + return numTasks; + } + + public final int getFailedTasksCount() { + return failedTasks; + } + + public final int getKilledTasksCount() { + return killedTasks; + } + + public final int getCompletedTasksCount() { + return completedTasks; + } + + public final int getSucceededTasksCount() { + return succeededTasks; + } + + public final int getNumFailedTaskAttemptsCount() { + return numFailedTaskAttempts; + } + + public final String getProcessorClassName() { + return processorClass; + + } + + /** + * Get all tasks + * + * @return list of taskInfo + */ + public final List<TaskInfo> getTasks() { + List<TaskInfo> taskInfoList = Lists.newLinkedList(taskInfoMap.values()); + Collections.sort(taskInfoList, orderingOnStartTime()); + return Collections.unmodifiableList(taskInfoList); + } + + /** + * Get list of failed tasks + * + * @return List<TaskAttemptInfo> + */ + public final List<TaskInfo> getFailedTasks() { + return getTasks(TaskState.FAILED); + } + + /** + * Get list of killed tasks + * + * @return List<TaskAttemptInfo> + */ + public final List<TaskInfo> getKilledTasks() { + return getTasks(TaskState.KILLED); + } + + /** + * Get list of failed tasks + * + * @return List<TaskAttemptInfo> + */ + public final List<TaskInfo> getSuccessfulTasks() { + return getTasks(TaskState.SUCCEEDED); + } + + /** + * Get list of tasks belonging to a specific state + * + * @param state + * @return List<TaskAttemptInfo> + */ + public final List<TaskInfo> getTasks(final TaskState state) { + return Collections.unmodifiableList(Lists.newLinkedList(Iterables.filter(Lists.newLinkedList + (taskInfoMap.values()), new Predicate<TaskInfo>() { + @Override public boolean apply(TaskInfo input) { + return input.getStatus() != null && input.getStatus().equals(state.toString()); + } + } + ) + ) + ); + } + + /** + * Get source vertices for this vertex + * + * @return List<VertexInfo> list of incoming vertices to this vertex + */ + public final List<VertexInfo> getInputVertices() { + List<VertexInfo> inputVertices = Lists.newLinkedList(); + for (EdgeInfo edge : inEdgeList) { + inputVertices.add(edge.getSourceVertex()); + } + return Collections.unmodifiableList(inputVertices); + } + + /** + * Get destination vertices for this vertex + * + * @return List<VertexInfo> list of output vertices + */ + public final List<VertexInfo> getOutputVertices() { + List<VertexInfo> outputVertices = Lists.newLinkedList(); + for (EdgeInfo edge : outEdgeList) { + outputVertices.add(edge.getDestinationVertex()); + } + return Collections.unmodifiableList(outputVertices); + } + + public List<TaskAttemptInfo> getTaskAttempts() { + List<TaskAttemptInfo> taskAttemptInfos = Lists.newLinkedList(); + for (TaskInfo taskInfo : getTasks()) { + taskAttemptInfos.addAll(taskInfo.getTaskAttempts()); + } + Collections.sort(taskAttemptInfos, orderingOnAttemptStartTime()); + return Collections.unmodifiableList(taskAttemptInfos); + } + + public final TaskInfo getTask(String taskId) { + return taskInfoMap.get(taskId); + } + + /** + * Get incoming edge information for a specific vertex + * + * @return List<EdgeInfo> list of input edges on this vertex + */ + public final List<EdgeInfo> getInputEdges() { + return Collections.unmodifiableList(inEdgeList); + } + + /** + * Get outgoing edge information for a specific vertex + * + * @return List<EdgeInfo> list of output edges on this vertex + */ + public final List<EdgeInfo> getOutputEdges() { + return Collections.unmodifiableList(outEdgeList); + } + + public final Multimap<Container, TaskAttemptInfo> getContainersMapping() { + Multimap<Container, TaskAttemptInfo> containerMapping = LinkedHashMultimap.create(); + for (TaskAttemptInfo attemptInfo : getTaskAttempts()) { + containerMapping.put(attemptInfo.getContainer(), attemptInfo); + } + return Multimaps.unmodifiableMultimap(containerMapping); + } + + /** + * Get first task to start + * + * @return TaskInfo + */ + public final TaskInfo getFirstTaskToStart() { + List<TaskInfo> taskInfoList = Lists.newLinkedList(taskInfoMap.values()); + if (taskInfoList.size() == 0) { + return null; + } + Collections.sort(taskInfoList, new Comparator<TaskInfo>() { + @Override public int compare(TaskInfo o1, TaskInfo o2) { + return (o1.getStartTime() < o2.getStartTime()) ? -1 : + ((o1.getStartTime() == o2.getStartTime()) ? + 0 : 1); + } + }); + return taskInfoList.get(0); + } + + /** + * Get last task to finish + * + * @return TaskInfo + */ + public final TaskInfo getLastTaskToFinish() { + List<TaskInfo> taskInfoList = Lists.newLinkedList(taskInfoMap.values()); + if (taskInfoList.size() == 0) { + return null; + } + Collections.sort(taskInfoList, new Comparator<TaskInfo>() { + @Override public int compare(TaskInfo o1, TaskInfo o2) { + return (o1.getFinishTime() > o2.getFinishTime()) ? -1 : + ((o1.getStartTime() == o2.getStartTime()) ? + 0 : 1); + } + }); + return taskInfoList.get(0); + } + + /** + * Get average task duration + * + * @return long + */ + public final float getAvgTaskDuration() { + float totalTaskDuration = 0; + List<TaskInfo> tasksList = getTasks(); + if (tasksList.size() == 0) { + return 0; + } + for (TaskInfo taskInfo : tasksList) { + totalTaskDuration += taskInfo.getTimeTaken(); + } + return ((totalTaskDuration * 1.0f) / tasksList.size()); + } + + /** + * Get min task duration in vertex + * + * @return long + */ + public final long getMinTaskDuration() { + TaskInfo taskInfo = getMinTaskDurationTask(); + return (taskInfo != null) ? taskInfo.getTimeTaken() : 0; + } + + /** + * Get max task duration in vertex + * + * @return long + */ + public final long getMaxTaskDuration() { + TaskInfo taskInfo = getMaxTaskDurationTask(); + return (taskInfo != null) ? taskInfo.getTimeTaken() : 0; + } + + private Ordering<TaskInfo> orderingOnTimeTaken() { + return Ordering.from(new Comparator<TaskInfo>() { + @Override public int compare(TaskInfo o1, TaskInfo o2) { + return (o1.getTimeTaken() < o2.getTimeTaken()) ? -1 : + ((o1.getTimeTaken() == o2.getTimeTaken()) ? 0 : 1); + } + }); + } + + private Ordering<TaskInfo> orderingOnStartTime() { + return Ordering.from(new Comparator<TaskInfo>() { + @Override public int compare(TaskInfo o1, TaskInfo o2) { + return (o1.getStartTime() < o2.getStartTime()) ? -1 : + ((o1.getStartTime() == o2.getStartTime()) ? 0 : 1); + } + }); + } + + private Ordering<TaskAttemptInfo> orderingOnAttemptStartTime() { + return Ordering.from(new Comparator<TaskAttemptInfo>() { + @Override public int compare(TaskAttemptInfo o1, TaskAttemptInfo o2) { + return (o1.getStartTime() < o2.getStartTime()) ? -1 : + ((o1.getStartTime() == o2.getStartTime()) ? 0 : 1); + } + }); + } + + /** + * Get min task duration in vertex + * + * @return TaskInfo + */ + public final TaskInfo getMinTaskDurationTask() { + List<TaskInfo> taskInfoList = getTasks(); + if (taskInfoList.size() == 0) { + return null; + } + + return orderingOnTimeTaken().min(taskInfoList); + } + + /** + * Get max task duration in vertex + * + * @return TaskInfo + */ + public final TaskInfo getMaxTaskDurationTask() { + List<TaskInfo> taskInfoList = getTasks(); + if (taskInfoList.size() == 0) { + return null; + } + return orderingOnTimeTaken().max(taskInfoList); + } + + public final String getStatus() { + return status; + } + + public final DagInfo getDagInfo() { + return dagInfo; + } + + @Override + public String toString() { + StringBuilder sb = new StringBuilder(); + sb.append("["); + sb.append("vertexName=").append(getVertexName()).append(", "); + sb.append("events=").append(getEvents()).append(", "); + sb.append("initTime=").append(getInitTime()).append(", "); + sb.append("startTime=").append(getStartTime()).append(", "); + sb.append("endTime=").append(getFinishTime()).append(", "); + sb.append("timeTaken=").append(getTimeTaken()).append(", "); + sb.append("diagnostics=").append(getDiagnostics()).append(", "); + sb.append("numTasks=").append(getNumTasks()).append(", "); + sb.append("processorClassName=").append(getProcessorClassName()).append(", "); + sb.append("numCompletedTasks=").append(getCompletedTasksCount()).append(", "); + sb.append("numFailedTaskAttempts=").append(getNumFailedTaskAttemptsCount()).append(", "); + sb.append("numSucceededTasks=").append(getSucceededTasksCount()).append(", "); + sb.append("numFailedTasks=").append(getFailedTasks()).append(", "); + sb.append("numKilledTasks=").append(getKilledTasks()).append(", "); + sb.append("tasksCount=").append(taskInfoMap.size()).append(", "); + sb.append("status=").append(getStatus()); + sb.append("]"); + return sb.toString(); + } +} http://git-wip-us.apache.org/repos/asf/tez/blob/11aa17e0/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/utils/Utils.java ---------------------------------------------------------------------- diff --git a/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/utils/Utils.java b/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/utils/Utils.java new file mode 100644 index 0000000..16ef4c9 --- /dev/null +++ b/tez-plugins/tez-history-parser/src/main/java/org/apache/tez/history/parser/utils/Utils.java @@ -0,0 +1,121 @@ +/** + * 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.history.parser.utils; + +import org.apache.directory.api.util.Strings; +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.log4j.ConsoleAppender; +import org.apache.log4j.Level; +import org.apache.log4j.Logger; +import org.apache.log4j.PatternLayout; +import org.apache.tez.history.parser.datamodel.Constants; +import org.apache.tez.history.parser.datamodel.Event; +import org.apache.tez.common.counters.CounterGroup; +import org.apache.tez.common.counters.TezCounter; +import org.apache.tez.common.counters.TezCounters; +import org.codehaus.jettison.json.JSONArray; +import org.codehaus.jettison.json.JSONException; +import org.codehaus.jettison.json.JSONObject; + +import java.util.List; + [email protected] +public class Utils { + + private static final String LOG4J_CONFIGURATION = "log4j.configuration"; + + /** + * Parse tez counters from json + * + * @param jsonObject + * @return TezCounters + * @throws JSONException + */ + public static TezCounters parseTezCountersFromJSON(JSONObject jsonObject) + throws JSONException { + TezCounters counters = new TezCounters(); + + if (jsonObject == null) { + return counters; //empty counters. + } + + final JSONArray counterGroupNodes = jsonObject.optJSONArray(Constants.COUNTER_GROUPS); + if (counterGroupNodes != null) { + for (int i = 0; i < counterGroupNodes.length(); i++) { + JSONObject counterGroupNode = counterGroupNodes.optJSONObject(i); + final String groupName = counterGroupNode.optString(Constants.COUNTER_GROUP_NAME); + final String groupDisplayName = counterGroupNode.optString( + Constants.COUNTER_GROUP_DISPLAY_NAME); + + CounterGroup group = counters.addGroup(groupName, groupDisplayName); + + final JSONArray counterNodes = counterGroupNode.optJSONArray(Constants.COUNTERS); + + //Parse counter nodes + for (int j = 0; j < counterNodes.length(); j++) { + JSONObject counterNode = counterNodes.optJSONObject(j); + final String counterName = counterNode.getString(Constants.COUNTER_NAME); + final String counterDisplayName = + counterNode.getString(Constants.COUNTER_DISPLAY_NAME); + final long counterValue = counterNode.getLong(Constants.COUNTER_VALUE); + TezCounter counter = group.findCounter( + counterName, + counterDisplayName); + counter.setValue(counterValue); + } + } + } + return counters; + } + + /** + * Parse events from json + * + * @param eventNodes + * @param eventList + * @throws JSONException + */ + public static void parseEvents(JSONArray eventNodes, List<Event> eventList) throws + JSONException { + if (eventNodes == null) { + return; + } + for (int i = 0; i < eventNodes.length(); i++) { + JSONObject eventNode = eventNodes.optJSONObject(i); + final String eventInfo = eventNode.optString(Constants.EVENT_INFO); + final String eventType = eventNode.optString(Constants.EVENT_TYPE); + final long time = eventNode.optLong(Constants.EVENT_TIME_STAMP); + + Event event = new Event(eventInfo, eventType, time); + + eventList.add(event); + + } + } + + public static void setupRootLogger() { + if (Strings.isEmpty(System.getProperty(LOG4J_CONFIGURATION))) { + //By default print to console with INFO level + Logger.getRootLogger(). + addAppender(new ConsoleAppender(new PatternLayout(PatternLayout.TTCC_CONVERSION_PATTERN))); + Logger.getRootLogger().setLevel(Level.INFO); + } + } + +} http://git-wip-us.apache.org/repos/asf/tez/blob/11aa17e0/tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestATSFileParser.java ---------------------------------------------------------------------- diff --git a/tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestATSFileParser.java b/tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestATSFileParser.java new file mode 100644 index 0000000..faff182 --- /dev/null +++ b/tez-plugins/tez-history-parser/src/test/java/org/apache/tez/history/TestATSFileParser.java @@ -0,0 +1,545 @@ +/** + * 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.history; + +import com.google.common.collect.Sets; +import org.apache.commons.io.FileUtils; +import org.apache.commons.lang.RandomStringUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hdfs.DFSConfigKeys; +import org.apache.hadoop.hdfs.MiniDFSCluster; +import org.apache.hadoop.hdfs.server.namenode.EditLogFileOutputStream; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; +import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; +import org.apache.hadoop.yarn.conf.YarnConfiguration; +import org.apache.tez.client.TezClient; +import org.apache.tez.common.counters.DAGCounter; +import org.apache.tez.common.counters.TaskCounter; +import org.apache.tez.common.counters.TezCounter; +import org.apache.tez.dag.api.DAG; +import org.apache.tez.dag.api.DataSinkDescriptor; +import org.apache.tez.dag.api.DataSourceDescriptor; +import org.apache.tez.dag.api.Edge; +import org.apache.tez.dag.api.EdgeProperty; +import org.apache.tez.dag.api.ProcessorDescriptor; +import org.apache.tez.dag.api.TezConfiguration; +import org.apache.tez.dag.api.TezException; +import org.apache.tez.dag.api.Vertex; +import org.apache.tez.dag.api.client.DAGClient; +import org.apache.tez.dag.api.client.StatusGetOpts; +import org.apache.tez.dag.api.event.VertexState; +import org.apache.tez.dag.api.oldrecords.TaskAttemptState; +import org.apache.tez.dag.api.oldrecords.TaskState; +import org.apache.tez.dag.app.dag.DAGState; +import org.apache.tez.dag.history.logging.ats.ATSHistoryLoggingService; +import org.apache.tez.dag.records.TezDAGID; +import org.apache.tez.examples.WordCount; +import org.apache.tez.history.parser.ATSFileParser; +import org.apache.tez.history.parser.datamodel.DagInfo; +import org.apache.tez.history.parser.datamodel.EdgeInfo; +import org.apache.tez.history.parser.datamodel.TaskAttemptInfo; +import org.apache.tez.history.parser.datamodel.TaskInfo; +import org.apache.tez.history.parser.datamodel.VersionInfo; +import org.apache.tez.history.parser.datamodel.VertexInfo; +import org.apache.tez.mapreduce.input.MRInput; +import org.apache.tez.mapreduce.output.MROutput; +import org.apache.tez.mapreduce.processor.SimpleMRProcessor; +import org.apache.tez.runtime.api.ProcessorContext; +import org.apache.tez.runtime.library.api.TezRuntimeConfiguration; +import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig; +import org.apache.tez.runtime.library.input.OrderedGroupedKVInput; +import org.apache.tez.runtime.library.output.OrderedPartitionedKVOutput; +import org.apache.tez.runtime.library.partitioner.HashPartitioner; +import org.apache.tez.tests.MiniTezClusterWithTimeline; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +import java.io.BufferedWriter; +import java.io.File; +import java.io.IOException; +import java.io.OutputStreamWriter; +import java.util.Map; + +import static org.junit.Assert.assertTrue; + +public class TestATSFileParser { + + private static MiniDFSCluster miniDFSCluster; + private static MiniTezClusterWithTimeline miniTezCluster; + + //location within miniHDFS cluster's hdfs + private static Path inputLoc = new Path("/tmp/sample.txt"); + + private final static String INPUT = "Input"; + private final static String OUTPUT = "Output"; + private final static String TOKENIZER = "Tokenizer"; + private final static String SUMMATION = "Summation"; + + private static Configuration conf = new Configuration(); + private static FileSystem fs; + private static String TEST_ROOT_DIR = + "target" + Path.SEPARATOR + TestATSFileParser.class.getName() + "-tmpDir"; + private static String TEZ_BASE_DIR = + "target" + Path.SEPARATOR + TestATSFileParser.class.getName() + "-tez"; + private static String DOWNLOAD_DIR = TEST_ROOT_DIR + Path.SEPARATOR + "download"; + + private static String timelineAddress; + private static TezClient tezClient; + + private static int dagNumber; + + @BeforeClass + public static void setupCluster() throws Exception { + conf = new Configuration(); + conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_EDITS_NOEDITLOGCHANNELFLUSH, false); + EditLogFileOutputStream.setShouldSkipFsyncForTesting(true); + conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR); + miniDFSCluster = + new MiniDFSCluster.Builder(conf).numDataNodes(1).format(true).build(); + fs = miniDFSCluster.getFileSystem(); + conf.set("fs.defaultFS", fs.getUri().toString()); + + setupTezCluster(); + } + + @AfterClass + public static void shutdownCluster() { + try { + if (tezClient != null) { + try { + tezClient.stop(); + } catch (TezException e) { + //ignore + } catch (IOException e) { + //ignore + } + } + if (miniDFSCluster != null) { + miniDFSCluster.shutdown(); + } + if (miniTezCluster != null) { + miniTezCluster.stop(); + } + } finally { + try { + FileUtils.deleteDirectory(new File(TEST_ROOT_DIR)); + FileUtils.deleteDirectory(new File(TEZ_BASE_DIR)); + } catch (IOException e) { + //safe to ignore + } + } + } + + // @Before + public static void setupTezCluster() throws Exception { + conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT, 3 * 1000); + conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT, 3 * 1000); + conf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT, 2); + + //Enable per edge counters + conf.setBoolean(TezConfiguration.TEZ_TASK_GENERATE_COUNTERS_PER_IO, true); + conf.setBoolean(TezConfiguration.TEZ_AM_ALLOW_DISABLED_TIMELINE_DOMAINS, true); + conf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true); + conf.set(TezConfiguration.TEZ_HISTORY_LOGGING_SERVICE_CLASS, ATSHistoryLoggingService + .class.getName()); + + miniTezCluster = + new MiniTezClusterWithTimeline(TEZ_BASE_DIR, 1, 1, 1, true); + + miniTezCluster.init(conf); + miniTezCluster.start(); + + createSampleFile(inputLoc); + + TezConfiguration tezConf = new TezConfiguration(miniTezCluster.getConfig()); + tezConf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true); + tezConf.set(YarnConfiguration.TIMELINE_SERVICE_WEBAPP_ADDRESS, "0.0.0.0:8188"); + tezConf.setBoolean(TezConfiguration.TEZ_AM_ALLOW_DISABLED_TIMELINE_DOMAINS, true); + tezConf.set(TezConfiguration.TEZ_HISTORY_LOGGING_SERVICE_CLASS, + ATSHistoryLoggingService.class.getName()); + + tezClient = TezClient.create("WordCount", tezConf, true); + tezClient.start(); + tezClient.waitTillReady(); + } + + + /** + * Run a word count example in mini cluster and check if it is possible to download + * data from ATS and parse it. + * + * @throws Exception + */ + @Test + public void testParserWithSuccessfulJob() throws Exception { + //Run basic word count example. + String dagId = runWordCount(WordCount.TokenProcessor.class.getName(), + WordCount.SumProcessor.class.getName(), "WordCount"); + + //Export the data from ATS + String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR }; + + int result = ATSImportTool.process(args); + assertTrue(result == 0); + + //Parse ATS data + DagInfo dagInfo = getDagInfo(dagId); + + //Verify DAGInfo. Verifies vertex, task, taskAttempts in recursive manner + verifyDagInfo(dagInfo); + + //Job specific + assertTrue(dagInfo.getNumVertices() == 2); + assertTrue(dagInfo.getName().equals("WordCount")); + assertTrue(dagInfo.getVertex(TOKENIZER).getProcessorClassName().equals( + WordCount.TokenProcessor.class.getName())); + assertTrue(dagInfo.getVertex(SUMMATION).getProcessorClassName() + .equals(WordCount.SumProcessor.class.getName())); + assertTrue(dagInfo.getEdges().size() == 1); + EdgeInfo edgeInfo = dagInfo.getEdges().iterator().next(); + assertTrue(edgeInfo.getDataMovementType(). + equals(EdgeProperty.DataMovementType.SCATTER_GATHER.toString())); + assertTrue(edgeInfo.getSourceVertex().getVertexName().equals(TOKENIZER)); + assertTrue(edgeInfo.getDestinationVertex().getVertexName().equals(SUMMATION)); + assertTrue(edgeInfo.getInputVertexName().equals(TOKENIZER)); + assertTrue(edgeInfo.getOutputVertexName().equals(SUMMATION)); + assertTrue(edgeInfo.getEdgeSourceClass().equals(OrderedPartitionedKVOutput.class.getName())); + assertTrue(edgeInfo.getEdgeDestinationClass().equals(OrderedGroupedKVInput.class.getName())); + assertTrue(dagInfo.getVertices().size() == 2); + for (VertexInfo vertexInfo : dagInfo.getVertices()) { + assertTrue(vertexInfo.getKilledTasksCount() == 0); + for (TaskInfo taskInfo : vertexInfo.getTasks()) { + assertTrue(taskInfo.getNumberOfTaskAttempts() == 1); + assertTrue(taskInfo.getMaxTaskAttemptDuration() >= 0); + assertTrue(taskInfo.getMinTaskAttemptDuration() >= 0); + assertTrue(taskInfo.getAvgTaskAttemptDuration() >= 0); + assertTrue(taskInfo.getLastTaskAttemptToFinish() != null); + assertTrue(taskInfo.getContainersMapping().size() > 0); + assertTrue(taskInfo.getSuccessfulTaskAttempts().size() > 0); + assertTrue(taskInfo.getFailedTaskAttempts().size() == 0); + assertTrue(taskInfo.getKilledTaskAttempts().size() == 0); + } + assertTrue(vertexInfo.getLastTaskToFinish() != null); + if (vertexInfo.getVertexName().equals(TOKENIZER)) { + assertTrue(vertexInfo.getInputEdges().size() == 0); + assertTrue(vertexInfo.getOutputEdges().size() == 1); + assertTrue(vertexInfo.getOutputVertices().size() == 1); + assertTrue(vertexInfo.getInputVertices().size() == 0); + } else { + assertTrue(vertexInfo.getInputEdges().size() == 1); + assertTrue(vertexInfo.getOutputEdges().size() == 0); + assertTrue(vertexInfo.getOutputVertices().size() == 0); + assertTrue(vertexInfo.getInputVertices().size() == 1); + } + } + } + + /** + * Run a word count example in mini cluster. + * Provide invalid URL for ATS. + * + * @throws Exception + */ + @Test + public void testParserWithSuccessfulJob_InvalidATS() throws Exception { + //Run basic word count example. + String dagId = runWordCount(WordCount.TokenProcessor.class.getName(), + WordCount.SumProcessor.class.getName(), "WordCount-With-WrongATS-URL"); + + //Export the data from ATS + String atsAddress = "--atsAddress=http://atsHost:8188";; + String[] args = { "--dagId=" + dagId, + "--downloadDir=" + DOWNLOAD_DIR, + atsAddress + }; + + int result = ATSImportTool.process(args); + assertTrue(result == -1); + } + + /** + * Run a failed job and parse the data from ATS + */ + @Test + public void testParserWithFailedJob() throws Exception { + //Run a job which would fail + String dagId = runWordCount(WordCount.TokenProcessor.class.getName(), FailProcessor.class + .getName(), "WordCount-With-Exception"); + + //Export the data from ATS + String[] args = { "--dagId=" + dagId, "--downloadDir=" + DOWNLOAD_DIR }; + + int result = ATSImportTool.process(args); + assertTrue(result == 0); + + //Parse ATS data + DagInfo dagInfo = getDagInfo(dagId); + + //Verify DAGInfo. Verifies vertex, task, taskAttempts in recursive manner + verifyDagInfo(dagInfo); + + //Dag specific + VertexInfo summationVertex = dagInfo.getVertex(SUMMATION); + assertTrue(summationVertex.getFailedTasks().size() == 1); //1 task, 4 attempts failed + assertTrue(summationVertex.getFailedTasks().get(0).getFailedTaskAttempts().size() == 4); + assertTrue(summationVertex.getStatus().equals(VertexState.FAILED.toString())); + + assertTrue(dagInfo.getFailedVertices().size() == 1); + assertTrue(dagInfo.getFailedVertices().get(0).getVertexName().equals(SUMMATION)); + assertTrue(dagInfo.getSuccessfullVertices().size() == 1); + assertTrue(dagInfo.getSuccessfullVertices().get(0).getVertexName().equals(TOKENIZER)); + + assertTrue(dagInfo.getStatus().equals(DAGState.FAILED.toString())); + + verifyCounter(dagInfo.getCounter(DAGCounter.NUM_FAILED_TASKS.toString()), null, 4); + verifyCounter(dagInfo.getCounter(DAGCounter.NUM_SUCCEEDED_TASKS.toString()), null, 1); + verifyCounter(dagInfo.getCounter(DAGCounter.TOTAL_LAUNCHED_TASKS.toString()), null, 5); + + verifyCounter(dagInfo.getCounter(TaskCounter.INPUT_RECORDS_PROCESSED.toString()), + "TaskCounter_Tokenizer_INPUT_Input", 10); + verifyCounter(dagInfo.getCounter(TaskCounter.ADDITIONAL_SPILLS_BYTES_READ.toString()), + "TaskCounter_Tokenizer_OUTPUT_Summation", 0); + verifyCounter(dagInfo.getCounter(TaskCounter.OUTPUT_RECORDS.toString()), + "TaskCounter_Tokenizer_OUTPUT_Summation", + 20); //Every line has 2 words. 10 lines x 2 words = 20 + verifyCounter(dagInfo.getCounter(TaskCounter.SPILLED_RECORDS.toString()), + "TaskCounter_Tokenizer_OUTPUT_Summation", 20); //Same as above + + //TODO: Need to check for SUMMATION vertex counters. Since all attempts are failed, counters are not getting populated. + //TaskCounter.REDUCE_INPUT_RECORDS + + //Verify if the processor exception is given in diagnostics + assertTrue(dagInfo.getDiagnostics().contains("Failing this processor for some reason")); + + } + + /** + * Create sample file for wordcount program + * + * @param inputLoc + * @throws IOException + */ + private static void createSampleFile(Path inputLoc) throws IOException { + fs.deleteOnExit(inputLoc); + FSDataOutputStream out = fs.create(inputLoc); + BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out)); + for (int i = 0; i < 10; i++) { + writer.write("Sample " + RandomStringUtils.randomAlphanumeric(5)); + writer.newLine(); + } + writer.close(); + } + + private DagInfo getDagInfo(String dagId) throws TezException { + //Parse downloaded contents + File downloadedFile = new File(DOWNLOAD_DIR + + Path.SEPARATOR + dagId + + Path.SEPARATOR + dagId + ".zip"); + ATSFileParser parser = new ATSFileParser(downloadedFile); + DagInfo dagInfo = parser.getDAGData(dagId); + assertTrue(dagInfo.getDagId().equals(dagId)); + return dagInfo; + } + + private void verifyCounter(Map<String, TezCounter> counterMap, + String counterGroupName, long expectedVal) { + //Iterate through group-->tezCounter + for (Map.Entry<String, TezCounter> entry : counterMap.entrySet()) { + if (counterGroupName != null) { + if (entry.getKey().equals(counterGroupName)) { + assertTrue(entry.getValue().getValue() == expectedVal); + } + } else { + assertTrue(entry.getValue().getValue() == expectedVal); + } + } + } + + private String runWordCount(String tokenizerProcessor, String summationProcessor, + String dagName) + throws Exception { + dagNumber++; + + //HDFS path + Path outputLoc = new Path("/tmp/outPath_" + System.currentTimeMillis()); + + DataSourceDescriptor dataSource = MRInput.createConfigBuilder(conf, + TextInputFormat.class, inputLoc.toString()).build(); + + DataSinkDescriptor dataSink = + MROutput.createConfigBuilder(conf, TextOutputFormat.class, outputLoc.toString()).build(); + + Vertex tokenizerVertex = Vertex.create(TOKENIZER, ProcessorDescriptor.create( + tokenizerProcessor)).addDataSource(INPUT, dataSource); + + OrderedPartitionedKVEdgeConfig edgeConf = OrderedPartitionedKVEdgeConfig + .newBuilder(Text.class.getName(), IntWritable.class.getName(), + HashPartitioner.class.getName()).build(); + + Vertex summationVertex = Vertex.create(SUMMATION, + ProcessorDescriptor.create(summationProcessor), 1).addDataSink(OUTPUT, dataSink); + + // Create DAG and add the vertices. Connect the producer and consumer vertices via the edge + DAG dag = DAG.create(dagName); + dag.addVertex(tokenizerVertex).addVertex(summationVertex).addEdge( + Edge.create(tokenizerVertex, summationVertex, edgeConf.createDefaultEdgeProperty())); + + DAGClient client = tezClient.submitDAG(dag); + client.waitForCompletionWithStatusUpdates(Sets.newHashSet(StatusGetOpts.GET_COUNTERS)); + TezDAGID tezDAGID = TezDAGID.getInstance(tezClient.getAppMasterApplicationId(), dagNumber); + + return tezDAGID.toString(); + } + + /** + * Processor which would just throw exception. + */ + public static class FailProcessor extends SimpleMRProcessor { + public FailProcessor(ProcessorContext context) { + super(context); + } + + @Override + public void run() throws Exception { + throw new Exception("Failing this processor for some reason"); + } + } + + private void verifyDagInfo(DagInfo dagInfo) { + VersionInfo versionInfo = dagInfo.getVersionInfo(); + assertTrue(versionInfo != null); //should be present post 0.5.4 + assertTrue(versionInfo.getVersion() != null); + assertTrue(versionInfo.getRevision() != null); + assertTrue(versionInfo.getBuildTime() != null); + + assertTrue(dagInfo.getAbsStartTime() > 0); + assertTrue(dagInfo.getFinishTime() > 0); + assertTrue(dagInfo.getStartTime() == 0); + assertTrue(dagInfo.getAbsStartTime() > 0); + if (dagInfo.getStatus().equalsIgnoreCase(DAGState.SUCCEEDED.toString())) { + assertTrue(dagInfo.getAbsFinishTime() >= dagInfo.getAbsStartTime()); + } + assertTrue(dagInfo.getFinishTime() > dagInfo.getStartTime()); + + assertTrue(dagInfo.getAbsStartTime() > dagInfo.getAbsoluteSubmitTime()); + assertTrue(dagInfo.getTimeTaken() > 0); + + //Verify all vertices + for (VertexInfo vertexInfo : dagInfo.getVertices()) { + verifyVertex(vertexInfo, vertexInfo.getFailedTasksCount() > 0); + } + + VertexInfo fastestVertex = dagInfo.getFastestVertex(); + assertTrue(fastestVertex != null); + + if (dagInfo.getStatus().equals(DAGState.SUCCEEDED)) { + assertTrue(dagInfo.getSlowestVertex() != null); + } + } + + private void verifyVertex(VertexInfo vertexInfo, boolean hasFailedTasks) { + assertTrue(vertexInfo != null); + if (hasFailedTasks) { + assertTrue(vertexInfo.getFailedTasksCount() > 0); + } + assertTrue(vertexInfo.getStartTime() > 0); + assertTrue(vertexInfo.getAbsStartTime() > 0); + assertTrue(vertexInfo.getFinishTime() > 0); + assertTrue(vertexInfo.getStartTime() < vertexInfo.getFinishTime()); + assertTrue(vertexInfo.getVertexName() != null); + if (!hasFailedTasks) { + assertTrue(vertexInfo.getAbsFinishTime() > 0); + assertTrue(vertexInfo.getFailedTasks().size() == 0); + assertTrue(vertexInfo.getSucceededTasksCount() == vertexInfo.getSuccessfulTasks().size()); + assertTrue(vertexInfo.getFailedTasksCount() == 0); + assertTrue(vertexInfo.getAvgTaskDuration() > 0); + assertTrue(vertexInfo.getMaxTaskDuration() > 0); + assertTrue(vertexInfo.getMinTaskDuration() > 0); + assertTrue(vertexInfo.getTimeTaken() > 0); + assertTrue(vertexInfo.getStatus().equalsIgnoreCase(VertexState.SUCCEEDED.toString())); + assertTrue(vertexInfo.getCompletedTasksCount() > 0); + assertTrue(vertexInfo.getFirstTaskToStart() != null); + assertTrue(vertexInfo.getSucceededTasksCount() > 0); + assertTrue(vertexInfo.getTasks().size() > 0); + } + + for (TaskInfo taskInfo : vertexInfo.getTasks()) { + if (taskInfo.getStatus().equals(TaskState.SUCCEEDED.toString())) { + verifyTask(taskInfo, false); + } + } + + for (TaskInfo taskInfo : vertexInfo.getFailedTasks()) { + verifyTask(taskInfo, true); + } + + assertTrue(vertexInfo.getProcessorClassName() != null); + assertTrue(vertexInfo.getStatus() != null); + assertTrue(vertexInfo.getDagInfo() != null); + assertTrue(vertexInfo.getInitTime() > 0); + assertTrue(vertexInfo.getNumTasks() > 0); + } + + private void verifyTask(TaskInfo taskInfo, boolean hasFailedAttempts) { + assertTrue(taskInfo != null); + assertTrue(taskInfo.getStatus() != null); + assertTrue(taskInfo.getStartTime() > 0); + + //Not testing for killed attempts. So if there are no failures, it should succeed + if (!hasFailedAttempts) { + assertTrue(taskInfo.getStatus().equals(TaskState.SUCCEEDED.toString())); + assertTrue(taskInfo.getFinishTime() > 0 && taskInfo.getAbsFinishTime() > taskInfo + .getFinishTime()); + assertTrue( + taskInfo.getStartTime() > 0 && taskInfo.getAbsStartTime() > taskInfo.getStartTime()); + assertTrue(taskInfo.getSuccessfulAttemptId() != null); + assertTrue(taskInfo.getSuccessfulTaskAttempt() != null); + } + assertTrue(taskInfo.getTaskId() != null); + + for (TaskAttemptInfo attemptInfo : taskInfo.getTaskAttempts()) { + verifyTaskAttemptInfo(attemptInfo); + } + } + + private void verifyTaskAttemptInfo(TaskAttemptInfo attemptInfo) { + if (attemptInfo.getStatus() != null && attemptInfo.getStatus() + .equals(TaskAttemptState.SUCCEEDED)) { + assertTrue(attemptInfo.getStartTime() > 0); + assertTrue(attemptInfo.getFinishTime() > 0); + assertTrue(attemptInfo.getAbsStartTime() > 0); + assertTrue(attemptInfo.getAbsFinishTime() > 0); + assertTrue(attemptInfo.getAbsFinishTime() > attemptInfo.getAbsStartTime()); + assertTrue(attemptInfo.getAbsFinishTime() > attemptInfo.getFinishTime()); + assertTrue(attemptInfo.getAbsStartTime() > attemptInfo.getStartTime()); + assertTrue(attemptInfo.getNodeId() != null); + assertTrue(attemptInfo.getTimeTaken() != -1); + assertTrue(attemptInfo.getEvents() != null); + assertTrue(attemptInfo.getTezCounters() != null); + assertTrue(attemptInfo.getContainer() != null); + } + assertTrue(attemptInfo.getTaskInfo() != null); + } +}
