<commit_before>/**
 * 
 */
package com.varone.web.aggregator;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeSet;

import com.varone.web.aggregator.unit.TimeUnitTransfer;
import com.varone.web.eventlog.bean.SparkEventLogBean;
import com.varone.web.eventlog.bean.SparkEventLogBean.ExecutorAdded;
import com.varone.web.eventlog.bean.SparkEventLogBean.JobEnd;
import com.varone.web.eventlog.bean.SparkEventLogBean.JobStart;
import com.varone.web.eventlog.bean.SparkEventLogBean.StageCompleted;
import com.varone.web.eventlog.bean.SparkEventLogBean.StageInfos;
import com.varone.web.eventlog.bean.SparkEventLogBean.StageSubmit;
import com.varone.web.eventlog.bean.SparkEventLogBean.TaskEnd;
import com.varone.web.eventlog.bean.SparkEventLogBean.TaskStart;
import com.varone.web.eventlog.bean.SparkEventLogBean.TaskEnd.TaskMetrics;
import com.varone.web.metrics.bean.MetricBean;
import com.varone.web.metrics.bean.NodeBean;
import com.varone.web.metrics.bean.TimeValuePairBean;
import com.varone.web.vo.DefaultApplicationVO;
import com.varone.web.vo.DefaultNodeVO;
import com.varone.web.vo.DefaultTotalNodeVO;
import com.varone.web.vo.JobVO;
import com.varone.web.vo.MetricPropVO;
import com.varone.web.vo.StageVO;
import com.varone.web.vo.TimeValueVO;

/**
 * @author allen
 *
 */
public class UIDataAggregator {
	
	private TimeUnitTransfer timeUnitTransfer;
	
