UIDataAggregator { public HistoryDetailStageVO aggregateHistoryDetialStage(SparkEventLogBean eventLog, int stageId){ HistoryDetailStageVO historyStage = new HistoryDetailStageVO(); List<TaskStart> taskStartList = eventLog.getTaskStart(); List<TaskEnd> taskEndList = eventLog.getTaskEnd(); Map<Integer, TaskStart> tempMapTaskStart = new HashMap<Integer, TaskStart>(); Map<String, Integer> totalTasks = new HashMap<String, Integer>(); int totalCompletedTask = 0; long mindurationTask = Integer.MAX_VALUE; long minGCTime = Integer.MAX_VALUE; long minInputSize = Integer.MAX_VALUE; long minRecord = Integer.MAX_VALUE; long maxdurationTask = 0; long maxGCTime = 0; long maxInputSize = 0; long maxRecord = 0; List<Long> durationTaskList = new ArrayList<Long>(); List<Long> gcTimeList = new ArrayList<Long>(); List<Long> inputSizeList = new ArrayList<Long>(); List<Long> recordList = new ArrayList<Long>(); Map<String, Integer> failedTotalTasks = new HashMap<String, Integer>(); Map<String, Integer> succeededTotalTasks = new HashMap<String, Integer>(); Map<String, Long> taskTotalTaskTime = new HashMap<String, Long>(); Map<String, Long> inputSizeTotalTasks = new HashMap<String, Long>(); Map<String, Long> recordTotalTasks = new HashMap<String, Long>(); for(TaskStart taskStart : taskStartList){ tempMapTaskStart.put(taskStart.getInfo().getIndex(), taskStart); } Map<Integer, TaskEnd> tempMapTaskEnd = new HashMap<Integer, TaskEnd>(); for(TaskEnd taskEnd : taskEndList){ tempMapTaskEnd.put(taskEnd.getInfo().getIndex(), taskEnd); } List<TasksVO> taskVOList = new ArrayList<TasksVO>(); for(int i = 0 ; i < tempMapTaskStart.size(); i++){ TasksVO taskVO = new TasksVO(); TaskStart taskStart = tempMapTaskStart.get(i); TaskEnd taskEnd = tempMapTaskEnd.get(i); if(taskStart != null && taskStart.getStageId() == stageId){ taskVO.setAttempt(taskStart.getInfo().getAttempt()); taskVO.setIndex(taskStart.getInfo().getIndex()); taskVO.setId(taskStart.getInfo().getId()); taskVO.setLaunchTime(String.valueOf(taskStart.getInfo().getLanuchTime())); taskVO.setLocality(taskStart.getInfo().getLocality()); String host = taskStart.getInfo().getHost(); taskVO.setExecutorID(taskStart.getInfo().getExecutorId()); taskVO.setHost(host); if(totalTasks.get(host) != null){ int taskNumber = totalTasks.get(host); totalTasks.put(host, taskNumber + 1); }else{ totalTasks.put(host, 1); } totalCompletedTask++; } if(taskEnd != null && taskEnd.getStageId() == stageId){ taskVO.setFinishTime(String.valueOf(taskEnd.getInfo().getFinishTime())); taskVO.setGcTime(String.valueOf(taskEnd.getMetrics().getGcTime())); minGCTime = this.minValue(minGCTime, taskEnd.getMetrics().getGcTime()); maxGCTime = this.maxValue(maxGCTime, taskEnd.getMetrics().getGcTime()); gcTimeList.add(taskEnd.getMetrics().getGcTime()); taskVO.setResultSize(taskEnd.getMetrics().getResultSize()); taskVO.setRunTime(taskEnd.getMetrics().getRunTime()); taskVO.setStatus(taskEnd.getReason().getReason()); if(taskEnd.getInfo().isFailed() == true){ if(failedTotalTasks.get(taskEnd.getMetrics().getHost()) != null){ Integer failedNumber = failedTotalTasks.get(taskEnd.getMetrics().getHost()); failedTotalTasks.put(taskEnd.getMetrics().getHost(), failedNumber + 1); }else{ failedTotalTasks.put(taskEnd.getMetrics().getHost(), 1); } } if(taskEnd.getReason().getReason().equals("Success")){ if(succeededTotalTasks.get(taskEnd.getMetrics().getHost()) != null){ Integer succeededNumber = succeededTotalTasks.get(taskEnd.getMetrics().getHost()); succeededTotalTasks.put(taskEnd.getMetrics().getHost(), succeededNumber + 1); }else{ succeededTotalTasks.put(taskEnd.getMetrics().getHost(), 1); } } if(taskTotalTaskTime.get(taskEnd.getMetrics().getHost()) != null){ Long taskTotalTimeNumber = taskTotalTaskTime.get(taskEnd.getMetrics().getHost()); taskTotalTaskTime.put(taskEnd.getMetrics().getHost(), taskTotalTimeNumber + taskEnd.getMetrics().getRunTime()); }else{ taskTotalTaskTime.put(taskEnd.getMetrics().getHost(), taskEnd.getMetrics().getRunTime()); } mindurationTask = minValue(mindurationTask, taskEnd.getMetrics().getRunTime()); maxdurationTask = maxValue(maxdurationTask, taskEnd.getMetrics().getRunTime()); durationTaskList.add(taskEnd.getMetrics().getRunTime()); } if(taskEnd.getMetrics() != null && taskEnd.getMetrics().getInputMetrics() != null && taskEnd.getStageId() == stageId){ taskVO.setInputSize(String.valueOf(taskEnd.getMetrics().getInputMetrics().getReadByte())); taskVO.setRecords(String.valueOf(taskEnd.getMetrics().getInputMetrics().getRecordRead())); if(inputSizeTotalTasks.get(taskEnd.getMetrics().getHost()) != null){ Long inputSizeTotalNumber = inputSizeTotalTasks.get(taskEnd.getMetrics().getHost()); inputSizeTotalTasks.put(taskEnd.getMetrics().getHost(), inputSizeTotalNumber + taskEnd.getMetrics().getInputMetrics().getReadByte()); }else{ inputSizeTotalTasks.put(taskEnd.getMetrics().getHost(), taskEnd.getMetrics().getInputMetrics().getReadByte()); } minInputSize = minValue(minInputSize, taskEnd.getMetrics().getInputMetrics().getReadByte()); maxInputSize = maxValue(maxInputSize, taskEnd.getMetrics().getInputMetrics().getReadByte()); inputSizeList.add(taskEnd.getMetrics().getInputMetrics().getReadByte()); if(recordTotalTasks.get(taskEnd.getMetrics().getHost()) != null){ Long recordTotalNumber = recordTotalTasks.get(taskEnd.getMetrics().getHost()); recordTotalTasks.put(taskEnd.getMetrics().getHost(), recordTotalNumber + taskEnd.getMetrics().getInputMetrics().getRecordRead()); }else{ recordTotalTasks.put(taskEnd.getMetrics().getHost(), taskEnd.getMetrics().getInputMetrics().getRecordRead()); } minRecord = minValue(minRecord, taskEnd.getMetrics().getInputMetrics().getRecordRead()); maxRecord = maxValue(maxRecord, taskEnd.getMetrics().getInputMetrics().getRecordRead()); recordList.add(taskEnd.getMetrics().getInputMetrics().getRecordRead()); } taskVOList.add(taskVO); } historyStage.setTasks(taskVOList); List<BlockManager> blockManagerList = eventLog.getBlockManager(); List<SummaryExecutorVO> aggregatorExecutor = new ArrayList<SummaryExecutorVO>(); for(BlockManager blockManager : blockManagerList){ SummaryExecutorVO summaryExecutorVO = new SummaryExecutorVO(); summaryExecutorVO.setAddress(blockManager.getBlockManagerID().getHost() + ":" + blockManager.getBlockManagerID().getPort()); if(totalTasks.get(blockManager.getBlockManagerID().getHost()) != null){ Integer resultTotalTasks = totalTasks.get(blockManager.getBlockManagerID().getHost()); summaryExecutorVO.setTotalTasks(resultTotalTasks); } if(failedTotalTasks.get(blockManager.getBlockManagerID().getHost()) != null){ Integer resultFailedTotalTasks = failedTotalTasks.get(blockManager.getBlockManagerID().getHost()); summaryExecutorVO.setFailedTasks(resultFailedTotalTasks); } if(succeededTotalTasks.get(blockManager.getBlockManagerID().getHost()) != null){ Integer resultSucceededTotalTasks = succeededTotalTasks.get(blockManager.getBlockManagerID().getHost()); summaryExecutorVO.setSucceededTasks(resultSucceededTotalTasks); } if(taskTotalTaskTime.get(blockManager.getBlockManagerID().getHost()) != null){ Long resultTaskTotalTaskTime = taskTotalTaskTime.get(blockManager.getBlockManagerID().getHost()); summaryExecutorVO.setTaskTime(resultTaskTotalTaskTime); } if(inputSizeTotalTasks.get(blockManager.getBlockManagerID().getHost()) != null){ Long inputTotalTask = inputSizeTotalTasks.get(blockManager.getBlockManagerID().getHost()); summaryExecutorVO.setInputSize(String.valueOf(inputTotalTask)); } if(recordTotalTasks.get(blockManager.getBlockManagerID().getHost()) != null){ Long recordTotalTask = recordTotalTasks.get(blockManager.getBlockManagerID().getHost()); summaryExecutorVO.setRecords(String.valueOf(recordTotalTask)); } summaryExecutorVO.setExecuteId(String.valueOf(blockManager.getBlockManagerID().getId())); summaryExecutorVO.setMaxMemory(blockManager.getMaxMemory()); aggregatorExecutor.add(summaryExecutorVO); } historyStage.setAggregatorExecutor(aggregatorExecutor); historyStage.setCompleteTaskSize(totalCompletedTask); List<MetricCompletedTasksVO> metricCompletedTasks = new ArrayList<MetricCompletedTasksVO>(); MetricCompletedTasksVO metricCompletedDuration = new MetricCompletedTasksVO(); metricCompletedDuration.setMetric("Duration"); if(durationTaskList.toArray().length > 0){ metricCompletedDuration.setMin(mindurationTask); metricCompletedDuration.setMax(maxdurationTask); metricCompletedDuration.setMedian(this.medianValue(durationTaskList.toArray())); } MetricCompletedTasksVO metricCompletedGCTime = new MetricCompletedTasksVO(); metricCompletedGCTime.setMetric("GC Time"); if(gcTimeList.toArray().length > 0){ metricCompletedGCTime.setMin(minGCTime); metricCompletedGCTime.setMax(maxGCTime); metricCompletedGCTime.setMedian(this.medianValue(gcTimeList.toArray())); } MetricCompletedTasksVO metricCompletedInputSize = new MetricCompletedTasksVO(); metricCompletedInputSize.setMetric("Input Size"); if(inputSizeList.toArray().length > 0){ metricCompletedInputSize.setMin(minInputSize); metricCompletedInputSize.setMax(maxInputSize); metricCompletedInputSize.setMedian(this.medianValue(inputSizeList.toArray())); } MetricCompletedTasksVO metricRecords = new MetricCompletedTasksVO(); metricRecords.setMetric("Record"); if(recordList.toArray().length > 0){ metricRecords.setMin(minRecord); metricRecords.setMax(maxRecord); metricRecords.setMedian(this.medianValue(recordList.toArray())); } metricCompletedTasks.add(metricCompletedDuration); metricCompletedTasks.add(metricCompletedGCTime); metricCompletedTasks.add(metricCompletedInputSize); metricCompletedTasks.add(metricRecords); historyStage.setMetricCompletedTasks(metricCompletedTasks); return historyStage; } UIDataAggregator(); DefaultTotalNodeVO aggregateClusterDashBoard(List<String> metrics, int runningAppNum, List<String> periodSparkAppId, 
			List<String> allNodeHost, Map<String, List<NodeBean>> nodeMetricsByAppId, 
			Map<String, SparkEventLogBean> inProgressEventLogByAppId, List<Long> plotPointInPeriod); DefaultApplicationVO aggregateJobDashBoard(List<String> metrics, List<String> allNodeHost, 
			SparkEventLogBean inProgressLog, List<NodeBean> nodeMetrics, List<Long> plotPointInPeriod); DefaultNodeVO aggregateNodeDashBoard(List<String> metrics, String node,
			Map<String, NodeBean> nodeMetricsByAppId, List<Long> plotPointInPeriod); List<JobVO> aggregateApplicationJobs(String applicationId,
			SparkEventLogBean eventLog); List<StageVO> aggregateJobStages(String applicationId, String jobId,
			SparkEventLogBean eventLog); HistoryDetailStageVO aggregateHistoryDetialStage(SparkEventLogBean eventLog, int stageId);  }
@Test public void testgetHistoryDetialStage() throws Exception{ VarOneConfiguration conf = VarOneConfiguration.create(); VarOneEnv env = new VarOneEnv(conf); Configuration config = env.loadHadoopConfiguration(); UIDataAggregator aggregator = new UIDataAggregator(); EventLogHdfsReaderImpl hdfsReader = new EventLogHdfsReaderImpl(config, env.getEventLogDir()); SparkEventLogBean sparkEventLog = hdfsReader.getHistoryStageDetails("application_1449569227858_0664"); HistoryDetailStageVO historyDetailsVO = aggregator.aggregateHistoryDetialStage(sparkEventLog, 1); List<TasksVO> taskVOList = historyDetailsVO.getTasks(); }