[EAGLE-538] Add Mapreduce task level apis https://issues.apache.org/jira/browse/EAGLE-538
https://issues.apache.org/jira/browse/EAGLE-554 Author: Qingwen Zhao <qingwen...@gmail.com> Author: Zhao, Qingwen <qingwz...@ebay.com> Closes #432 from qingwen220/dataSkew. Project: http://git-wip-us.apache.org/repos/asf/incubator-eagle/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-eagle/commit/cfd5c7f8 Tree: http://git-wip-us.apache.org/repos/asf/incubator-eagle/tree/cfd5c7f8 Diff: http://git-wip-us.apache.org/repos/asf/incubator-eagle/diff/cfd5c7f8 Branch: refs/heads/master Commit: cfd5c7f826a2816bd5bf8db194932d2b3ec9ff4c Parents: 020a5b3 Author: Qingwen Zhao <qingwen...@gmail.com> Authored: Thu Sep 22 15:59:49 2016 +0800 Committer: Zhao, Qingwen <qingwz...@ebay.com> Committed: Thu Sep 22 15:59:49 2016 +0800 ---------------------------------------------------------------------- .../eagle/service/app/TestServiceAppWithZk.java | 8 +- .../client/impl/EagleServiceBaseClient.java | 2 +- .../client/impl/EagleServiceClientImpl.java | 4 + .../eagle/service/jpm/MRJobCountHelper.java | 140 ----------- .../service/jpm/MRJobExecutionResource.java | 133 +++------- .../service/jpm/MRJobTaskCountResponse.java | 12 +- .../service/jpm/MRTaskExecutionResource.java | 241 +++++++++++++++++++ .../service/jpm/MRTaskExecutionResponse.java | 85 +++++++ .../apache/eagle/service/jpm/ResourceUtils.java | 87 +++++++ .../service/jpm/TaskCountByDurationHelper.java | 106 -------- .../eagle/service/jpm/count/MRJobCountImpl.java | 146 +++++++++++ .../service/jpm/count/MRTaskCountImpl.java | 124 ++++++++++ .../service/jpm/suggestion/AbstractGCFunc.java | 104 ++++++++ .../jpm/suggestion/AbstractInputFunc.java | 86 +++++++ .../eagle/service/jpm/suggestion/MapGCFunc.java | 38 +++ .../service/jpm/suggestion/MapInputFunc.java | 40 +++ .../service/jpm/suggestion/MapSpillFunc.java | 90 +++++++ .../service/jpm/suggestion/ReduceGCFunc.java | 38 +++ .../service/jpm/suggestion/ReduceInputFunc.java | 39 +++ .../service/jpm/suggestion/SuggestionFunc.java | 27 +++ .../jpm/TestJobCountPerBucketHelper.java | 90 ------- .../service/jpm/TestTaskCountPerJobHelper.java | 96 -------- .../service/jpm/count/TestMRJobCountImpl.java | 92 +++++++ .../service/jpm/count/TestMRTaskCountImpl.java | 129 ++++++++++ .../jpm/suggestion/TestDataSkewFunc.java | 89 +++++++ .../jpm/suggestion/TestTaskCounterFunc.java | 88 +++++++ .../spark/history/SparkHistoryJobAppConfig.java | 8 +- .../history/crawl/JHFSparkEventReader.java | 1 + .../jpm/spark/history/crawl/JHFSparkParser.java | 7 +- .../status/JobHistoryZKStateManager.java | 8 +- .../history/storm/SparkHistoryJobParseBolt.java | 4 +- .../history/storm/SparkHistoryJobSpout.java | 17 +- ...spark.history.SparkHistoryJobAppProvider.xml | 6 - .../src/main/resources/application.conf | 3 +- .../org/apache/eagle/jpm/util/Constants.java | 4 + .../eagle/jpm/util/jobcounter/JobCounters.java | 78 ++++++ eagle-server/pom.xml | 11 +- 37 files changed, 1701 insertions(+), 580 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-core/eagle-alert-parent/eagle-alert/alert-service/src/test/java/com/apache/eagle/service/app/TestServiceAppWithZk.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-alert-parent/eagle-alert/alert-service/src/test/java/com/apache/eagle/service/app/TestServiceAppWithZk.java b/eagle-core/eagle-alert-parent/eagle-alert/alert-service/src/test/java/com/apache/eagle/service/app/TestServiceAppWithZk.java index a19889b..f7f6c57 100644 --- a/eagle-core/eagle-alert-parent/eagle-alert/alert-service/src/test/java/com/apache/eagle/service/app/TestServiceAppWithZk.java +++ b/eagle-core/eagle-alert-parent/eagle-alert/alert-service/src/test/java/com/apache/eagle/service/app/TestServiceAppWithZk.java @@ -31,18 +31,12 @@ import org.apache.eagle.alert.service.IMetadataServiceClient; import org.apache.eagle.alert.service.MetadataServiceClientImpl; import org.apache.eagle.alert.utils.ZookeeperEmbedded; import org.apache.eagle.service.app.ServiceApp; -import org.junit.After; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Test; +import org.junit.*; import com.google.common.base.Joiner; import com.typesafe.config.Config; import com.typesafe.config.ConfigFactory; -/** - * @author xiancli - */ public class TestServiceAppWithZk { ZookeeperEmbedded zkEmbed; http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-core/eagle-query/eagle-client-base/src/main/java/org/apache/eagle/service/client/impl/EagleServiceBaseClient.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-client-base/src/main/java/org/apache/eagle/service/client/impl/EagleServiceBaseClient.java b/eagle-core/eagle-query/eagle-client-base/src/main/java/org/apache/eagle/service/client/impl/EagleServiceBaseClient.java index ea5d9a5..3b717d8 100644 --- a/eagle-core/eagle-query/eagle-client-base/src/main/java/org/apache/eagle/service/client/impl/EagleServiceBaseClient.java +++ b/eagle-core/eagle-query/eagle-client-base/src/main/java/org/apache/eagle/service/client/impl/EagleServiceBaseClient.java @@ -67,7 +67,7 @@ public abstract class EagleServiceBaseClient implements IEagleServiceClient { private final static Logger LOG = LoggerFactory.getLogger(EagleServiceBaseClient.class); - protected static final String DEFAULT_BASE_PATH = "/rest"; + public static final String DEFAULT_BASE_PATH = "/rest"; protected static final MediaType DEFAULT_MEDIA_TYPE = MediaType.APPLICATION_JSON_TYPE; protected static final String DEFAULT_HTTP_HEADER_CONTENT_TYPE = "application/json"; protected static final String CONTENT_TYPE = "Content-Type"; http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-core/eagle-query/eagle-client-base/src/main/java/org/apache/eagle/service/client/impl/EagleServiceClientImpl.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-query/eagle-client-base/src/main/java/org/apache/eagle/service/client/impl/EagleServiceClientImpl.java b/eagle-core/eagle-query/eagle-client-base/src/main/java/org/apache/eagle/service/client/impl/EagleServiceClientImpl.java index 6018469..912f1f7 100644 --- a/eagle-core/eagle-query/eagle-client-base/src/main/java/org/apache/eagle/service/client/impl/EagleServiceClientImpl.java +++ b/eagle-core/eagle-query/eagle-client-base/src/main/java/org/apache/eagle/service/client/impl/EagleServiceClientImpl.java @@ -45,6 +45,10 @@ public class EagleServiceClientImpl extends EagleServiceBaseClient { super(host, port, username, password); } + public EagleServiceClientImpl(String host, int port, String basePath, String username, String password){ + super(host, port, basePath, username, password); + } + private String getWholePath(String urlString){ return getBaseEndpoint() + urlString; } http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobCountHelper.java ---------------------------------------------------------------------- diff --git a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobCountHelper.java b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobCountHelper.java deleted file mode 100644 index 2fa5c04..0000000 --- a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobCountHelper.java +++ /dev/null @@ -1,140 +0,0 @@ -/* - * 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.eagle.service.jpm; - -import org.apache.eagle.common.DateTimeUtil; -import org.apache.eagle.jpm.mr.historyentity.JobExecutionAPIEntity; -import org.apache.eagle.jpm.util.Constants; -import org.apache.eagle.jpm.util.MRJobTagName; -import org.apache.eagle.service.jpm.MRJobTaskCountResponse.JobCountResponse; -import org.apache.eagle.service.jpm.MRJobTaskCountResponse.UnitJobCount; - -import java.text.ParseException; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.List; -import java.util.Set; - -public class MRJobCountHelper { - - public void initJobCountList(List<UnitJobCount> jobCounts, long startTime, long endTime, long intervalInSecs) { - for (long i = startTime / intervalInSecs; i * intervalInSecs <= endTime; i++) { - jobCounts.add(new UnitJobCount(i * intervalInSecs * DateTimeUtil.ONESECOND)); - } - } - - public String moveTimeForwardOneDay(String startTime) throws ParseException { - long timeInSecs = DateTimeUtil.humanDateToSeconds(startTime); - timeInSecs -= 24L * 60L * 60L; - return DateTimeUtil.secondsToHumanDate(timeInSecs); - } - - public JobCountResponse getRunningJobCount(List<JobExecutionAPIEntity> jobDurations, - List<org.apache.eagle.jpm.mr.runningentity.JobExecutionAPIEntity> runningJobs, - long startTimeInSecs, - long endTimeInSecs, - long intervalInSecs) { - List<UnitJobCount> jobCounts = new ArrayList<>(); - Set<String> jobTypes = new HashSet<>(); - initJobCountList(jobCounts, startTimeInSecs, endTimeInSecs, intervalInSecs); - for (JobExecutionAPIEntity jobDuration: jobDurations) { - String jobType = jobDuration.getTags().get(MRJobTagName.JOB_TYPE.toString()); - jobTypes.add(jobType); - countJob(jobCounts, jobDuration.getStartTime() / 1000, jobDuration.getEndTime() / 1000, intervalInSecs, jobType); - } - for (org.apache.eagle.jpm.mr.runningentity.JobExecutionAPIEntity job : runningJobs) { - if (job.getInternalState() != null && !job.getInternalState().equalsIgnoreCase(Constants.JobState.FINISHED.toString())) { - String jobType = job.getTags().get(MRJobTagName.JOB_TYPE.toString()); - jobTypes.add(jobType); - countJob(jobCounts, job.getStartTime() / 1000, endTimeInSecs, intervalInSecs, jobType); - } - } - JobCountResponse response = new JobCountResponse(); - response.jobCounts = jobCounts; - response.jobTypes = jobTypes; - return response; - } - - public JobCountResponse getHistoryJobCount(List<JobExecutionAPIEntity> jobDurations, String timeList) { - JobCountResponse response = new JobCountResponse(); - List<UnitJobCount> jobCounts = new ArrayList<>(); - Set<String> jobTypes = new HashSet<>(); - List<Long> times = TaskCountByDurationHelper.parseTimeList(timeList); - for (int i = 0; i < times.size(); i++) { - jobCounts.add(new UnitJobCount(times.get(i))); - } - for (JobExecutionAPIEntity job : jobDurations) { - int jobIndex = TaskCountByDurationHelper.getPosition(times, job.getDurationTime()); - UnitJobCount counter = jobCounts.get(jobIndex); - String jobType = job.getTags().get(MRJobTagName.JOB_TYPE.toString()); - jobTypes.add(jobType); - countJob(counter, jobType); - } - response.jobCounts = jobCounts; - response.jobTypes = jobTypes; - return response; - } - - public void countJob(UnitJobCount counter, String jobType) { - if (null == jobType) { - jobType = "null"; - } - counter.jobCount++; - if (counter.jobCountByType.containsKey(jobType)) { - counter.jobCountByType.put(jobType, counter.jobCountByType.get(jobType) + 1); - } else { - counter.jobCountByType.put(jobType, 1L); - } - } - - public void countJob(List<UnitJobCount> jobCounts, long jobStartTimeSecs, long jobEndTimeSecs, long intervalInSecs, String jobType) { - long startCountPoint = jobCounts.get(0).timeBucket / DateTimeUtil.ONESECOND; - if (jobEndTimeSecs < startCountPoint) { - return; - } - int startIndex = 0; - if (jobStartTimeSecs > startCountPoint) { - long relativeStartTime = jobStartTimeSecs - startCountPoint; - startIndex = (int) (relativeStartTime / intervalInSecs) + (relativeStartTime % intervalInSecs == 0 ? 0 : 1); - } - long relativeEndTime = jobEndTimeSecs - startCountPoint; - int endIndex = (int) (relativeEndTime / intervalInSecs); - - for (int i = startIndex; i <= endIndex && i < jobCounts.size(); i++) { - countJob(jobCounts.get(i), jobType); - } - } - - public List<String> getSearchTimeDuration(List<JobExecutionAPIEntity> jobEntities) { - List<String> pair = new ArrayList<>(); - long minStartTime = System.currentTimeMillis(); - long maxEndTime = 0; - for (JobExecutionAPIEntity jobEntity : jobEntities) { - if (minStartTime > jobEntity.getStartTime()) { - minStartTime = jobEntity.getStartTime(); - } - if (maxEndTime < jobEntity.getEndTime()) { - maxEndTime = jobEntity.getEndTime(); - } - } - pair.add(DateTimeUtil.millisecondsToHumanDateWithSeconds(minStartTime)); - pair.add(DateTimeUtil.millisecondsToHumanDateWithSeconds(maxEndTime)); - return pair; - } -} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobExecutionResource.java ---------------------------------------------------------------------- diff --git a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobExecutionResource.java b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobExecutionResource.java index e6041f2..204412b 100644 --- a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobExecutionResource.java +++ b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobExecutionResource.java @@ -18,21 +18,18 @@ package org.apache.eagle.service.jpm; -import static org.apache.eagle.jpm.util.MRJobTagName.JOB_ID; -import static org.apache.eagle.jpm.util.MRJobTagName.TASK_TYPE; - import org.apache.eagle.common.DateTimeUtil; import org.apache.eagle.jpm.mr.historyentity.JobExecutionAPIEntity; -import org.apache.eagle.jpm.mr.runningentity.TaskExecutionAPIEntity; import org.apache.eagle.jpm.util.Constants; +import org.apache.eagle.jpm.util.MRJobTagName; import org.apache.eagle.log.base.taggedlog.TaggedLogAPIEntity; import org.apache.eagle.log.entity.GenericServiceAPIResponseEntity; import org.apache.eagle.service.generic.GenericEntityServiceResource; import org.apache.eagle.service.generic.ListQueryResource; import org.apache.eagle.service.jpm.MRJobTaskCountResponse.JobCountResponse; -import org.apache.eagle.service.jpm.MRJobTaskCountResponse.TaskCountPerJobResponse; import org.apache.commons.lang.time.StopWatch; +import org.apache.eagle.service.jpm.count.MRJobCountImpl; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -43,11 +40,14 @@ import javax.ws.rs.core.MediaType; @Path("mrJobs") public class MRJobExecutionResource { - GenericEntityServiceResource resource = new GenericEntityServiceResource(); + + private static final Logger LOG = LoggerFactory.getLogger(MRJobExecutionResource.class); + public static final String ELAPSEDMS = "elapsedms"; public static final String TOTAL_RESULTS = "totalResults"; - private static final Logger LOG = LoggerFactory.getLogger(MRJobExecutionResource.class); + private final MRJobCountImpl helper = new MRJobCountImpl(); + private GenericEntityServiceResource resource = new GenericEntityServiceResource(); @GET @Produces(MediaType.APPLICATION_JSON) @@ -76,14 +76,15 @@ public class MRJobExecutionResource { if (res.isSuccess() && res.getObj() != null) { for (TaggedLogAPIEntity o : res.getObj()) { finishedJobs.add(o); - jobIds.add(o.getTags().get(JOB_ID.toString())); + jobIds.add(o.getTags().get(MRJobTagName.JOB_ID.toString())); } jobQuery = String.format(query, Constants.JPA_RUNNING_JOB_EXECUTION_SERVICE_NAME); res = resource.search(jobQuery, startTime, endTime, pageSize, startRowkey, treeAgg, timeSeries, intervalmin, top, filterIfMissing, parallel, metricName, verbose); if (res.isSuccess() && res.getObj() != null) { for (TaggedLogAPIEntity o : res.getObj()) { - if (!isDuplicate(jobIds, o)) { + String key = o.getTags().get(MRJobTagName.JOB_ID.toString()); + if (!ResourceUtils.isDuplicate(jobIds, key)) { jobs.add(o); } } @@ -105,13 +106,6 @@ public class MRJobExecutionResource { } - private boolean isDuplicate(Set<String> keys, TaggedLogAPIEntity o) { - if (keys.isEmpty()) { - return false; - } - return keys.contains(o.getTags().get(JOB_ID.toString())); - } - private String buildCondition(String jobId, String jobDefId, String site) { String conditionFormat = "@site=\"%s\""; String condition = null; @@ -142,7 +136,7 @@ public class MRJobExecutionResource { List<TaggedLogAPIEntity> jobs = new ArrayList<>(); Set<String> jobIds = new HashSet<>(); String condition = buildCondition(jobId, jobDefId, site); - final int pageSize = Integer.MAX_VALUE; + if (condition == null) { response.setException(new Exception("Search condition is empty")); response.setSuccess(false); @@ -155,18 +149,19 @@ public class MRJobExecutionResource { stopWatch.start(); String queryFormat = "%s[%s]{*}"; String queryString = String.format(queryFormat, Constants.JPA_JOB_EXECUTION_SERVICE_NAME, condition); - GenericServiceAPIResponseEntity<TaggedLogAPIEntity> res = resource.search(queryString, null, null, pageSize, null, false, true, 0L, 0, true, 0, null, false); + GenericServiceAPIResponseEntity<TaggedLogAPIEntity> res = ResourceUtils.getQueryResult(queryString, null, null); if (res.isSuccess() && res.getObj() != null) { for (TaggedLogAPIEntity o : res.getObj()) { jobs.add(o); - jobIds.add(o.getTags().get(JOB_ID.toString())); + jobIds.add(o.getTags().get(MRJobTagName.JOB_ID.toString())); } } queryString = String.format(queryFormat, Constants.JPA_RUNNING_JOB_EXECUTION_SERVICE_NAME, condition); - res = resource.search(queryString, null, null, pageSize, null, false, true, 0L, 0, true, 0, null, false); + res = ResourceUtils.getQueryResult(queryString, null, null); if (res.isSuccess() && res.getObj() != null) { for (TaggedLogAPIEntity o : res.getObj()) { - if (!isDuplicate(jobIds, o)) { + String key = o.getTags().get(MRJobTagName.JOB_ID.toString()); + if (!ResourceUtils.isDuplicate(jobIds, key)) { jobs.add(o); } } @@ -193,67 +188,6 @@ public class MRJobExecutionResource { return response; } - - - - @GET - @Path("{jobId}/taskCountsByDuration") - @Produces(MediaType.APPLICATION_JSON) - public TaskCountPerJobResponse getTaskCountsPerJob(@PathParam("jobId") String jobId, - @QueryParam("site") String site, - @QueryParam("timelineInSecs") String timeList, - @QueryParam("top") long top) { - TaskCountPerJobResponse response = new TaskCountPerJobResponse(); - if (jobId == null || site == null || timeList == null || timeList.isEmpty()) { - response.errMessage = "IllegalArgumentException: jobId == null || site == null || timelineInSecs == null or isEmpty"; - return response; - } - TaskCountByDurationHelper helper = new TaskCountByDurationHelper(); - List<MRJobTaskCountResponse.UnitTaskCount> runningTaskCount = new ArrayList<>(); - List<MRJobTaskCountResponse.UnitTaskCount> finishedTaskCount = new ArrayList<>(); - - List<Long> times = helper.parseTimeList(timeList); - String query = String.format("%s[@site=\"%s\" AND @jobId=\"%s\"]{*}", Constants.JPA_TASK_EXECUTION_SERVICE_NAME, site, jobId); - GenericServiceAPIResponseEntity<org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity> historyRes = - resource.search(query, null, null, Integer.MAX_VALUE, null, false, true, 0L, 0, true, 0, null, false); - if (historyRes.isSuccess() && historyRes.getObj() != null && historyRes.getObj().size() > 0) { - helper.initTaskCountList(runningTaskCount, finishedTaskCount, times, new TaskCountByDurationHelper.HistoryTaskComparator()); - for (org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity o : historyRes.getObj()) { - int index = helper.getPosition(times, o.getDuration()); - MRJobTaskCountResponse.UnitTaskCount counter = finishedTaskCount.get(index); - helper.countTask(counter, o.getTags().get(TASK_TYPE.toString())); - counter.entities.add(o); - } - } else { - query = String.format("%s[@site=\"%s\" AND @jobId=\"%s\"]{*}", Constants.JPA_RUNNING_TASK_EXECUTION_SERVICE_NAME, site, jobId); - GenericServiceAPIResponseEntity<TaskExecutionAPIEntity> runningRes = - resource.search(query, null, null, Integer.MAX_VALUE, null, false, true, 0L, 0, true, 0, null, false); - if (runningRes.isSuccess() && runningRes.getObj() != null) { - helper.initTaskCountList(runningTaskCount, finishedTaskCount, times, new TaskCountByDurationHelper.RunningTaskComparator()); - for (TaskExecutionAPIEntity o : runningRes.getObj()) { - int index = helper.getPosition(times, o.getDuration()); - if (o.getTaskStatus().equalsIgnoreCase(Constants.TaskState.RUNNING.toString())) { - MRJobTaskCountResponse.UnitTaskCount counter = runningTaskCount.get(index); - helper.countTask(counter, o.getTags().get(TASK_TYPE.toString())); - counter.entities.add(o); - } else if (o.getEndTime() != 0) { - MRJobTaskCountResponse.UnitTaskCount counter = finishedTaskCount.get(index); - helper.countTask(counter, o.getTags().get(TASK_TYPE.toString())); - counter.entities.add(o); - } - } - } - } - if (top > 0) { - helper.getTopTasks(runningTaskCount, top); - response.runningTaskCount = runningTaskCount; - helper.getTopTasks(finishedTaskCount, top); - response.finishedTaskCount = finishedTaskCount; - } - response.topNumber = top; - return response; - } - @GET @Path("runningJobCounts") @Produces(MediaType.APPLICATION_JSON) @@ -262,7 +196,6 @@ public class MRJobExecutionResource { @QueryParam("durationEnd") String endTime, @QueryParam("intervalInSecs") long intervalInSecs) { JobCountResponse response = new JobCountResponse(); - MRJobCountHelper helper = new MRJobCountHelper(); if (site == null || startTime == null || endTime == null) { response.errMessage = "IllegalArgument: site, durationBegin, durationEnd is null"; return response; @@ -271,26 +204,22 @@ public class MRJobExecutionResource { response.errMessage = String.format("IllegalArgument: intervalInSecs=%s is invalid", intervalInSecs); return response; } - long startTimeInMills; String searchStartTime = startTime; String searchEndTime = endTime; try { - startTimeInMills = DateTimeUtil.humanDateToSeconds(startTime) * DateTimeUtil.ONESECOND; searchStartTime = helper.moveTimeForwardOneDay(searchStartTime); } catch (Exception e) { response.errMessage = e.getMessage(); return response; } - String query = String.format("%s[@site=\"%s\" AND @endTime>=%s]{@startTime,@endTime,@jobType}", Constants.JPA_JOB_EXECUTION_SERVICE_NAME, site, startTimeInMills); - GenericServiceAPIResponseEntity<JobExecutionAPIEntity> historyRes = - resource.search(query, searchStartTime, searchEndTime, Integer.MAX_VALUE, null, false, true, 0L, 0, true, 0, null, false); + String query = String.format("%s[@site=\"%s\"]{@startTime,@endTime,@jobType,@jobId}", Constants.JPA_JOB_EXECUTION_SERVICE_NAME, site); + GenericServiceAPIResponseEntity<JobExecutionAPIEntity> historyRes = ResourceUtils.getQueryResult(query, searchStartTime, searchEndTime); if (!historyRes.isSuccess() || historyRes.getObj() == null) { response.errMessage = String.format("Catch an exception during fetch history jobs: %s with query=%s", historyRes.getException(), query); return response; } - query = String.format("%s[@site=\"%s\"]{@startTime,@endTime,@jobType}", Constants.JPA_RUNNING_JOB_EXECUTION_SERVICE_NAME, site); - GenericServiceAPIResponseEntity<org.apache.eagle.jpm.mr.runningentity.JobExecutionAPIEntity> runningRes = - resource.search(query, searchStartTime, searchEndTime, Integer.MAX_VALUE, null, false, true, 0L, 0, true, 0, null, false); + query = String.format("%s[@site=\"%s\"]{@startTime,@endTime,@jobType,@jobId}", Constants.JPA_RUNNING_JOB_EXECUTION_SERVICE_NAME, site); + GenericServiceAPIResponseEntity<org.apache.eagle.jpm.mr.runningentity.JobExecutionAPIEntity> runningRes = ResourceUtils.getQueryResult(query, searchStartTime, searchEndTime); if (!runningRes.isSuccess() || runningRes.getObj() == null) { response.errMessage = String.format("Catch an exception during fetch running jobs: %s with query=%s", runningRes.getException(), query); return response; @@ -332,7 +261,6 @@ public class MRJobExecutionResource { public Object getJobMetrics(String site, String timePoint, String metricName, long intervalmin, int top, Function6<String, String, String, Long, Integer, String, Object> metricQueryFunc) { GenericServiceAPIResponseEntity response = new GenericServiceAPIResponseEntity(); - MRJobCountHelper helper = new MRJobCountHelper(); if (site == null || timePoint == null || metricName == null) { response.setException(new IllegalArgumentException("Error: site, timePoint, metricName may be unset")); response.setSuccess(false); @@ -360,8 +288,7 @@ public class MRJobExecutionResource { } String query = String.format("%s[@site=\"%s\" AND @startTime<=\"%s\" AND @endTime>=\"%s\"]{@startTime,@endTime}", Constants.JPA_JOB_EXECUTION_SERVICE_NAME, site, timePointsInMills, timePointsInMills); - GenericServiceAPIResponseEntity<JobExecutionAPIEntity> historyRes = - resource.search(query, searchStartTime, searchEndTime, Integer.MAX_VALUE, null, false, true, 0L, 0, true, 0, null, false); + GenericServiceAPIResponseEntity<JobExecutionAPIEntity> historyRes = ResourceUtils.getQueryResult(query, searchStartTime, searchEndTime); if (!historyRes.isSuccess() || historyRes.getObj() == null) { return historyRes; } @@ -372,14 +299,14 @@ public class MRJobExecutionResource { return metricQueryFunc.apply(query, timeDuration.get(0), timeDuration.get(1), intervalmin, top, metricName); } - Function6<String, String, String, Long, Integer, String, Object> queryMetricEntitiesFunc + private Function6<String, String, String, Long, Integer, String, Object> queryMetricEntitiesFunc = (query, startTime, endTime, intervalmin, top, metricName) -> { GenericEntityServiceResource resource = new GenericEntityServiceResource(); return resource.search(query, startTime, endTime, Integer.MAX_VALUE, null, false, true, intervalmin, top, true, 0, metricName, false); }; - Function6<String, String, String, Long, Integer, String, Object> queryMetricListFunc + private Function6<String, String, String, Long, Integer, String, Object> queryMetricListFunc = (query, startTime, endTime, intervalmin, top, metricName) -> { ListQueryResource resource = new ListQueryResource(); return resource.listQuery(query, startTime, endTime, Integer.MAX_VALUE, null, @@ -395,24 +322,22 @@ public class MRJobExecutionResource { @Path("jobCountsByDuration") @Produces(MediaType.APPLICATION_JSON) public JobCountResponse getJobCountGroupByDuration(@QueryParam("site") String site, - @QueryParam("timelineInSecs") String timeList, - @QueryParam("jobStartTimeBegin") String startTime, - @QueryParam("jobStartTimeEnd") String endTime) { + @QueryParam("timeDistInSecs") String timeList, + @QueryParam("startTime") String startTime, + @QueryParam("endTime") String endTime) { JobCountResponse response = new JobCountResponse(); - MRJobCountHelper helper = new MRJobCountHelper(); if (site == null || startTime == null || endTime == null || timeList == null) { - response.errMessage = "IllegalArgument: site, jobStartTimeBegin, jobStartTimeEnd, or timelineInSecs is null"; + response.errMessage = "IllegalArgument: site, startTime, endTime, or timeDistInSecs is null"; return response; } String query = String.format("%s[@site=\"%s\"]{@durationTime,@jobType}", Constants.JPA_JOB_EXECUTION_SERVICE_NAME, site); - GenericServiceAPIResponseEntity<JobExecutionAPIEntity> historyRes = - resource.search(query, startTime, endTime, Integer.MAX_VALUE, null, false, true, 0L, 0, true, 0, null, false); + GenericServiceAPIResponseEntity<JobExecutionAPIEntity> historyRes = ResourceUtils.getQueryResult(query, startTime, endTime); if (!historyRes.isSuccess() || historyRes.getObj() == null) { response.errMessage = String.format("Catch an exception: %s with query=%s", historyRes.getException(), query); return response; } try { - return helper.getHistoryJobCount(historyRes.getObj(), timeList); + return helper.getHistoryJobCountGroupByDuration(historyRes.getObj(), timeList); } catch (Exception e) { response.errMessage = e.getMessage(); return response; http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobTaskCountResponse.java ---------------------------------------------------------------------- diff --git a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobTaskCountResponse.java b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobTaskCountResponse.java index 170533c..45ffbd7 100644 --- a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobTaskCountResponse.java +++ b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRJobTaskCountResponse.java @@ -29,12 +29,16 @@ public class MRJobTaskCountResponse { public List<UnitTaskCount> finishedTaskCount; } + public static class HistoryTaskCountResponse extends MRJobTaskCountResponse { + public List<UnitTaskCount> taskCount; + } + public static class JobCountResponse extends MRJobTaskCountResponse { public Set<String> jobTypes; public List<UnitJobCount> jobCounts; } - static class UnitTaskCount { + public static class UnitTaskCount { public long timeBucket; public int taskCount; public int mapTaskCount; @@ -42,7 +46,7 @@ public class MRJobTaskCountResponse { public Set entities; public List topEntities; - UnitTaskCount(long timeBucket, Comparator comparator) { + public UnitTaskCount(long timeBucket, Comparator comparator) { this.timeBucket = timeBucket; this.taskCount = 0; this.mapTaskCount = 0; @@ -52,12 +56,12 @@ public class MRJobTaskCountResponse { } } - static class UnitJobCount { + public static class UnitJobCount { public long timeBucket; public long jobCount; public Map<String, Long> jobCountByType; - UnitJobCount(long timeBucket) { + public UnitJobCount(long timeBucket) { this.timeBucket = timeBucket; this.jobCount = 0; this.jobCountByType = new HashMap<>(); http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRTaskExecutionResource.java ---------------------------------------------------------------------- diff --git a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRTaskExecutionResource.java b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRTaskExecutionResource.java new file mode 100644 index 0000000..1125387 --- /dev/null +++ b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRTaskExecutionResource.java @@ -0,0 +1,241 @@ +/* + * 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.eagle.service.jpm; + +import org.apache.eagle.common.DateTimeUtil; +import org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity; +import org.apache.eagle.jpm.util.Constants; +import org.apache.eagle.jpm.util.MRJobTagName; +import org.apache.eagle.log.entity.GenericServiceAPIResponseEntity; + +import org.apache.eagle.service.jpm.count.MRTaskCountImpl; +import org.apache.eagle.service.jpm.suggestion.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.ws.rs.*; +import javax.ws.rs.core.MediaType; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; + +import static org.apache.eagle.jpm.util.MRJobTagName.TASK_TYPE; + +@Path("mrTasks") +public class MRTaskExecutionResource { + private static final Logger LOG = LoggerFactory.getLogger(MRTaskExecutionResource.class); + MRTaskCountImpl taskCountImpl = new MRTaskCountImpl(); + + @GET + @Path("taskCountsByDuration") + @Produces(MediaType.APPLICATION_JSON) + public MRJobTaskCountResponse.TaskCountPerJobResponse getTaskCountsGroupByDuration(@QueryParam("site") String site, + @QueryParam("jobId") String jobId, + @QueryParam("jobStartTime") String jobStartTime, + @QueryParam("jobEndTime") String jobEndTime, + @QueryParam("timeDistInSecs") String timeDistInSecs, + @QueryParam("top") long top) { + MRJobTaskCountResponse.TaskCountPerJobResponse response = new MRJobTaskCountResponse.TaskCountPerJobResponse(); + if (jobId == null || site == null || timeDistInSecs == null || timeDistInSecs.isEmpty()) { + response.errMessage = "IllegalArgumentException: jobId == null || site == null || timeDistInSecs == null or isEmpty"; + return response; + } + List<MRJobTaskCountResponse.UnitTaskCount> runningTaskCount = new ArrayList<>(); + List<MRJobTaskCountResponse.UnitTaskCount> finishedTaskCount = new ArrayList<>(); + + List<Long> times = ResourceUtils.parseDistributionList(timeDistInSecs); + String query = String.format("%s[@site=\"%s\" AND @jobId=\"%s\"]{*}", Constants.JPA_TASK_EXECUTION_SERVICE_NAME, site, jobId); + GenericServiceAPIResponseEntity<org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity> historyRes = + ResourceUtils.getQueryResult(query, jobStartTime, jobEndTime); + if (historyRes.isSuccess() && historyRes.getObj() != null && historyRes.getObj().size() > 0) { + taskCountImpl.initTaskCountList(runningTaskCount, finishedTaskCount, times, new MRTaskCountImpl.HistoryTaskComparator()); + for (org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity o : historyRes.getObj()) { + int index = ResourceUtils.getDistributionPosition(times, o.getDuration() / DateTimeUtil.ONESECOND); + MRJobTaskCountResponse.UnitTaskCount counter = finishedTaskCount.get(index); + taskCountImpl.countTask(counter, o.getTags().get(TASK_TYPE.toString())); + counter.entities.add(o); + } + } else { + query = String.format("%s[@site=\"%s\" AND @jobId=\"%s\"]{*}", Constants.JPA_RUNNING_TASK_EXECUTION_SERVICE_NAME, site, jobId); + GenericServiceAPIResponseEntity<org.apache.eagle.jpm.mr.runningentity.TaskExecutionAPIEntity> runningRes = + ResourceUtils.getQueryResult(query, jobStartTime, jobEndTime); + if (runningRes.isSuccess() && runningRes.getObj() != null) { + taskCountImpl.initTaskCountList(runningTaskCount, finishedTaskCount, times, new MRTaskCountImpl.RunningTaskComparator()); + for (org.apache.eagle.jpm.mr.runningentity.TaskExecutionAPIEntity o : runningRes.getObj()) { + int index = ResourceUtils.getDistributionPosition(times, o.getDuration() / DateTimeUtil.ONESECOND); + if (o.getTaskStatus().equalsIgnoreCase(Constants.TaskState.RUNNING.toString())) { + MRJobTaskCountResponse.UnitTaskCount counter = runningTaskCount.get(index); + taskCountImpl.countTask(counter, o.getTags().get(TASK_TYPE.toString())); + counter.entities.add(o); + } else if (o.getEndTime() != 0) { + MRJobTaskCountResponse.UnitTaskCount counter = finishedTaskCount.get(index); + taskCountImpl.countTask(counter, o.getTags().get(TASK_TYPE.toString())); + counter.entities.add(o); + } + } + } + } + if (top > 0) { + taskCountImpl.getTopTasks(runningTaskCount, top); + response.runningTaskCount = runningTaskCount; + taskCountImpl.getTopTasks(finishedTaskCount, top); + response.finishedTaskCount = finishedTaskCount; + } + response.topNumber = top; + return response; + } + + private MRTaskExecutionResponse.TaskGroupResponse getTaskGroups(@QueryParam("site") String site, + @QueryParam("shortJob_id") String shortDurationJobId, + @QueryParam("longJob_id") String longDurationJobId) { + MRTaskExecutionResponse.TaskGroupResponse result = new MRTaskExecutionResponse.TaskGroupResponse(); + String query = String.format("%s[@site=\"%s\" AND @jobId=\"%s\"]{*}", Constants.JPA_TASK_EXECUTION_SERVICE_NAME, site, shortDurationJobId); + GenericServiceAPIResponseEntity<TaskExecutionAPIEntity> smallResponse = ResourceUtils.getQueryResult(query, null, null); + if (!smallResponse.isSuccess() || smallResponse.getObj() == null) { + result.errMessage = smallResponse.getException(); + return result; + } + long longestDuration = 0; + for (TaskExecutionAPIEntity entity : smallResponse.getObj()) { + if (entity.getDuration() > longestDuration) { + longestDuration = entity.getDuration(); + } + } + query = String.format("%s[@site=\"%s\" AND @jobId=\"%s\"]{*}", Constants.JPA_TASK_EXECUTION_SERVICE_NAME, site, longDurationJobId); + GenericServiceAPIResponseEntity<TaskExecutionAPIEntity> largeResponse = ResourceUtils.getQueryResult(query, null, null); + if (!largeResponse.isSuccess() || largeResponse.getObj() == null) { + result.errMessage = largeResponse.getException(); + return result; + } + result.tasksGroupByType = new HashMap<>(); + result.tasksGroupByType.put(Constants.TaskType.MAP.toString(), new MRTaskExecutionResponse.TaskGroup()); + result.tasksGroupByType.put(Constants.TaskType.REDUCE.toString(), new MRTaskExecutionResponse.TaskGroup()); + groupTasksByValue(result, false, largeResponse.getObj(), longestDuration); + groupTasksByValue(result, true, smallResponse.getObj(), longestDuration); + + return result; + } + + public MRTaskExecutionResponse.TaskGroupResponse groupTasksByValue(MRTaskExecutionResponse.TaskGroupResponse result, boolean keepShort, List<TaskExecutionAPIEntity> tasks, long value) { + for (TaskExecutionAPIEntity entity : tasks) { + String taskType = entity.getTags().get(MRJobTagName.TASK_TYPE.toString()); + MRTaskExecutionResponse.TaskGroup taskGroup = result.tasksGroupByType.get(taskType.toUpperCase()); + if (entity.getDuration() <= value && keepShort) { + taskGroup.shortTasks.add(entity); + } + if (entity.getDuration() > value) { + taskGroup.longTasks.add(entity); + } + } + return result; + } + + @GET + @Produces(MediaType.APPLICATION_JSON) + @Path("taskSuggestion") + public List<MRTaskExecutionResponse.JobSuggestionResponse> getSuggestion(@QueryParam("site") String site, + @QueryParam("shortJob_id") String shortDurationJobId, + @QueryParam("longJob_id") String longDurationJobId, + @QueryParam("mapInputThreshold") long mapInputThreshold, + @QueryParam("reduceInputThreshold") long reduceInputThreshold, + @QueryParam("mapGcThreshold") long mapGcThreshold, + @QueryParam("reduceGcThreshold") long reduceGcThreshold, + @QueryParam("mapSpillThreshold") long mapSpillThreshold) { + List<MRTaskExecutionResponse.JobSuggestionResponse> result = new ArrayList<>(); + MRTaskExecutionResponse.TaskGroupResponse taskGroups = getTaskGroups(site, shortDurationJobId, longDurationJobId); + if (taskGroups.errMessage != null) { + LOG.error(taskGroups.errMessage); + return result; + } + List<SuggestionFunc> suggestionFuncs = new ArrayList<>(); + suggestionFuncs.add(new MapInputFunc(mapInputThreshold)); + suggestionFuncs.add(new ReduceInputFunc(reduceInputThreshold)); + suggestionFuncs.add(new MapGCFunc(mapGcThreshold)); + suggestionFuncs.add(new ReduceGCFunc(reduceGcThreshold)); + suggestionFuncs.add(new MapSpillFunc(mapSpillThreshold)); + try { + for (SuggestionFunc func : suggestionFuncs) { + result.add(func.apply(taskGroups)); + } + } catch (Exception ex) { + ex.printStackTrace(); + return result; + } + return result; + } + + @GET + @Produces(MediaType.APPLICATION_JSON) + @Path("historyTaskCount") + public MRJobTaskCountResponse.HistoryTaskCountResponse getTaskCountInMinute(@QueryParam("site") String site, + @QueryParam("jobId") String jobId, + @QueryParam("jobStartTime") String jobStartTime, + @QueryParam("jobEndTime") String jobEndTime) { + MRJobTaskCountResponse.HistoryTaskCountResponse result = new MRJobTaskCountResponse.HistoryTaskCountResponse(); + if (jobId == null || site == null || jobStartTime == null || jobEndTime == null) { + result.errMessage = "IllegalArgumentException: jobId, or site, or jobStartTime, or jobEndTime is null"; + return result; + } + + String query = String.format("%s[@site=\"%s\" AND @jobId=\"%s\"]{@startTime,@endTime,@taskType}", Constants.JPA_TASK_EXECUTION_SERVICE_NAME, site, jobId); + GenericServiceAPIResponseEntity<TaskExecutionAPIEntity> response = ResourceUtils.getQueryResult(query, jobStartTime, jobEndTime); + if (!response.isSuccess() || response.getObj() == null) { + result.errMessage = response.getException(); + return result; + } + try { + long startTimeInMin = DateTimeUtil.humanDateToSeconds(jobStartTime) / 60; + long endTimeInMin = DateTimeUtil.humanDateToSeconds(jobEndTime) / 60; + return taskCountImpl.countHistoryTask(response.getObj(), startTimeInMin, endTimeInMin); + } catch (Exception e) { + e.printStackTrace(); + result.errMessage = e.getMessage(); + return result; + } + } + + @GET + @Produces(MediaType.APPLICATION_JSON) + @Path("taskDistribution/{counterName}") + public MRTaskExecutionResponse.TaskDistributionResponse getTaskDistributionByCounterName(@QueryParam("site") String site, + @QueryParam("jobId") String jobId, + @QueryParam("jobStartTime") String jobStartTime, + @QueryParam("jobEndTime") String jobEndTime, + @QueryParam("taskType") String taskType, + @PathParam("counterName") String counterName, + @QueryParam("distRange") String distRange) { + MRTaskExecutionResponse.TaskDistributionResponse result = new MRTaskExecutionResponse.TaskDistributionResponse(); + String query = String.format("%s[@site=\"%s\" AND @jobId=\"%s\" AND @taskType=\"%s\"]{@jobCounters}", Constants.JPA_TASK_EXECUTION_SERVICE_NAME, + site, jobId, Constants.TaskType.MAP.toString()); + GenericServiceAPIResponseEntity<TaskExecutionAPIEntity> response = ResourceUtils.getQueryResult(query, jobStartTime, jobEndTime); + if (!response.isSuccess() || response.getObj() == null) { + result.errMessage = response.getException(); + return result; + } + try { + return taskCountImpl.getHistoryTaskDistribution(response.getObj(), counterName, distRange); + } catch (Exception e) { + e.printStackTrace(); + result.errMessage = e.getMessage(); + return result; + } + } + + +} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRTaskExecutionResponse.java ---------------------------------------------------------------------- diff --git a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRTaskExecutionResponse.java b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRTaskExecutionResponse.java new file mode 100644 index 0000000..c7cc258 --- /dev/null +++ b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/MRTaskExecutionResponse.java @@ -0,0 +1,85 @@ +/* + * 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.eagle.service.jpm; + + +import org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +public class MRTaskExecutionResponse { + public String errMessage; + + public static class TaskGroupResponse extends MRTaskExecutionResponse { + public Map<String, TaskGroup> tasksGroupByType; + } + + public static class TaskGroup { + public List<TaskExecutionAPIEntity> longTasks; + public List<TaskExecutionAPIEntity> shortTasks; + + public TaskGroup() { + longTasks = new ArrayList<>(); + shortTasks = new ArrayList<>(); + } + } + + public static class JobSuggestionResponse extends MRTaskExecutionResponse { + public String suggestionType; + public List<SuggestionResult> suggestionResults; + } + + public static class SuggestionResult { + public String name; + public double value; + public String suggestion; + + public SuggestionResult(String name, double value, String suggestion) { + this.name = name; + this.value = value; + this.suggestion = suggestion; + } + + public SuggestionResult(String name, double value) { + this.value = value; + this.name = name; + } + } + + public static class TaskDistributionResponse extends MRTaskExecutionResponse { + public String counterName; + public List<CountUnit> taskBuckets; + + public TaskDistributionResponse() { + taskBuckets = new ArrayList<>(); + } + } + + public static class CountUnit { + public long bucket; + public long countVal; + + public CountUnit(long bucket) { + this.bucket = bucket; + this.countVal = 0; + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/ResourceUtils.java ---------------------------------------------------------------------- diff --git a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/ResourceUtils.java b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/ResourceUtils.java new file mode 100644 index 0000000..381c7ec --- /dev/null +++ b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/ResourceUtils.java @@ -0,0 +1,87 @@ +/* + * 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.eagle.service.jpm; + +import org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity; +import org.apache.eagle.jpm.util.jobcounter.JobCounters; +import org.apache.eagle.log.entity.GenericServiceAPIResponseEntity; +import org.apache.eagle.service.generic.GenericEntityServiceResource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.List; +import java.util.Set; + +public class ResourceUtils { + + private static final Logger LOG = LoggerFactory.getLogger(ResourceUtils.class); + + static GenericEntityServiceResource resource = new GenericEntityServiceResource(); + + public static GenericServiceAPIResponseEntity getQueryResult(String query, String startTime, String endTime) { + return resource.search(query, startTime, endTime, Integer.MAX_VALUE, null, false, true, 0L, 0, true, 0, null, false); + } + + public static double[] getCounterValues(List<TaskExecutionAPIEntity> tasks, JobCounters.CounterName counterName) { + List<Double> values = new ArrayList<>(); + for (TaskExecutionAPIEntity task : tasks) { + values.add(Double.valueOf(task.getJobCounters().getCounterValue(counterName))); + } + return toArray(values); + } + + public static double[] toArray(List<Double> input) { + double[] result = new double[input.size()]; + for (int i = 0; i < result.length; i++) { + result[i] = input.get(i); + } + return result; + } + + public static List<Long> parseDistributionList(String timelist) { + List<Long> times = new ArrayList<>(); + String [] strs = timelist.split("[,\\s]"); + for (String str : strs) { + try { + times.add(Long.parseLong(str)); + } catch (Exception ex) { + LOG.warn(str + " is not a number"); + } + } + return times; + } + + public static int getDistributionPosition(List<Long> rangeList, Long value) { + for (int i = 1; i < rangeList.size(); i++) { + if (value < rangeList.get(i)) { + return i - 1; + } + } + return rangeList.size() - 1; + } + + public static boolean isDuplicate(Set<String> keys, String jobId) { + if (keys.isEmpty()) { + return false; + } + return keys.contains(jobId); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/TaskCountByDurationHelper.java ---------------------------------------------------------------------- diff --git a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/TaskCountByDurationHelper.java b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/TaskCountByDurationHelper.java deleted file mode 100644 index 0eeb440..0000000 --- a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/TaskCountByDurationHelper.java +++ /dev/null @@ -1,106 +0,0 @@ -/* - * 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.eagle.service.jpm; - -import org.apache.eagle.jpm.mr.runningentity.TaskExecutionAPIEntity; -import org.apache.eagle.jpm.util.Constants; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.ArrayList; -import java.util.Comparator; -import java.util.Iterator; -import java.util.List; - -public class TaskCountByDurationHelper { - - private static final Logger LOG = LoggerFactory.getLogger(TaskCountByDurationHelper.class); - - public static List<Long> parseTimeList(String timelist) { - List<Long> times = new ArrayList<>(); - String [] strs = timelist.split("[,\\s]"); - for (String str : strs) { - try { - times.add(Long.parseLong(str)); - } catch (Exception ex) { - LOG.warn(str + " is not a number"); - } - } - return times; - } - - public static int getPosition(List<Long> times, Long duration) { - duration = duration / 1000; - for (int i = 1; i < times.size(); i++) { - if (duration < times.get(i)) { - return i - 1; - } - } - return times.size() - 1; - } - - public void getTopTasks(List<MRJobTaskCountResponse.UnitTaskCount> list, long top) { - for (MRJobTaskCountResponse.UnitTaskCount taskCounter : list) { - Iterator<TaskExecutionAPIEntity> iterator = taskCounter.entities.iterator(); - for (int i = 0; i < top && iterator.hasNext(); i++) { - taskCounter.topEntities.add(iterator.next()); - } - taskCounter.entities.clear(); - } - } - - public void countTask(MRJobTaskCountResponse.UnitTaskCount counter, String taskType) { - counter.taskCount++; - if (taskType.equalsIgnoreCase(Constants.TaskType.MAP.toString())) { - counter.mapTaskCount++; - } else if (taskType.equalsIgnoreCase(Constants.TaskType.REDUCE.toString())) { - counter.reduceTaskCount++; - } - } - - public void initTaskCountList(List<MRJobTaskCountResponse.UnitTaskCount> runningTaskCount, - List<MRJobTaskCountResponse.UnitTaskCount> finishedTaskCount, - List<Long> times, - Comparator comparator) { - for (int i = 0; i < times.size(); i++) { - runningTaskCount.add(new MRJobTaskCountResponse.UnitTaskCount(times.get(i), comparator)); - finishedTaskCount.add(new MRJobTaskCountResponse.UnitTaskCount(times.get(i), comparator)); - } - } - - static class RunningTaskComparator implements Comparator<TaskExecutionAPIEntity> { - @Override - public int compare(TaskExecutionAPIEntity o1, TaskExecutionAPIEntity o2) { - Long time1 = o1.getDuration(); - Long time2 = o2.getDuration(); - return (time1 > time2 ? -1 : (time1 == time2) ? 0 : 1); - } - } - - static class HistoryTaskComparator implements Comparator<org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity> { - @Override - public int compare(org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity o1, - org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity o2) { - Long time1 = o1.getDuration(); - Long time2 = o2.getDuration(); - return (time1 > time2 ? -1 : (time1 == time2) ? 0 : 1); - } - } - -} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/count/MRJobCountImpl.java ---------------------------------------------------------------------- diff --git a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/count/MRJobCountImpl.java b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/count/MRJobCountImpl.java new file mode 100644 index 0000000..d8fc004 --- /dev/null +++ b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/count/MRJobCountImpl.java @@ -0,0 +1,146 @@ +/* + * 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.eagle.service.jpm.count; + +import org.apache.eagle.common.DateTimeUtil; +import org.apache.eagle.jpm.mr.historyentity.JobExecutionAPIEntity; +import org.apache.eagle.jpm.util.Constants; +import org.apache.eagle.jpm.util.MRJobTagName; +import org.apache.eagle.service.jpm.MRJobTaskCountResponse.JobCountResponse; +import org.apache.eagle.service.jpm.MRJobTaskCountResponse.UnitJobCount; +import org.apache.eagle.service.jpm.ResourceUtils; + +import java.text.ParseException; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +public class MRJobCountImpl { + + public void initJobCountList(List<UnitJobCount> jobCounts, long startTime, long endTime, long intervalInSecs) { + for (long i = startTime / intervalInSecs; i * intervalInSecs <= endTime; i++) { + jobCounts.add(new UnitJobCount(i * intervalInSecs * DateTimeUtil.ONESECOND)); + } + } + + public String moveTimeForwardOneDay(String startTime) throws ParseException { + long timeInSecs = DateTimeUtil.humanDateToSeconds(startTime); + timeInSecs -= 24L * 60L * 60L; + return DateTimeUtil.secondsToHumanDate(timeInSecs); + } + + public JobCountResponse getRunningJobCount(List<JobExecutionAPIEntity> historyJobs, + List<org.apache.eagle.jpm.mr.runningentity.JobExecutionAPIEntity> runningJobs, + long startTimeInSecs, + long endTimeInSecs, + long intervalInSecs) { + List<UnitJobCount> jobCounts = new ArrayList<>(); + Set<String> jobTypes = new HashSet<>(); + Set<String> jobIds = new HashSet<>(); + initJobCountList(jobCounts, startTimeInSecs, endTimeInSecs, intervalInSecs); + long startTimeInMs = startTimeInSecs * DateTimeUtil.ONESECOND; + for (JobExecutionAPIEntity job: historyJobs) { + jobIds.add(job.getTags().get(MRJobTagName.JOB_ID.toString())); + if (job.getEndTime() >= startTimeInMs) { + String jobType = job.getTags().get(MRJobTagName.JOB_TYPE.toString()); + jobTypes.add(jobType); + countJob(jobCounts, job.getStartTime() / DateTimeUtil.ONESECOND, job.getEndTime() / DateTimeUtil.ONESECOND, intervalInSecs, jobType); + } + } + for (org.apache.eagle.jpm.mr.runningentity.JobExecutionAPIEntity job : runningJobs) { + if (!ResourceUtils.isDuplicate(jobIds, job.getTags().get(MRJobTagName.JOB_ID.toString()))) { + String jobType = job.getTags().get(MRJobTagName.JOB_TYPE.toString()); + jobTypes.add(jobType); + countJob(jobCounts, job.getStartTime() / DateTimeUtil.ONESECOND, endTimeInSecs, intervalInSecs, jobType); + } + } + JobCountResponse response = new JobCountResponse(); + response.jobCounts = jobCounts; + response.jobTypes = jobTypes; + return response; + } + + public JobCountResponse getHistoryJobCountGroupByDuration(List<JobExecutionAPIEntity> jobDurations, String timeList) { + JobCountResponse response = new JobCountResponse(); + List<UnitJobCount> jobCounts = new ArrayList<>(); + Set<String> jobTypes = new HashSet<>(); + List<Long> times = ResourceUtils.parseDistributionList(timeList); + for (int i = 0; i < times.size(); i++) { + jobCounts.add(new UnitJobCount(times.get(i))); + } + for (JobExecutionAPIEntity job : jobDurations) { + int jobIndex = ResourceUtils.getDistributionPosition(times, job.getDurationTime() / DateTimeUtil.ONESECOND); + UnitJobCount counter = jobCounts.get(jobIndex); + String jobType = job.getTags().get(MRJobTagName.JOB_TYPE.toString()); + jobTypes.add(jobType); + countJob(counter, jobType); + } + response.jobCounts = jobCounts; + response.jobTypes = jobTypes; + return response; + } + + public void countJob(UnitJobCount counter, String jobType) { + if (null == jobType) { + jobType = "null"; + } + counter.jobCount++; + if (counter.jobCountByType.containsKey(jobType)) { + counter.jobCountByType.put(jobType, counter.jobCountByType.get(jobType) + 1); + } else { + counter.jobCountByType.put(jobType, 1L); + } + } + + public void countJob(List<UnitJobCount> jobCounts, long jobStartTimeSecs, long jobEndTimeSecs, long intervalInSecs, String jobType) { + long startCountPoint = jobCounts.get(0).timeBucket / DateTimeUtil.ONESECOND; + if (jobEndTimeSecs < startCountPoint) { + return; + } + int startIndex = 0; + if (jobStartTimeSecs > startCountPoint) { + long relativeStartTime = jobStartTimeSecs - startCountPoint; + startIndex = (int) (relativeStartTime / intervalInSecs) + (relativeStartTime % intervalInSecs == 0 ? 0 : 1); + } + long relativeEndTime = jobEndTimeSecs - startCountPoint; + int endIndex = (int) (relativeEndTime / intervalInSecs); + + for (int i = startIndex; i <= endIndex && i < jobCounts.size(); i++) { + countJob(jobCounts.get(i), jobType); + } + } + + public List<String> getSearchTimeDuration(List<JobExecutionAPIEntity> jobEntities) { + List<String> pair = new ArrayList<>(); + long minStartTime = System.currentTimeMillis(); + long maxEndTime = 0; + for (JobExecutionAPIEntity jobEntity : jobEntities) { + if (minStartTime > jobEntity.getStartTime()) { + minStartTime = jobEntity.getStartTime(); + } + if (maxEndTime < jobEntity.getEndTime()) { + maxEndTime = jobEntity.getEndTime(); + } + } + pair.add(DateTimeUtil.millisecondsToHumanDateWithSeconds(minStartTime)); + pair.add(DateTimeUtil.millisecondsToHumanDateWithSeconds(maxEndTime)); + return pair; + } +} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/count/MRTaskCountImpl.java ---------------------------------------------------------------------- diff --git a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/count/MRTaskCountImpl.java b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/count/MRTaskCountImpl.java new file mode 100644 index 0000000..f5017eb --- /dev/null +++ b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/count/MRTaskCountImpl.java @@ -0,0 +1,124 @@ +/* + * 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.eagle.service.jpm.count; + +import org.apache.commons.io.FileUtils; +import org.apache.eagle.common.DateTimeUtil; +import org.apache.eagle.jpm.mr.runningentity.TaskExecutionAPIEntity; +import org.apache.eagle.jpm.util.Constants; +import org.apache.eagle.jpm.util.MRJobTagName; +import org.apache.eagle.jpm.util.jobcounter.JobCounters; +import org.apache.eagle.service.jpm.MRJobTaskCountResponse; +import org.apache.eagle.service.jpm.MRTaskExecutionResponse; +import org.apache.eagle.service.jpm.ResourceUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.*; + +public class MRTaskCountImpl { + + private static final Logger LOG = LoggerFactory.getLogger(MRTaskCountImpl.class); + + public void getTopTasks(List<MRJobTaskCountResponse.UnitTaskCount> list, long top) { + for (MRJobTaskCountResponse.UnitTaskCount taskCounter : list) { + Iterator<TaskExecutionAPIEntity> iterator = taskCounter.entities.iterator(); + for (int i = 0; i < top && iterator.hasNext(); i++) { + taskCounter.topEntities.add(iterator.next()); + } + taskCounter.entities.clear(); + } + } + + public void countTask(MRJobTaskCountResponse.UnitTaskCount counter, String taskType) { + counter.taskCount++; + if (taskType.equalsIgnoreCase(Constants.TaskType.MAP.toString())) { + counter.mapTaskCount++; + } else if (taskType.equalsIgnoreCase(Constants.TaskType.REDUCE.toString())) { + counter.reduceTaskCount++; + } + } + + public void initTaskCountList(List<MRJobTaskCountResponse.UnitTaskCount> runningTaskCount, + List<MRJobTaskCountResponse.UnitTaskCount> finishedTaskCount, + List<Long> times, + Comparator comparator) { + for (int i = 0; i < times.size(); i++) { + runningTaskCount.add(new MRJobTaskCountResponse.UnitTaskCount(times.get(i), comparator)); + finishedTaskCount.add(new MRJobTaskCountResponse.UnitTaskCount(times.get(i), comparator)); + } + } + + public static class RunningTaskComparator implements Comparator<TaskExecutionAPIEntity> { + @Override + public int compare(TaskExecutionAPIEntity o1, TaskExecutionAPIEntity o2) { + Long time1 = o1.getDuration(); + Long time2 = o2.getDuration(); + return (time1 > time2 ? -1 : (time1.equals(time2)) ? 0 : 1); + } + } + + public static class HistoryTaskComparator implements Comparator<org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity> { + @Override + public int compare(org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity o1, + org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity o2) { + Long time1 = o1.getDuration(); + Long time2 = o2.getDuration(); + return (time1 > time2 ? -1 : (time1.equals(time2)) ? 0 : 1); + } + } + + public MRJobTaskCountResponse.HistoryTaskCountResponse countHistoryTask(List<org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity> tasks, long startTimeInMin, long endTimeInMin) { + List<MRJobTaskCountResponse.UnitTaskCount> taskCounts = new ArrayList<>(); + for (long i = startTimeInMin; i <= endTimeInMin; i++) { + taskCounts.add(new MRJobTaskCountResponse.UnitTaskCount(i * DateTimeUtil.ONEMINUTE, null)); + } + for (org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity task : tasks) { + String taskType = task.getTags().get(MRJobTagName.TASK_TYPE.toString()); + long taskStarTimeMin = task.getStartTime() / DateTimeUtil.ONEMINUTE; + long taskEndTimeMin = task.getEndTime() / DateTimeUtil.ONEMINUTE; + int relativeStartTime = (int) (taskStarTimeMin - startTimeInMin); + int relativeEndTime = (int) (taskEndTimeMin - startTimeInMin); + for (int i = relativeStartTime; i <= relativeEndTime; i++) { + countTask(taskCounts.get(i), taskType); + } + } + MRJobTaskCountResponse.HistoryTaskCountResponse response = new MRJobTaskCountResponse.HistoryTaskCountResponse(); + response.taskCount = taskCounts; + return response; + } + + public MRTaskExecutionResponse.TaskDistributionResponse getHistoryTaskDistribution(List<org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity> tasks, String counterName, String distRange) { + MRTaskExecutionResponse.TaskDistributionResponse response = new MRTaskExecutionResponse.TaskDistributionResponse(); + response.counterName = counterName; + List<Long> distRangeList = ResourceUtils.parseDistributionList(distRange); + for (int i = 0; i < distRangeList.size(); i++) { + response.taskBuckets.add(new MRTaskExecutionResponse.CountUnit(distRangeList.get(i))); + } + JobCounters.CounterName jobCounterName = JobCounters.CounterName.valueOf(counterName.toUpperCase()); + for (org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity task : tasks) { + Long counterValue = task.getJobCounters().getCounterValue(jobCounterName); + int pos = ResourceUtils.getDistributionPosition(distRangeList, counterValue); + response.taskBuckets.get(pos).countVal++; + } + return response; + } + + +} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/suggestion/AbstractGCFunc.java ---------------------------------------------------------------------- diff --git a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/suggestion/AbstractGCFunc.java b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/suggestion/AbstractGCFunc.java new file mode 100644 index 0000000..ad38fcb --- /dev/null +++ b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/suggestion/AbstractGCFunc.java @@ -0,0 +1,104 @@ +/* + * 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.eagle.service.jpm.suggestion; + +import org.apache.commons.math.stat.descriptive.DescriptiveStatistics; +import org.apache.eagle.jpm.mr.historyentity.TaskExecutionAPIEntity; +import org.apache.eagle.jpm.util.Constants; +import org.apache.eagle.jpm.util.jobcounter.JobCounters; +import org.apache.eagle.service.jpm.MRTaskExecutionResponse; +import org.apache.eagle.service.jpm.MRTaskExecutionResponse.TaskGroup; +import org.apache.eagle.service.jpm.MRTaskExecutionResponse.JobSuggestionResponse; +import org.apache.eagle.service.jpm.MRTaskExecutionResponse.TaskGroupResponse; +import org.apache.eagle.service.jpm.ResourceUtils; + +import java.util.ArrayList; +import java.util.List; + +public abstract class AbstractGCFunc implements SuggestionFunc { + + private static final String GC_RATIO_NAME_FORMAT = "gcRatio (%s / %s) deviation"; + private static final String GC_SUGGESTION_FORMAT = "gcRatio deviation exceeds threshold %.2f, where the deviation is %.2f / %.2f"; + + private static final double GC_RATIO_DEVIATION_THRESHOLD = 2; + + private Constants.SuggestionType suggestionType; + private double threshold; + + public AbstractGCFunc(Constants.SuggestionType suggestionType) { + this.suggestionType = suggestionType; + this.threshold = GC_RATIO_DEVIATION_THRESHOLD; + } + + public AbstractGCFunc(Constants.SuggestionType suggestionType, double threshold) { + this.suggestionType = suggestionType; + this.threshold = threshold > 0 ? threshold : GC_RATIO_DEVIATION_THRESHOLD; + } + + protected abstract TaskGroup getTasks(TaskGroupResponse tasks); + + + private double getGcRatio(List<TaskExecutionAPIEntity> tasks) { + if (tasks.isEmpty()) { + return 0; + } + double[] gcMs = ResourceUtils.getCounterValues(tasks, JobCounters.CounterName.GC_MILLISECONDS); + double[] cpuMs = ResourceUtils.getCounterValues(tasks, JobCounters.CounterName.CPU_MILLISECONDS); + + DescriptiveStatistics statistics = new DescriptiveStatistics(); + double averageCpuMs = statistics.getMeanImpl().evaluate(cpuMs); + double averageGcMs = statistics.getMeanImpl().evaluate(gcMs); + if (averageCpuMs == 0) { + averageCpuMs = 1; + } + return averageGcMs / averageCpuMs; + } + + @Override + public JobSuggestionResponse apply(TaskGroupResponse data) { + JobSuggestionResponse response = new JobSuggestionResponse(); + response.suggestionType = suggestionType.name(); + + TaskGroup taskGroup = getTasks(data); + if (taskGroup.longTasks.isEmpty()) { + return response; + } + + double smallerGcRatio = getGcRatio(taskGroup.shortTasks); + double largerGcRatio = getGcRatio(taskGroup.longTasks); + response.suggestionResults = getGCsuggest(smallerGcRatio, largerGcRatio); + return response; + } + + private List<MRTaskExecutionResponse.SuggestionResult> getGCsuggest(double smallerRatio, double largerRatio) { + if (smallerRatio <= 0) { + smallerRatio = 1; + } + double deviation = largerRatio / smallerRatio; + String suggestName = String.format(GC_RATIO_NAME_FORMAT, JobCounters.CounterName.GC_MILLISECONDS.getName(), JobCounters.CounterName.CPU_MILLISECONDS.getName()); + String suggestion = null; + if (deviation > threshold) { + suggestion = String.format(GC_SUGGESTION_FORMAT, threshold, largerRatio, smallerRatio); + } + List<MRTaskExecutionResponse.SuggestionResult> suggestionResults = new ArrayList<>(); + suggestionResults.add(new MRTaskExecutionResponse.SuggestionResult(suggestName, deviation, suggestion)); + return suggestionResults; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/suggestion/AbstractInputFunc.java ---------------------------------------------------------------------- diff --git a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/suggestion/AbstractInputFunc.java b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/suggestion/AbstractInputFunc.java new file mode 100644 index 0000000..5a89055 --- /dev/null +++ b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/suggestion/AbstractInputFunc.java @@ -0,0 +1,86 @@ +/* + * 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.eagle.service.jpm.suggestion; + +import org.apache.commons.math.stat.descriptive.DescriptiveStatistics; +import org.apache.eagle.jpm.util.Constants; +import org.apache.eagle.jpm.util.jobcounter.JobCounters; +import org.apache.eagle.service.jpm.MRTaskExecutionResponse; +import org.apache.eagle.service.jpm.MRTaskExecutionResponse.JobSuggestionResponse; +import org.apache.eagle.service.jpm.MRTaskExecutionResponse.TaskGroupResponse; +import org.apache.eagle.service.jpm.ResourceUtils; + +import java.util.ArrayList; +import java.util.List; + +public abstract class AbstractInputFunc implements SuggestionFunc { + + private JobCounters.CounterName counterName; + private Constants.SuggestionType suggestType; + private double threshold; + private static final long DATA_SKEW_THRESHOLD = 2; + + private static final String DEVIATION_SUGGEST_FORMAT = "average %s deviation"; + private static final String DATA_SKEW_SUGGESTION_FORMAT = "%s deviation exceeds threshold %.2f, where the deviation is %.2f / %.2f"; + + public AbstractInputFunc(JobCounters.CounterName counterName, Constants.SuggestionType type) { + this.counterName = counterName; + this.suggestType = type; + this.threshold = DATA_SKEW_THRESHOLD; + } + + public AbstractInputFunc(JobCounters.CounterName counterName, Constants.SuggestionType type, double threshold) { + this.counterName = counterName; + this.suggestType = type; + this.threshold = threshold > 0 ? threshold : DATA_SKEW_THRESHOLD; + } + + protected abstract MRTaskExecutionResponse.TaskGroup getTasks(TaskGroupResponse tasks); + + @Override + public JobSuggestionResponse apply(TaskGroupResponse data) { + MRTaskExecutionResponse.TaskGroup taskGroup = getTasks(data); + double[] smallerGroup = ResourceUtils.getCounterValues(taskGroup.shortTasks, counterName); + double[] largerGroup = ResourceUtils.getCounterValues(taskGroup.longTasks, counterName); + DescriptiveStatistics statistics = new DescriptiveStatistics(); + double avgSmaller = statistics.getMeanImpl().evaluate(smallerGroup); + double avgLarger = statistics.getMeanImpl().evaluate(largerGroup); + + List<MRTaskExecutionResponse.SuggestionResult> suggestionResults = getDeviationSuggest(avgSmaller, avgLarger); + MRTaskExecutionResponse.JobSuggestionResponse response = new MRTaskExecutionResponse.JobSuggestionResponse(); + response.suggestionResults = suggestionResults; + response.suggestionType = suggestType.toString(); + return response; + } + + private List<MRTaskExecutionResponse.SuggestionResult> getDeviationSuggest(double avgSmaller, double avgLarger) { + if (avgSmaller <= 0) { + avgSmaller = 1; + } + double deviation = avgLarger / avgSmaller; + String suggestName = String.format(DEVIATION_SUGGEST_FORMAT, counterName.getName()); + String suggestion = null; + if (deviation > threshold) { + suggestion = String.format(DATA_SKEW_SUGGESTION_FORMAT, counterName.getName(), threshold, avgLarger, avgSmaller); + } + List<MRTaskExecutionResponse.SuggestionResult> suggestionResults = new ArrayList<>(); + suggestionResults.add(new MRTaskExecutionResponse.SuggestionResult(suggestName, deviation, suggestion)); + return suggestionResults; + } +} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/cfd5c7f8/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/suggestion/MapGCFunc.java ---------------------------------------------------------------------- diff --git a/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/suggestion/MapGCFunc.java b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/suggestion/MapGCFunc.java new file mode 100644 index 0000000..9fd54b0 --- /dev/null +++ b/eagle-jpm/eagle-jpm-service/src/main/java/org/apache/eagle/service/jpm/suggestion/MapGCFunc.java @@ -0,0 +1,38 @@ +/* + * 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.eagle.service.jpm.suggestion; + +import org.apache.eagle.jpm.util.Constants; +import org.apache.eagle.service.jpm.MRTaskExecutionResponse; + +public class MapGCFunc extends AbstractGCFunc { + + public MapGCFunc() { + super(Constants.SuggestionType.MapGC); + } + + public MapGCFunc(double threshold) { + super(Constants.SuggestionType.MapGC, threshold); + } + + @Override + protected MRTaskExecutionResponse.TaskGroup getTasks(MRTaskExecutionResponse.TaskGroupResponse tasks) { + return tasks.tasksGroupByType.get(Constants.TaskType.MAP.toString()); + } +}