	/**
	 * 
	 */
	public UIDataAggregator() {
		this.timeUnitTransfer = new TimeUnitTransfer();
	}
	
	
	public 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) throws Exception{
		DefaultTotalNodeVO result = new DefaultTotalNodeVO();
		int nodeNum = allNodeHost.size();
		int jobNum  = runningAppNum;
		int taskNum = 0, executorNum = 0;
		
		List<MetricPropVO> metricProps = new ArrayList<MetricPropVO>();
		Map<String, Integer> taskStartedNumByNode = new LinkedHashMap<String, Integer>();
		Map<String, Integer> executorNumByNode = new LinkedHashMap<String, Integer>();
		Map<String, Map<String, List<TimeValueVO>>> propToMetrics = new LinkedHashMap<String, Map<String, List<TimeValueVO>>>();
		
		for(String metric: metrics){
			String[] propertyAndTitle = UIMrtricsPropTransfer.getUIMetricPropertyByMetricName(metric);
			MetricPropVO metricPropVO = new MetricPropVO();
			metricPropVO.setProperty(propertyAndTitle[0]);
			metricPropVO.setTitle(propertyAndTitle[1]);
			metricProps.add(metricPropVO);
			Map<String, List<TimeValueVO>> host2Metrics = new LinkedHashMap<String, List<TimeValueVO>>();
			for(String host: allNodeHost){
				
				List<TimeValueVO> defaultValues = new ArrayList<TimeValueVO>();
				for(Long time: plotPointInPeriod){
					TimeValueVO pair = new TimeValueVO();
					pair.setTime(this.timeUnitTransfer.transferToAxisX(time));
					pair.setValue("0");
					defaultValues.add(pair);
				}
				
				host2Metrics.put(host, defaultValues);
			}
			propToMetrics.put(propertyAndTitle[0], host2Metrics);
		}
		
		for(String host: allNodeHost){
			executorNumByNode.put(host, 0);
			taskStartedNumByNode.put(host, 0);
		}

		
		
		for(Entry<String, SparkEventLogBean> entry: inProgressEventLogByAppId.entrySet()){
			SparkEventLogBean eventLog = entry.getValue();
			taskNum += (eventLog.getTaskStart().size() - eventLog.getTaskEnd().size());
			executorNum += eventLog.getExecutorAdd().size();
			
			for(TaskStart taskStart: eventLog.getTaskStart()){
				int n = taskStartedNumByNode.get(taskStart.getInfo().getHost());
				taskStartedNumByNode.put(taskStart.getInfo().getHost(), ++n);
			}
			
			for(ExecutorAdded executorAdd: eventLog.getExecutorAdd()){
				int n = executorNumByNode.get(executorAdd.getInfo().getHost());
				executorNumByNode.put(executorAdd.getInfo().getHost(), ++n);
			}
		}
		
		for(Entry<String, List<NodeBean>> entry: nodeMetricsByAppId.entrySet()){
			for(NodeBean node: entry.getValue()){
				String host = node.getHost();
				for(MetricBean metric: node.getMetrics()){
					String propertyName = UIMrtricsPropTransfer.getUIMetricPropertyByMetricValue(metric.getName());
					
					Map<String, List<TimeValueVO>> host2Metric = propToMetrics.get(propertyName);
					
					// get a new one
					List<TimeValuePairBean> newValues = metric.getValues();
					// get current value
					List<TimeValueVO> currValues = host2Metric.get(host);
					// aggregate and store
					
					for(TimeValuePairBean newPair: newValues){
						for(TimeValueVO currPair: currValues){
							if(this.timeUnitTransfer.transferToAxisX(
									newPair.getTime()).equals(currPair.getTime())){
								long newValue = Long.parseLong(newPair.getValue());
								long currValue = Long.parseLong(currPair.getValue());
								currPair.setValue(String.valueOf(newValue+currValue));
								break;
							}
						}
					}
				}
			}
		}
		result.setExecutorNumByNode(executorNumByNode);
		result.setExecutorNum(executorNum);
		result.setJobNum(jobNum);
		result.setNodeNum(nodeNum);
		result.setTaskNum(taskNum);
		result.setTaskStartedNumByNode(taskStartedNumByNode);
		result.setPropToMetrics(propToMetrics);<commit_msg>aggregate backend metrics data with data format<commit_after>/**
 * 
 */
package com.varone.web.aggregator;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeSet;

import org.apache.commons.lang.NumberUtils;

import com.varone.web.aggregator.unit.TimeUnitTransfer;
import com.varone.web.eventlog.bean.SparkEventLogBean;
import com.varone.web.eventlog.bean.SparkEventLogBean.ExecutorAdded;
import com.varone.web.eventlog.bean.SparkEventLogBean.JobEnd;
import com.varone.web.eventlog.bean.SparkEventLogBean.JobStart;
import com.varone.web.eventlog.bean.SparkEventLogBean.StageCompleted;
import com.varone.web.eventlog.bean.SparkEventLogBean.StageInfos;
import com.varone.web.eventlog.bean.SparkEventLogBean.StageSubmit;
import com.varone.web.eventlog.bean.SparkEventLogBean.TaskEnd;
import com.varone.web.eventlog.bean.SparkEventLogBean.TaskStart;
import com.varone.web.eventlog.bean.SparkEventLogBean.TaskEnd.TaskMetrics;
import com.varone.web.metrics.bean.MetricBean;
import com.varone.web.metrics.bean.NodeBean;
import com.varone.web.metrics.bean.TimeValuePairBean;
import com.varone.web.vo.DefaultApplicationVO;
import com.varone.web.vo.DefaultNodeVO;
import com.varone.web.vo.DefaultTotalNodeVO;
import com.varone.web.vo.JobVO;
import com.varone.web.vo.MetricPropVO;
import com.varone.web.vo.StageVO;
import com.varone.web.vo.TimeValueVO;

/**
 * @author allen
 *
 */
public class UIDataAggregator {
	
	private TimeUnitTransfer timeUnitTransfer;
	
	/**
	 * 
	 */
	public UIDataAggregator() {
		this.timeUnitTransfer = new TimeUnitTransfer();
	}
	
	
	public 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) throws Exception{
		DefaultTotalNodeVO result = new DefaultTotalNodeVO();
		int nodeNum = allNodeHost.size();
		int jobNum  = runningAppNum;
		int taskNum = 0, executorNum = 0;
		
		List<MetricPropVO> metricProps = new ArrayList<MetricPropVO>();
		Map<String, Integer> taskStartedNumByNode = new LinkedHashMap<String, Integer>();
		Map<String, Integer> executorNumByNode = new LinkedHashMap<String, Integer>();
		Map<String, Map<String, List<TimeValueVO>>> propToMetrics = new LinkedHashMap<String, Map<String, List<TimeValueVO>>>();
		
		for(String metric: metrics){
			String[] metricInfo = UIMrtricsPropTransfer.getUIMetricPropertyByMetricName(metric);
			MetricPropVO metricPropVO = new MetricPropVO();
			metricPropVO.setProperty(metricInfo[0]);
			metricPropVO.setTitle(metricInfo[1]);
			metricPropVO.setFormat(metricInfo[2]);
			metricProps.add(metricPropVO);
			Map<String, List<TimeValueVO>> host2Metrics = new LinkedHashMap<String, List<TimeValueVO>>();
			for(String host: allNodeHost){
				
				List<TimeValueVO> defaultValues = new ArrayList<TimeValueVO>();
				for(Long time: plotPointInPeriod){
					TimeValueVO pair = new TimeValueVO();
					pair.setTime(this.timeUnitTransfer.transferToAxisX(time));
					if(metricInfo[2].equals("PERCENTAGE") || 
					   metricInfo[2].equals("MILLIS") || 
					   metricInfo[2].equals("OPS")){
						pair.setValue("0,0");
					} else {
						pair.setValue("0");
					}
					defaultValues.add(pair);
				}
				
				host2Metrics.put(host, defaultValues);
			}
			propToMetrics.put(metricInfo[0], host2Metrics);
		}
		
		for(String host: allNodeHost){
			executorNumByNode.put(host, 0);
			taskStartedNumByNode.put(host, 0);
		}

		
		
		for(Entry<String, SparkEventLogBean> entry: inProgressEventLogByAppId.entrySet()){
			SparkEventLogBean eventLog = entry.getValue();
			taskNum += (eventLog.getTaskStart().size() - eventLog.getTaskEnd().size());
			executorNum += eventLog.getExecutorAdd().size();
			
			for(TaskStart taskStart: eventLog.getTaskStart()){
				int n = taskStartedNumByNode.get(taskStart.getInfo().getHost());
				taskStartedNumByNode.put(taskStart.getInfo().getHost(), ++n);
			}
			
			for(ExecutorAdded executorAdd: eventLog.getExecutorAdd()){
				int n = executorNumByNode.get(executorAdd.getInfo().getHost());
				executorNumByNode.put(executorAdd.getInfo().getHost(), ++n);
			}
		}
		
		for(Entry<String, List<NodeBean>> entry: nodeMetricsByAppId.entrySet()){
			for(NodeBean node: entry.getValue()){
				String host = node.getHost();
				for(MetricBean metric: node.getMetrics()){
					String propertyName = UIMrtricsPropTransfer.getUIMetricPropertyByMetricValue(metric.getName());
					String format = UIMrtricsPropTransfer.getUIMetricPropertyByJsProp(propertyName)[2];
					Map<String, List<TimeValueVO>> host2Metric = propToMetrics.get(propertyName);
					
					// get a new one
					List<TimeValuePairBean> newValues = metric.getValues();
					// get current value
					List<TimeValueVO> currValues = host2Metric.get(host);
					// aggregate and store
					
					for(TimeValuePairBean newPair: newValues){
						for(TimeValueVO currPair: currValues){
							if(this.timeUnitTransfer.transferToAxisX(
									newPair.getTime()).equals(currPair.getTime())){
								if(format.equals("BYTE") || format.equals("NONE")){
									long newValue = Long.parseLong(newPair.getValue());
									long currValue = Long.parseLong(currPair.getValue());
									currPair.setValue(String.valueOf(newValue+currValue));
								} else if(format.equals("PERCENTAGE")){
									String[] valueAndCount = currPair.getValue().split(",");
									double newValue = Double.parseDouble(newPair.getValue());
									double currValue = Double.parseDouble(valueAndCount[0]);
									valueAndCount[1] = Integer.valueOf(valueAndCount[1])+1+"";
									currPair.setValue(String.valueOf(newValue+currValue)+","+valueAndCount[1]);
								} else {
									//MILLIS or OPS
									String[] valueAndCount = currPair.getValue().split(",");
									long newValue = Long.parseLong(newPair.getValue());
									long currValue = Long.parseLong(valueAndCount[0]);
									valueAndCount[1] = Integer.valueOf(valueAndCount[1])+1+"";
									currPair.setValue(String.valueOf(newValue+currValue)+","+valueAndCount[1]);
								}
								
								break;
							}
						}
					}
				}
			}
		}
		
		for(MetricPropVO metricPropVO: metricProps){
			if(metricPropVO.getFormat().equals("PERCENTAGE")){
				Map<String, List<TimeValueVO>> map = propToMetrics.get(metricPropVO.getProperty());
				for(List<TimeValueVO> periodData: map.values()){
					for(TimeValueVO timeValue: periodData){
						String[] valueAndCount = timeValue.getValue().split(",");
						double sum = Double.parseDouble(valueAndCount[0]);
						if(sum > 0){
							double avg = sum / Integer.parseInt(valueAndCount[1]);
							timeValue.setValue(String.valueOf(avg));
						} else {
							timeValue.setValue("0");
						}
					}
				}
			} else if(metricPropVO.getFormat().equals("MILLIS") || metricPropVO.getFormat().equals("OPS")) {
				Map<String, List<TimeValueVO>> map = propToMetrics.get(metricPropVO.getProperty());
				for(List<TimeValueVO> periodData: map.values()){
					for(TimeValueVO timeValue: periodData){
						String[] valueAndCount = timeValue.getValue().split(",");
						long sum = Long.parseLong(valueAndCount[0]);
						if(sum > 0){
							long avg = sum / Integer.parseInt(valueAndCount[1]);
							timeValue.setValue(String.valueOf(avg));
						} else {
							timeValue.setValue("0");
						}
					}
				}
			}
		}

		result.setExecutorNumByNode(executorNumByNode);
		result.setExecutorNum(executorNum);
		result.setJobNum(jobNum);
		result.setNodeNum(nodeNum);
		result.setTaskNum(taskNum);
		result.setTaskStartedNumByNode(taskStartedNumByNode);
		result.setPropToMetrics(propToMetrics);
{'fraction_non_alphanumeric': 0.12805711681604132, 'fraction_numerical': 0.0038736138538660184, 'mean_word_length': 4.586338565973695, 'pattern_counts': {'":': 0, '<': 73, '<?xml version=': 0, '>': 75, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '680845', 'n_tokens_mistral': 5007, 'n_tokens_neox': 4199, 'n_words': 635}