| /** |
| * 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.ambari.log4j.hadoop.mapreduce.jobhistory; |
| |
| import java.io.IOException; |
| import java.sql.Connection; |
| import java.sql.PreparedStatement; |
| import java.sql.ResultSet; |
| import java.sql.SQLException; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Set; |
| import java.util.TreeMap; |
| import java.util.TreeSet; |
| import java.util.regex.Matcher; |
| import java.util.regex.Pattern; |
| |
| import org.apache.ambari.eventdb.model.WorkflowContext; |
| import org.apache.ambari.eventdb.model.WorkflowDag; |
| import org.apache.ambari.eventdb.model.WorkflowDag.WorkflowDagEntry; |
| import org.apache.ambari.log4j.common.LogStoreUpdateProvider; |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.mapreduce.Counter; |
| import org.apache.hadoop.mapreduce.CounterGroup; |
| import org.apache.hadoop.mapreduce.Counters; |
| import org.apache.hadoop.mapreduce.TaskType; |
| import org.apache.hadoop.tools.rumen.HistoryEvent; |
| import org.apache.hadoop.tools.rumen.JhCounter; |
| import org.apache.hadoop.tools.rumen.JhCounterGroup; |
| import org.apache.hadoop.tools.rumen.JhCounters; |
| import org.apache.hadoop.tools.rumen.JobFinishedEvent; |
| import org.apache.hadoop.tools.rumen.JobInfoChangeEvent; |
| import org.apache.hadoop.tools.rumen.JobInitedEvent; |
| import org.apache.hadoop.tools.rumen.JobStatusChangedEvent; |
| import org.apache.hadoop.tools.rumen.JobSubmittedEvent; |
| import org.apache.hadoop.tools.rumen.JobUnsuccessfulCompletionEvent; |
| import org.apache.hadoop.tools.rumen.MapAttemptFinishedEvent; |
| import org.apache.hadoop.tools.rumen.ReduceAttemptFinishedEvent; |
| import org.apache.hadoop.tools.rumen.TaskAttemptFinishedEvent; |
| import org.apache.hadoop.tools.rumen.TaskAttemptStartedEvent; |
| import org.apache.hadoop.tools.rumen.TaskAttemptUnsuccessfulCompletionEvent; |
| import org.apache.hadoop.tools.rumen.TaskFailedEvent; |
| import org.apache.hadoop.tools.rumen.TaskFinishedEvent; |
| import org.apache.hadoop.tools.rumen.TaskStartedEvent; |
| import org.apache.hadoop.util.StringUtils; |
| import org.apache.log4j.spi.LoggingEvent; |
| import org.codehaus.jackson.map.ObjectMapper; |
| |
| public class MapReduceJobHistoryUpdater implements LogStoreUpdateProvider { |
| |
| private static final Log LOG = |
| LogFactory.getLog(MapReduceJobHistoryUpdater.class); |
| |
| private Connection connection; |
| |
| private static final String WORKFLOW_TABLE = "workflow"; |
| private static final String JOB_TABLE = "job"; |
| private static final String TASK_TABLE = "task"; |
| private static final String TASKATTEMPT_TABLE = "taskAttempt"; |
| |
| private PreparedStatement workflowPS = null; |
| private PreparedStatement workflowSelectPS = null; |
| private PreparedStatement workflowUpdateTimePS = null; |
| private PreparedStatement workflowUpdateNumCompletedPS = null; |
| |
| private Map<Class<? extends HistoryEvent>, PreparedStatement> entitySqlMap = |
| new HashMap<Class<? extends HistoryEvent>, PreparedStatement>(); |
| |
| @Override |
| public void init(Connection connection) throws IOException { |
| this.connection = connection; |
| |
| try { |
| initializePreparedStatements(); |
| } catch (SQLException sqle) { |
| throw new IOException(sqle); |
| } |
| } |
| |
| private void initializePreparedStatements() throws SQLException { |
| initializeJobPreparedStatements(); |
| initializeTaskPreparedStatements(); |
| initializeTaskAttemptPreparedStatements(); |
| } |
| |
| private PreparedStatement jobEndUpdate; |
| |
| private void initializeJobPreparedStatements() throws SQLException { |
| |
| /** |
| * Job events |
| */ |
| |
| // JobSubmittedEvent |
| |
| PreparedStatement jobSubmittedPrepStmnt = |
| connection.prepareStatement( |
| "INSERT INTO " + |
| JOB_TABLE + |
| " (" + |
| "jobId, " + |
| "jobName, " + |
| "userName, " + |
| "confPath, " + |
| "queue, " + |
| "submitTime, " + |
| "workflowId, " + |
| "workflowEntityName " + |
| ") " + |
| "VALUES" + |
| " (?, ?, ?, ?, ?, ?, ?, ?)" |
| ); |
| entitySqlMap.put(JobSubmittedEvent.class, jobSubmittedPrepStmnt); |
| |
| workflowSelectPS = |
| connection.prepareStatement( |
| "SELECT workflowContext FROM " + WORKFLOW_TABLE + " where workflowId = ?" |
| ); |
| |
| workflowPS = |
| connection.prepareStatement( |
| "INSERT INTO " + |
| WORKFLOW_TABLE + |
| " (" + |
| "workflowId, " + |
| "workflowName, " + |
| "workflowContext, " + |
| "userName, " + |
| "startTime, " + |
| "lastUpdateTime, " + |
| "duration, " + |
| "numJobsTotal, " + |
| "numJobsCompleted" + |
| ") " + |
| "VALUES" + |
| " (?, ?, ?, ?, ?, ?, 0, ?, 0)" |
| ); |
| |
| workflowUpdateTimePS = |
| connection.prepareStatement( |
| "UPDATE " + |
| WORKFLOW_TABLE + |
| " SET " + |
| "workflowContext = ?, " + |
| "numJobsTotal = ?, " + |
| "lastUpdateTime = ?, " + |
| "duration = ? - (SELECT startTime FROM " + |
| WORKFLOW_TABLE + |
| " WHERE workflowId = ?) " + |
| "WHERE workflowId = ?" |
| ); |
| |
| workflowUpdateNumCompletedPS = |
| connection.prepareStatement( |
| "UPDATE " + |
| WORKFLOW_TABLE + |
| " SET " + |
| "lastUpdateTime = ?, " + |
| "duration = ? - (SELECT startTime FROM " + |
| WORKFLOW_TABLE + |
| " WHERE workflowId = selectid), " + |
| "numJobsCompleted = rows, " + |
| "inputBytes = input, " + |
| "outputBytes = output " + |
| "FROM (SELECT count(*) as rows, sum(inputBytes) as input, " + |
| "sum(outputBytes) as output, workflowId as selectid FROM " + |
| JOB_TABLE + |
| " WHERE workflowId = (SELECT workflowId FROM " + |
| JOB_TABLE + |
| " WHERE jobId = ?) AND status = 'SUCCESS' " + |
| "GROUP BY workflowId) as jobsummary " + |
| "WHERE workflowId = selectid" |
| ); |
| |
| // JobFinishedEvent |
| |
| PreparedStatement jobFinishedPrepStmnt = |
| connection.prepareStatement( |
| "UPDATE " + |
| JOB_TABLE + |
| " SET " + |
| "finishTime = ?, " + |
| "finishedMaps = ?, " + |
| "finishedReduces= ?, " + |
| "failedMaps = ?, " + |
| "failedReduces = ?, " + |
| "inputBytes = ?, " + |
| "outputBytes = ? " + |
| "WHERE " + |
| "jobId = ?" |
| ); |
| entitySqlMap.put(JobFinishedEvent.class, jobFinishedPrepStmnt); |
| |
| // JobInitedEvent |
| |
| PreparedStatement jobInitedPrepStmnt = |
| connection.prepareStatement( |
| "UPDATE " + |
| JOB_TABLE + |
| " SET " + |
| "launchTime = ?, " + |
| "maps = ?, " + |
| "reduces = ?, " + |
| "status = ? "+ |
| "WHERE " + |
| "jobId = ?" |
| ); |
| entitySqlMap.put(JobInitedEvent.class, jobInitedPrepStmnt); |
| |
| // JobStatusChangedEvent |
| |
| PreparedStatement jobStatusChangedPrepStmnt = |
| connection.prepareStatement( |
| "UPDATE " + |
| JOB_TABLE + |
| " SET " + |
| "status = ? "+ |
| "WHERE " + |
| "jobId = ?" |
| ); |
| entitySqlMap.put(JobStatusChangedEvent.class, jobStatusChangedPrepStmnt); |
| |
| // JobInfoChangedEvent |
| |
| PreparedStatement jobInfoChangedPrepStmnt = |
| connection.prepareStatement( |
| "UPDATE " + |
| JOB_TABLE + |
| " SET " + |
| "submitTime = ?, " + |
| "launchTime = ? " + |
| "WHERE " + |
| "jobId = ?" |
| ); |
| entitySqlMap.put(JobInfoChangeEvent.class, jobInfoChangedPrepStmnt); |
| |
| // JobUnsuccessfulCompletionEvent |
| PreparedStatement jobUnsuccessfulPrepStmnt = |
| connection.prepareStatement( |
| "UPDATE " + |
| JOB_TABLE + |
| " SET " + |
| "finishTime = ?, " + |
| "finishedMaps = ?, " + |
| "finishedReduces = ?, " + |
| "status = ? " + |
| "WHERE " + |
| "jobId = ?" |
| ); |
| entitySqlMap.put( |
| JobUnsuccessfulCompletionEvent.class, jobUnsuccessfulPrepStmnt); |
| |
| // Job update at the end |
| jobEndUpdate = |
| connection.prepareStatement( |
| "UPDATE " + |
| JOB_TABLE + |
| " SET " + |
| " mapsRuntime = (" + |
| "SELECT " + |
| "SUM(" + |
| TASKATTEMPT_TABLE + ".finishTime" + " - " + |
| TASKATTEMPT_TABLE + ".startTime" + |
| ")" + |
| " FROM " + |
| TASKATTEMPT_TABLE + |
| " WHERE " + |
| TASKATTEMPT_TABLE + ".jobId = " + JOB_TABLE + ".jobId " + |
| " AND " + |
| TASKATTEMPT_TABLE + ".taskType = ?)" + |
| ", " + |
| " reducesRuntime = (" + |
| "SELECT SUM(" + |
| TASKATTEMPT_TABLE + ".finishTime" + " - " + |
| TASKATTEMPT_TABLE + ".startTime" + |
| ")" + |
| " FROM " + |
| TASKATTEMPT_TABLE + |
| " WHERE " + |
| TASKATTEMPT_TABLE + ".jobId = " + JOB_TABLE + ".jobId " + |
| " AND " + |
| TASKATTEMPT_TABLE + ".taskType = ?) " + |
| " WHERE " + |
| "jobId = ?" |
| ); |
| } |
| |
| private void initializeTaskPreparedStatements() throws SQLException { |
| |
| /** |
| * Task events |
| */ |
| |
| // TaskStartedEvent |
| |
| PreparedStatement taskStartedPrepStmnt = |
| connection.prepareStatement( |
| "INSERT INTO " + |
| TASK_TABLE + |
| " (" + |
| "jobId, " + |
| "taskType, " + |
| "splits, " + |
| "startTime, " + |
| "taskId" + |
| ") " + |
| "VALUES (?, ?, ?, ?, ?)" |
| ); |
| entitySqlMap.put(TaskStartedEvent.class, taskStartedPrepStmnt); |
| |
| // TaskFinishedEvent |
| |
| PreparedStatement taskFinishedPrepStmnt = |
| connection.prepareStatement( |
| "UPDATE " + |
| TASK_TABLE + |
| " SET " + |
| "jobId = ?, " + |
| "taskType = ?, " + |
| "status = ?, " + |
| "finishTime = ? " + |
| " WHERE " + |
| "taskId = ?" |
| ); |
| entitySqlMap.put(TaskFinishedEvent.class, taskFinishedPrepStmnt); |
| |
| // TaskFailedEvent |
| |
| PreparedStatement taskFailedPrepStmnt = |
| connection.prepareStatement( |
| "UPDATE " + |
| TASK_TABLE + |
| " SET " + |
| "jobId = ?, " + |
| "taskType = ?, " + |
| "status = ?, " + |
| "finishTime = ?, " + |
| "error = ?, " + |
| "failedAttempt = ? " + |
| "WHERE " + |
| "taskId = ?" |
| ); |
| entitySqlMap.put(TaskFailedEvent.class, taskFailedPrepStmnt); |
| } |
| |
| private void initializeTaskAttemptPreparedStatements() throws SQLException { |
| |
| /** |
| * TaskAttempt events |
| */ |
| |
| // TaskAttemptStartedEvent |
| |
| PreparedStatement taskAttemptStartedPrepStmnt = |
| connection.prepareStatement( |
| "INSERT INTO " + |
| TASKATTEMPT_TABLE + |
| " (" + |
| "jobId, " + |
| "taskId, " + |
| "taskType, " + |
| "startTime, " + |
| "taskTracker, " + |
| "locality, " + |
| "avataar, " + |
| "taskAttemptId" + |
| ") " + |
| "VALUES (?, ?, ?, ?, ?, ?, ?, ?)" |
| ); |
| entitySqlMap.put( |
| TaskAttemptStartedEvent.class, taskAttemptStartedPrepStmnt); |
| |
| // TaskAttemptFinishedEvent |
| |
| PreparedStatement taskAttemptFinishedPrepStmnt = |
| connection.prepareStatement( |
| "UPDATE " + |
| TASKATTEMPT_TABLE + |
| " SET " + |
| "jobId = ?, " + |
| "taskId = ?, " + |
| "taskType = ?, " + |
| "finishTime = ?, " + |
| "status = ?, " + |
| "taskTracker = ? " + |
| " WHERE " + |
| "taskAttemptId = ?" |
| ); |
| entitySqlMap.put( |
| TaskAttemptFinishedEvent.class, taskAttemptFinishedPrepStmnt); |
| |
| // TaskAttemptUnsuccessfulEvent |
| |
| PreparedStatement taskAttemptUnsuccessfulPrepStmnt = |
| connection.prepareStatement( |
| "UPDATE " + |
| TASKATTEMPT_TABLE + |
| " SET " + |
| "jobId = ?, " + |
| "taskId = ?, " + |
| "taskType = ?, " + |
| "finishTime = ?, " + |
| "status = ?, " + |
| "taskTracker = ?, " + |
| "error = ? " + |
| " WHERE " + |
| "taskAttemptId = ?" |
| ); |
| entitySqlMap.put( |
| TaskAttemptUnsuccessfulCompletionEvent.class, |
| taskAttemptUnsuccessfulPrepStmnt); |
| |
| // MapAttemptFinishedEvent |
| |
| PreparedStatement mapAttemptFinishedPrepStmnt = |
| connection.prepareStatement( |
| "UPDATE " + |
| TASKATTEMPT_TABLE + |
| " SET " + |
| "jobId = ?, " + |
| "taskId = ?, " + |
| "taskType = ?, " + |
| "mapFinishTime = ?, " + |
| "finishTime = ?, " + |
| "inputBytes = ?, " + |
| "outputBytes = ?, " + |
| "status = ?, " + |
| "taskTracker = ? " + |
| " WHERE " + |
| "taskAttemptId = ?" |
| ); |
| entitySqlMap.put( |
| MapAttemptFinishedEvent.class, mapAttemptFinishedPrepStmnt); |
| |
| // ReduceAttemptFinishedEvent |
| |
| PreparedStatement reduceAttemptFinishedPrepStmnt = |
| connection.prepareStatement( |
| "UPDATE " + |
| TASKATTEMPT_TABLE + |
| " SET " + |
| "jobId = ?, " + |
| "taskId = ?, " + |
| "taskType = ?, " + |
| "shuffleFinishTime = ?, " + |
| "sortFinishTime = ?, " + |
| "finishTime = ?, " + |
| "inputBytes = ?, " + |
| "outputBytes = ?, " + |
| "status = ?, " + |
| "taskTracker = ? " + |
| " WHERE " + |
| "taskAttemptId = ?" |
| ); |
| entitySqlMap.put( |
| ReduceAttemptFinishedEvent.class, reduceAttemptFinishedPrepStmnt); |
| } |
| |
| private void doUpdates(LoggingEvent originalEvent, |
| Object parsedEvent) throws SQLException { |
| Class<?> eventClass = parsedEvent.getClass(); |
| |
| PreparedStatement entityPS = entitySqlMap.get(eventClass); |
| if (entityPS == null) { |
| LOG.debug("No prepared statement for " + eventClass); |
| return; |
| } |
| |
| if (eventClass == JobSubmittedEvent.class) { |
| processJobSubmittedEvent(entityPS, workflowSelectPS, workflowPS, |
| workflowUpdateTimePS, originalEvent, |
| (JobSubmittedEvent)parsedEvent); |
| } else if (eventClass == JobFinishedEvent.class) { |
| processJobFinishedEvent(entityPS, workflowUpdateNumCompletedPS, |
| originalEvent, (JobFinishedEvent)parsedEvent); |
| } else if (eventClass == JobInitedEvent.class){ |
| processJobInitedEvent(entityPS, |
| originalEvent, (JobInitedEvent)parsedEvent); |
| } else if (eventClass == JobStatusChangedEvent.class) { |
| processJobStatusChangedEvent(entityPS, |
| originalEvent, (JobStatusChangedEvent)parsedEvent); |
| } else if (eventClass == JobInfoChangeEvent.class) { |
| processJobInfoChangeEvent(entityPS, |
| originalEvent, (JobInfoChangeEvent)parsedEvent); |
| } else if (eventClass == JobUnsuccessfulCompletionEvent.class) { |
| processJobUnsuccessfulEvent(entityPS, |
| originalEvent, (JobUnsuccessfulCompletionEvent)parsedEvent); |
| } else if (eventClass == TaskStartedEvent.class) { |
| processTaskStartedEvent(entityPS, |
| originalEvent, (TaskStartedEvent)parsedEvent); |
| } else if (eventClass == TaskFinishedEvent.class) { |
| processTaskFinishedEvent(entityPS, |
| originalEvent, (TaskFinishedEvent)parsedEvent); |
| } else if (eventClass == TaskFailedEvent.class) { |
| processTaskFailedEvent(entityPS, |
| originalEvent, (TaskFailedEvent)parsedEvent); |
| } else if (eventClass == TaskAttemptStartedEvent.class) { |
| processTaskAttemptStartedEvent(entityPS, |
| originalEvent, (TaskAttemptStartedEvent)parsedEvent); |
| } else if (eventClass == TaskAttemptFinishedEvent.class) { |
| processTaskAttemptFinishedEvent(entityPS, |
| originalEvent, (TaskAttemptFinishedEvent)parsedEvent); |
| } else if (eventClass == TaskAttemptUnsuccessfulCompletionEvent.class) { |
| processTaskAttemptUnsuccessfulEvent(entityPS, |
| originalEvent, (TaskAttemptUnsuccessfulCompletionEvent)parsedEvent); |
| } else if (eventClass == MapAttemptFinishedEvent.class) { |
| processMapAttemptFinishedEvent(entityPS, |
| originalEvent, (MapAttemptFinishedEvent)parsedEvent); |
| } else if (eventClass == ReduceAttemptFinishedEvent.class) { |
| processReduceAttemptFinishedEvent(entityPS, |
| originalEvent, (ReduceAttemptFinishedEvent)parsedEvent); |
| } |
| } |
| |
| private void updateJobStatsAtFinish(String jobId) { |
| try { |
| jobEndUpdate.setString(1, "MAP"); |
| jobEndUpdate.setString(2, "REDUCE"); |
| jobEndUpdate.setString(3, jobId); |
| jobEndUpdate.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to update mapsRuntime/reducesRuntime for " + jobId, |
| sqle); |
| } |
| } |
| |
| private static WorkflowContext generateWorkflowContext( |
| JobSubmittedEvent historyEvent) { |
| WorkflowDag wfDag = new WorkflowDag(); |
| WorkflowDagEntry wfDagEntry = new WorkflowDagEntry(); |
| wfDagEntry.setSource("X"); |
| wfDag.addEntry(wfDagEntry); |
| |
| WorkflowContext wc = new WorkflowContext(); |
| wc.setWorkflowId(historyEvent.getJobId().toString().replace("job_", "mr_")); |
| wc.setWorkflowName(historyEvent.getJobName()); |
| wc.setWorkflowEntityName("X"); |
| wc.setWorkflowDag(wfDag); |
| return wc; |
| } |
| |
| // this is based on the regex in org.apache.hadoop.tools.rumen.ParsedLine |
| // except this assumes the format "key"="value" so that both key and value |
| // are quoted and may contain escaped characters |
| private static final Pattern adjPattern = |
| Pattern.compile("\"([^\"\\\\]*+(?:\\\\.[^\"\\\\]*+)*+)\"" + "=" + |
| "\"([^\"\\\\]*+(?:\\\\.[^\"\\\\]*+)*+)\" "); |
| |
| public static WorkflowContext buildWorkflowContext(JobSubmittedEvent historyEvent) { |
| String workflowId = historyEvent.getWorkflowId() |
| .replace("\\", ""); |
| if (workflowId.isEmpty()) |
| return generateWorkflowContext(historyEvent); |
| String workflowName = historyEvent.getWorkflowName() |
| .replace("\\", ""); |
| String workflowNodeName = historyEvent.getWorkflowNodeName() |
| .replace("\\", ""); |
| String workflowAdjacencies = StringUtils.unEscapeString( |
| historyEvent.getWorkflowAdjacencies(), |
| StringUtils.ESCAPE_CHAR, new char[] {'"', '=', '.'}); |
| WorkflowContext context = new WorkflowContext(); |
| context.setWorkflowId(workflowId); |
| context.setWorkflowName(workflowName); |
| context.setWorkflowEntityName(workflowNodeName); |
| WorkflowDag dag = new WorkflowDag(); |
| Matcher matcher = adjPattern.matcher(workflowAdjacencies); |
| |
| while(matcher.find()){ |
| WorkflowDagEntry dagEntry = new WorkflowDagEntry(); |
| dagEntry.setSource(matcher.group(1).replace("\\", "")); |
| String[] values = StringUtils.getStrings( |
| matcher.group(2).replace("\\", "")); |
| if (values != null) { |
| for (String target : values) { |
| dagEntry.addTarget(target); |
| } |
| } |
| dag.addEntry(dagEntry); |
| } |
| if (dag.getEntries().isEmpty()) { |
| WorkflowDagEntry wfDagEntry = new WorkflowDagEntry(); |
| wfDagEntry.setSource(workflowNodeName); |
| dag.addEntry(wfDagEntry); |
| } |
| context.setWorkflowDag(dag); |
| return context; |
| } |
| |
| public static void mergeEntries(Map<String, Set<String>> edges, List<WorkflowDagEntry> entries) { |
| if (entries == null) |
| return; |
| for (WorkflowDagEntry entry : entries) { |
| if (!edges.containsKey(entry.getSource())) |
| edges.put(entry.getSource(), new TreeSet<String>()); |
| Set<String> targets = edges.get(entry.getSource()); |
| targets.addAll(entry.getTargets()); |
| } |
| } |
| |
| public static WorkflowDag constructMergedDag(WorkflowContext workflowContext, WorkflowContext existingWorkflowContext) { |
| Map<String, Set<String>> edges = new TreeMap<String, Set<String>>(); |
| if (existingWorkflowContext.getWorkflowDag() != null) |
| mergeEntries(edges, existingWorkflowContext.getWorkflowDag().getEntries()); |
| if (workflowContext.getWorkflowDag() != null) |
| mergeEntries(edges, workflowContext.getWorkflowDag().getEntries()); |
| WorkflowDag mergedDag = new WorkflowDag(); |
| for (Entry<String,Set<String>> edge : edges.entrySet()) { |
| WorkflowDagEntry entry = new WorkflowDagEntry(); |
| entry.setSource(edge.getKey()); |
| entry.getTargets().addAll(edge.getValue()); |
| mergedDag.addEntry(entry); |
| } |
| return mergedDag; |
| } |
| |
| private static WorkflowContext getSanitizedWorkflow(WorkflowContext workflowContext, WorkflowContext existingWorkflowContext) { |
| WorkflowContext sanitizedWC = new WorkflowContext(); |
| if (existingWorkflowContext == null) { |
| sanitizedWC.setWorkflowDag(workflowContext.getWorkflowDag()); |
| sanitizedWC.setParentWorkflowContext(workflowContext.getParentWorkflowContext()); |
| } else { |
| sanitizedWC.setWorkflowDag(constructMergedDag(existingWorkflowContext, workflowContext)); |
| sanitizedWC.setParentWorkflowContext(existingWorkflowContext.getParentWorkflowContext()); |
| } |
| return sanitizedWC; |
| } |
| |
| private static String getWorkflowString(WorkflowContext sanitizedWC) { |
| String sanitizedWCString = null; |
| try { |
| ObjectMapper om = new ObjectMapper(); |
| sanitizedWCString = om.writeValueAsString(sanitizedWC); |
| } catch (IOException e) { |
| e.printStackTrace(); |
| sanitizedWCString = ""; |
| } |
| return sanitizedWCString; |
| } |
| |
| private void processJobSubmittedEvent( |
| PreparedStatement jobPS, |
| PreparedStatement workflowSelectPS, PreparedStatement workflowPS, |
| PreparedStatement workflowUpdateTimePS, LoggingEvent logEvent, |
| JobSubmittedEvent historyEvent) { |
| |
| try { |
| String jobId = historyEvent.getJobId().toString(); |
| jobPS.setString(1, jobId); |
| jobPS.setString(2, historyEvent.getJobName()); |
| jobPS.setString(3, historyEvent.getUserName()); |
| jobPS.setString(4, historyEvent.getJobConfPath()); |
| jobPS.setString(5, historyEvent.getJobQueueName()); |
| jobPS.setLong(6, historyEvent.getSubmitTime()); |
| |
| WorkflowContext workflowContext = buildWorkflowContext(historyEvent); |
| |
| // Get workflow information |
| boolean insertWorkflow = false; |
| String existingContextString = null; |
| |
| ResultSet rs = null; |
| try { |
| workflowSelectPS.setString(1, workflowContext.getWorkflowId()); |
| workflowSelectPS.execute(); |
| rs = workflowSelectPS.getResultSet(); |
| if (rs.next()) { |
| existingContextString = rs.getString(1); |
| } else { |
| insertWorkflow = true; |
| } |
| } catch (SQLException sqle) { |
| LOG.warn("workflow select failed with: ", sqle); |
| insertWorkflow = false; |
| } finally { |
| try { |
| if (rs != null) |
| rs.close(); |
| } catch (SQLException e) { |
| LOG.error("Exception while closing ResultSet", e); |
| } |
| } |
| |
| // Insert workflow |
| if (insertWorkflow) { |
| workflowPS.setString(1, workflowContext.getWorkflowId()); |
| workflowPS.setString(2, workflowContext.getWorkflowName()); |
| workflowPS.setString(3, getWorkflowString(getSanitizedWorkflow(workflowContext, null))); |
| workflowPS.setString(4, historyEvent.getUserName()); |
| workflowPS.setLong(5, historyEvent.getSubmitTime()); |
| workflowPS.setLong(6, historyEvent.getSubmitTime()); |
| workflowPS.setLong(7, workflowContext.getWorkflowDag().size()); |
| workflowPS.executeUpdate(); |
| LOG.debug("Successfully inserted workflowId = " + |
| workflowContext.getWorkflowId()); |
| } else { |
| ObjectMapper om = new ObjectMapper(); |
| WorkflowContext existingWorkflowContext = null; |
| try { |
| if (existingContextString != null) |
| existingWorkflowContext = om.readValue(existingContextString.getBytes(), WorkflowContext.class); |
| } catch (IOException e) { |
| LOG.warn("Couldn't read existing workflow context for " + workflowContext.getWorkflowId(), e); |
| } |
| |
| WorkflowContext sanitizedWC = getSanitizedWorkflow(workflowContext, existingWorkflowContext); |
| workflowUpdateTimePS.setString(1, getWorkflowString(sanitizedWC)); |
| workflowUpdateTimePS.setLong(2, sanitizedWC.getWorkflowDag().size()); |
| workflowUpdateTimePS.setLong(3, historyEvent.getSubmitTime()); |
| workflowUpdateTimePS.setLong(4, historyEvent.getSubmitTime()); |
| workflowUpdateTimePS.setString(5, workflowContext.getWorkflowId()); |
| workflowUpdateTimePS.setString(6, workflowContext.getWorkflowId()); |
| workflowUpdateTimePS.executeUpdate(); |
| LOG.debug("Successfully updated workflowId = " + |
| workflowContext.getWorkflowId()); |
| } |
| |
| // Insert job |
| jobPS.setString(7, workflowContext.getWorkflowId()); |
| jobPS.setString(8, workflowContext.getWorkflowEntityName()); |
| jobPS.executeUpdate(); |
| LOG.debug("Successfully inserted job = " + jobId + |
| " and workflowId = " + workflowContext.getWorkflowId()); |
| |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + " for job " + |
| historyEvent.getJobId() + " into " + JOB_TABLE, sqle); |
| } catch (Exception e) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + " for job " + |
| historyEvent.getJobId() + " into " + JOB_TABLE, e); |
| } |
| } |
| |
| private void processJobFinishedEvent( |
| PreparedStatement entityPS, |
| PreparedStatement workflowUpdateNumCompletedPS, |
| LoggingEvent logEvent, JobFinishedEvent historyEvent) { |
| Counters counters = historyEvent.getMapCounters(); |
| long inputBytes = 0; |
| if (counters != null) { |
| for (CounterGroup group : counters) { |
| for (Counter counter : group) { |
| if (counter.getName().equals("HDFS_BYTES_READ")) |
| inputBytes += counter.getValue(); |
| } |
| } |
| } |
| if (historyEvent.getFinishedReduces() != 0) |
| counters = historyEvent.getReduceCounters(); |
| long outputBytes = 0; |
| if (counters != null) { |
| for (CounterGroup group : counters) { |
| for (Counter counter : group) { |
| if (counter.getName().equals("HDFS_BYTES_WRITTEN")) |
| outputBytes += counter.getValue(); |
| } |
| } |
| } |
| try { |
| entityPS.setLong(1, historyEvent.getFinishTime()); |
| entityPS.setInt(2, historyEvent.getFinishedMaps()); |
| entityPS.setInt(3, historyEvent.getFinishedReduces()); |
| entityPS.setInt(4, historyEvent.getFailedMaps()); |
| entityPS.setInt(5, historyEvent.getFailedReduces()); |
| entityPS.setLong(6, inputBytes); |
| entityPS.setLong(7, outputBytes); |
| entityPS.setString(8, historyEvent.getJobid().toString()); |
| entityPS.executeUpdate(); |
| // job finished events always have success status |
| workflowUpdateNumCompletedPS.setLong(1, historyEvent.getFinishTime()); |
| workflowUpdateNumCompletedPS.setLong(2, historyEvent.getFinishTime()); |
| workflowUpdateNumCompletedPS.setString(3, historyEvent.getJobid().toString()); |
| workflowUpdateNumCompletedPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + " for job " + |
| historyEvent.getJobid() + " into " + JOB_TABLE, sqle); |
| } |
| |
| updateJobStatsAtFinish(historyEvent.getJobid().toString()); |
| |
| } |
| |
| private void processJobInitedEvent( |
| PreparedStatement entityPS, |
| LoggingEvent logEvent, JobInitedEvent historyEvent) { |
| try { |
| entityPS.setLong(1, historyEvent.getLaunchTime()); |
| entityPS.setInt(2, historyEvent.getTotalMaps()); |
| entityPS.setInt(3, historyEvent.getTotalReduces()); |
| entityPS.setString(4, historyEvent.getStatus()); |
| entityPS.setString(5, historyEvent.getJobId().toString()); |
| entityPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + " for job " + |
| historyEvent.getJobId() + " into " + JOB_TABLE, sqle); |
| } |
| } |
| |
| private void processJobStatusChangedEvent( |
| PreparedStatement entityPS, |
| LoggingEvent logEvent, JobStatusChangedEvent historyEvent) { |
| try { |
| entityPS.setString(1, historyEvent.getStatus()); |
| entityPS.setString(2, historyEvent.getJobId().toString()); |
| entityPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + " for job " + |
| historyEvent.getJobId() + " into " + JOB_TABLE, sqle); |
| } |
| } |
| |
| private void processJobInfoChangeEvent( |
| PreparedStatement entityPS, |
| LoggingEvent logEvent, JobInfoChangeEvent historyEvent) { |
| try { |
| entityPS.setLong(1, historyEvent.getSubmitTime()); |
| entityPS.setLong(2, historyEvent.getLaunchTime()); |
| entityPS.setString(3, historyEvent.getJobId().toString()); |
| entityPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + " for job " + |
| historyEvent.getJobId() + " into " + JOB_TABLE, sqle); |
| } |
| } |
| |
| private void processJobUnsuccessfulEvent( |
| PreparedStatement entityPS, |
| LoggingEvent logEvent, JobUnsuccessfulCompletionEvent historyEvent) { |
| try { |
| entityPS.setLong(1, historyEvent.getFinishTime()); |
| entityPS.setLong(2, historyEvent.getFinishedMaps()); |
| entityPS.setLong(3, historyEvent.getFinishedReduces()); |
| entityPS.setString(4, historyEvent.getStatus()); |
| entityPS.setString(5, historyEvent.getJobId().toString()); |
| entityPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + " for job " + |
| historyEvent.getJobId() + " into " + JOB_TABLE, sqle); |
| } |
| |
| updateJobStatsAtFinish(historyEvent.getJobId().toString()); |
| } |
| |
| private void processTaskStartedEvent(PreparedStatement entityPS, |
| LoggingEvent logEvent, TaskStartedEvent historyEvent) { |
| try { |
| entityPS.setString(1, |
| historyEvent.getTaskId().getJobID().toString()); |
| entityPS.setString(2, historyEvent.getTaskType().toString()); |
| entityPS.setString(3, historyEvent.getSplitLocations()); |
| entityPS.setLong(4, historyEvent.getStartTime()); |
| entityPS.setString(5, historyEvent.getTaskId().toString()); |
| entityPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + " for task " + |
| historyEvent.getTaskId() + " into " + TASK_TABLE, sqle); |
| } |
| } |
| |
| private void processTaskFinishedEvent( |
| PreparedStatement entityPS, |
| LoggingEvent logEvent, TaskFinishedEvent historyEvent) { |
| try { |
| entityPS.setString(1, |
| historyEvent.getTaskId().getJobID().toString()); |
| entityPS.setString(2, historyEvent.getTaskType().toString()); |
| entityPS.setString(3, historyEvent.getTaskStatus()); |
| entityPS.setLong(4, historyEvent.getFinishTime()); |
| entityPS.setString(5, historyEvent.getTaskId().toString()); |
| entityPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + " for task " + |
| historyEvent.getTaskId() + " into " + TASK_TABLE, sqle); |
| } |
| } |
| |
| private void processTaskFailedEvent( |
| PreparedStatement entityPS, |
| LoggingEvent logEvent, TaskFailedEvent historyEvent) { |
| try { |
| entityPS.setString(1, |
| historyEvent.getTaskId().getJobID().toString()); |
| entityPS.setString(2, historyEvent.getTaskType().toString()); |
| entityPS.setString(3, historyEvent.getTaskStatus()); |
| entityPS.setLong(4, historyEvent.getFinishTime()); |
| entityPS.setString(5, historyEvent.getError()); |
| if (historyEvent.getFailedAttemptID() != null) { |
| entityPS.setString(6, historyEvent.getFailedAttemptID().toString()); |
| } else { |
| entityPS.setString(6, "task_na"); |
| } |
| entityPS.setString(7, historyEvent.getTaskId().toString()); |
| entityPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + " for task " + |
| historyEvent.getTaskId() + " into " + TASK_TABLE, sqle); |
| } |
| } |
| |
| private void processTaskAttemptStartedEvent( |
| PreparedStatement entityPS, |
| LoggingEvent logEvent, TaskAttemptStartedEvent historyEvent) { |
| try { |
| entityPS.setString(1, |
| historyEvent.getTaskId().getJobID().toString()); |
| entityPS.setString(2, historyEvent.getTaskId().toString()); |
| entityPS.setString(3, historyEvent.getTaskType().toString()); |
| entityPS.setLong(4, historyEvent.getStartTime()); |
| entityPS.setString(5, historyEvent.getTrackerName()); |
| entityPS.setString(6, historyEvent.getLocality().toString()); |
| entityPS.setString(7, historyEvent.getAvataar().toString()); |
| entityPS.setString(8, historyEvent.getTaskAttemptId().toString()); |
| entityPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + |
| " for taskAttempt " + historyEvent.getTaskAttemptId() + |
| " into " + TASKATTEMPT_TABLE, sqle); |
| } |
| } |
| |
| private void processTaskAttemptFinishedEvent( |
| PreparedStatement entityPS, |
| LoggingEvent logEvent, TaskAttemptFinishedEvent historyEvent) { |
| |
| if (historyEvent.getTaskType() == TaskType.MAP || |
| historyEvent.getTaskType() == TaskType.REDUCE) { |
| LOG.debug("Ignoring TaskAttemptFinishedEvent for " + |
| historyEvent.getTaskType()); |
| return; |
| } |
| |
| try { |
| entityPS.setString(1, |
| historyEvent.getTaskId().getJobID().toString()); |
| entityPS.setString(2, historyEvent.getTaskId().toString()); |
| entityPS.setString(3, historyEvent.getTaskType().toString()); |
| entityPS.setLong(4, historyEvent.getFinishTime()); |
| entityPS.setString(5, historyEvent.getTaskStatus()); |
| entityPS.setString(6, historyEvent.getHostname()); |
| entityPS.setString(7, historyEvent.getAttemptId().toString()); |
| entityPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + |
| " for taskAttempt " + historyEvent.getAttemptId() + |
| " into " + TASKATTEMPT_TABLE, sqle); |
| } |
| } |
| |
| private void processTaskAttemptUnsuccessfulEvent( |
| PreparedStatement entityPS, |
| LoggingEvent logEvent, |
| TaskAttemptUnsuccessfulCompletionEvent historyEvent) { |
| try { |
| entityPS.setString(1, |
| historyEvent.getTaskId().getJobID().toString()); |
| entityPS.setString(2, historyEvent.getTaskId().toString()); |
| entityPS.setString(3, historyEvent.getTaskType().toString()); |
| entityPS.setLong(4, historyEvent.getFinishTime()); |
| entityPS.setString(5, historyEvent.getTaskStatus()); |
| entityPS.setString(6, historyEvent.getHostname()); |
| entityPS.setString(7, historyEvent.getError()); |
| entityPS.setString(8, historyEvent.getTaskAttemptId().toString()); |
| entityPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + |
| " for taskAttempt " + historyEvent.getTaskAttemptId() + |
| " into " + TASKATTEMPT_TABLE, sqle); |
| } |
| } |
| |
| private void processMapAttemptFinishedEvent( |
| PreparedStatement entityPS, |
| LoggingEvent logEvent, MapAttemptFinishedEvent historyEvent) { |
| |
| if (historyEvent.getTaskType() != TaskType.MAP) { |
| LOG.debug("Ignoring MapAttemptFinishedEvent for " + |
| historyEvent.getTaskType()); |
| return; |
| } |
| |
| long[] ioBytes = getInputOutputBytes(historyEvent.getCounters()); |
| |
| try { |
| entityPS.setString(1, |
| historyEvent.getTaskId().getJobID().toString()); |
| entityPS.setString(2, historyEvent.getTaskId().toString()); |
| entityPS.setString(3, historyEvent.getTaskType().toString()); |
| entityPS.setLong(4, historyEvent.getMapFinishTime()); |
| entityPS.setLong(5, historyEvent.getFinishTime()); |
| entityPS.setLong(6, ioBytes[0]); |
| entityPS.setLong(7, ioBytes[1]); |
| entityPS.setString(8, historyEvent.getTaskStatus()); |
| entityPS.setString(9, historyEvent.getHostname()); |
| entityPS.setString(10, historyEvent.getAttemptId().toString()); |
| entityPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + |
| " for taskAttempt " + historyEvent.getAttemptId() + |
| " into " + TASKATTEMPT_TABLE, sqle); |
| } |
| } |
| |
| |
| private void processReduceAttemptFinishedEvent( |
| PreparedStatement entityPS, |
| LoggingEvent logEvent, ReduceAttemptFinishedEvent historyEvent) { |
| if (historyEvent.getTaskType() != TaskType.REDUCE) { |
| LOG.debug("Ignoring ReduceAttemptFinishedEvent for " + |
| historyEvent.getTaskType()); |
| return; |
| } |
| |
| long[] ioBytes = getInputOutputBytes(historyEvent.getCounters()); |
| |
| try { |
| entityPS.setString(1, |
| historyEvent.getTaskId().getJobID().toString()); |
| entityPS.setString(2, historyEvent.getTaskId().toString()); |
| entityPS.setString(3, historyEvent.getTaskType().toString()); |
| entityPS.setLong(4, historyEvent.getShuffleFinishTime()); |
| entityPS.setLong(5, historyEvent.getSortFinishTime()); |
| entityPS.setLong(6, historyEvent.getFinishTime()); |
| entityPS.setLong(7, ioBytes[0]); |
| entityPS.setLong(8, ioBytes[1]); |
| entityPS.setString(9, historyEvent.getTaskStatus()); |
| entityPS.setString(10, historyEvent.getHostname()); |
| entityPS.setString(11, historyEvent.getAttemptId().toString()); |
| entityPS.executeUpdate(); |
| } catch (SQLException sqle) { |
| LOG.info("Failed to store " + historyEvent.getEventType() + |
| " for taskAttempt " + historyEvent.getAttemptId() + |
| " into " + TASKATTEMPT_TABLE, sqle); |
| } |
| } |
| |
| public static long[] getInputOutputBytes(JhCounters counters) { |
| long inputBytes = 0; |
| long outputBytes = 0; |
| if (counters != null) { |
| for (JhCounterGroup counterGroup : counters.groups) { |
| if (counterGroup.name.equals("FileSystemCounters")) { |
| for (JhCounter counter : counterGroup.counts) { |
| if (counter.name.equals("HDFS_BYTES_READ") || |
| counter.name.equals("FILE_BYTES_READ")) |
| inputBytes += counter.value; |
| else if (counter.name.equals("HDFS_BYTES_WRITTEN") || |
| counter.name.equals("FILE_BYTES_WRITTEN")) |
| outputBytes += counter.value; |
| } |
| } |
| } |
| } |
| return new long[]{inputBytes, outputBytes}; |
| } |
| |
| |
| @Override |
| public void update(LoggingEvent originalEvent, Object parsedEvent) |
| throws IOException { |
| try { |
| doUpdates(originalEvent, parsedEvent); |
| } catch (SQLException sqle) { |
| throw new IOException(sqle); |
| } |
| } |
| |
| } |
