Skip to content

Job metrics of Dashboard for long term #2131

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Syleechan wants to merge 2 commits into
base: kylin5
Choose a base branch
from
Open

Job metrics of Dashboard for long term #2131

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
05c0ce2
Job metrics of Dashboard for long term
Syleechan Jul 3, 2023
26cbf6a
tests
Syleechan Jul 4, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions src/core-common/src/main/resources/metadata-jdbc-h2.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,37 @@ create.queryhistoryrealization.store.tableindex2=CREATE INDEX %s_ix2 ON %s ( mod
create.queryhistoryrealization.store.tableindex3=CREATE INDEX %s_ix3 ON %s ( query_first_day_of_month );
create.queryhistoryrealization.store.tableindex4=CREATE INDEX %s_ix4 ON %s ( query_first_day_of_week );
create.queryhistoryrealization.store.tableindex5=CREATE INDEX %s_ix5 ON %s ( query_day );

#Job Metrics Store
create.jobmetrics.store.table=CREATE TABLE IF NOT EXISTS `%s` ( \
id bigint not null auto_increment, \
`job_id` VARCHAR(255) , \
`submitter` VARCHAR(100), \
`model` VARCHAR(128), \
`job_type` VARCHAR(100), \
`job_engine` VARCHAR(255), \
`project_name` VARCHAR(100), \
`job_state` VARCHAR(50), \
`error_type` VARCHAR(128), \
`error_info` TEXT, \
`duration` BIGINT, \
`per_bytes_time_cost` DOUBLE, \
`model_size` BIGINT, \
`wait_time` BIGINT, \
`build_time` BIGINT, \
`build_first_day_of_month` BIGINT, \
`build_first_day_of_week` BIGINT, \
`build_day` BIGINT, \
`build_date` DATE, \
primary key (`id`,`project_name`) \
);
create.jobmetrics.store.tableindex1=CREATE INDEX %s_ix1 ON %s ( build_time );
create.jobmetrics.store.tableindex2=CREATE INDEX %s_ix2 ON %s ( model );
create.jobmetrics.store.tableindex3=CREATE INDEX %s_ix3 ON %s ( build_first_day_of_month );
create.jobmetrics.store.tableindex4=CREATE INDEX %s_ix4 ON %s ( build_first_day_of_week );
create.jobmetrics.store.tableindex5=CREATE INDEX %s_ix5 ON %s ( build_day );


# RAW RECOMMENDATION STORE
create.rawrecommendation.store.table=CREATE TABLE IF NOT EXISTS `%s` ( \
`id` int not null auto_increment, \
Expand Down
29 changes: 29 additions & 0 deletions src/core-common/src/main/resources/metadata-jdbc-mysql.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,35 @@ create.streamingjobrecord.store.table=CREATE TABLE IF NOT EXISTS %s ( \
create.streamingjobrecord.store.tableindex1=CREATE INDEX %s_idx1 ON %s USING btree ( job_id );
create.streamingjobrecord.store.tableindex2=CREATE INDEX %s_idx2 ON %s USING btree ( create_time );

#Job Metrics Store
create.jobmetrics.store.table=CREATE TABLE IF NOT EXISTS `%s` ( \
id bigint not null auto_increment, \
`job_id` VARCHAR(255) , \
`submitter` VARCHAR(100), \
`model` VARCHAR(128), \
`job_type` VARCHAR(100), \
`job_engine` VARCHAR(255), \
`project_name` VARCHAR(100), \
`job_state` VARCHAR(50), \
`error_type` VARCHAR(128), \
`error_info` TEXT, \
`duration` BIGINT, \
`per_bytes_time_cost` DOUBLE, \
`model_size` BIGINT, \
`wait_time` BIGINT, \
`build_time` BIGINT, \
`build_first_day_of_month` BIGINT, \
`build_first_day_of_week` BIGINT, \
`build_day` BIGINT, \
`build_date` DATE, \
primary key (`id`,`project_name`) \
);
create.jobmetrics.store.tableindex1=CREATE INDEX %s_ix1 ON %s ( build_time );
create.jobmetrics.store.tableindex2=CREATE INDEX %s_ix2 ON %s ( model );
create.jobmetrics.store.tableindex3=CREATE INDEX %s_ix3 ON %s ( build_first_day_of_month );
create.jobmetrics.store.tableindex4=CREATE INDEX %s_ix4 ON %s ( build_first_day_of_week );
create.jobmetrics.store.tableindex5=CREATE INDEX %s_ix5 ON %s ( build_day );

# RAW RECOMMENDATION STORE
create.rawrecommendation.store.table=CREATE TABLE IF NOT EXISTS `%s` ( \
`id` int not null auto_increment, \
Expand Down
81 changes: 76 additions & 5 deletions src/core-job/src/main/java/org/apache/kylin/job/execution/DefaultExecutable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,11 @@

package org.apache.kylin.job.execution;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.Map;
Expand All @@ -35,10 +38,15 @@
import org.apache.kylin.common.persistence.transaction.UnitOfWork;
import org.apache.kylin.common.scheduler.EventBusFactory;
import org.apache.kylin.common.scheduler.JobFinishedNotifier;
import org.apache.kylin.common.util.StringHelper;
import org.apache.kylin.common.util.TimeUtil;
import org.apache.kylin.job.exception.ExecuteException;
import org.apache.kylin.job.exception.ExecuteRuntimeException;
import org.apache.kylin.job.exception.JobStoppedException;
import org.apache.kylin.job.exception.PersistentException;
import org.apache.kylin.job.metrics.JobMetrics;
import org.apache.kylin.job.metrics.RDBMJobMetricsDAO;
import org.apache.kylin.job.util.MailNotificationUtil;
import org.apache.kylin.metadata.project.EnhancedUnitOfWork;

import org.apache.kylin.guava30.shaded.common.collect.Lists;
Expand Down Expand Up @@ -229,15 +237,16 @@ protected void onExecuteFinished(ExecuteResult result) throws JobStoppedExceptio
switch (state) {
case SUCCEED:
updateToFinalState(ExecutableState.SUCCEED, this::afterUpdateOutput, result.getShortErrMsg());
onStatusChange(ExecutableState.SUCCEED);
onStatusChange(ExecutableState.SUCCEED, context, result);
updateJobMetrics(state, context, result);
break;
case DISCARDED:
updateToFinalState(ExecutableState.DISCARDED, this::onExecuteDiscardHook, result.getShortErrMsg());
onStatusChange(ExecutableState.DISCARDED);
onStatusChange(ExecutableState.DISCARDED, context, result);
break;
case SUICIDAL:
updateToFinalState(ExecutableState.SUICIDAL, this::onExecuteSuicidalHook, result.getShortErrMsg());
onStatusChange(ExecutableState.SUICIDAL);
onStatusChange(ExecutableState.SUICIDAL, context, result);
break;
case ERROR:
case PAUSED:
Expand All @@ -260,7 +269,8 @@ protected void onExecuteFinished(ExecuteResult result) throws JobStoppedExceptio
}
updateJobOutput(getProject(), getId(), state, info, output, shortErrMsg, hook);
if (state == ExecutableState.ERROR) {
onStatusChange(ExecutableState.ERROR);
onStatusChange(ExecutableState.ERROR, context, result);
updateJobMetrics(state, context, result);
}
break;
default:
Expand Down Expand Up @@ -411,7 +421,68 @@ protected void afterUpdateOutput(String jobId) {
// just implement it
}

protected void onStatusChange(ExecutableState state) {
protected void onStatusChange(ExecutableState state, ExecutableContext context, ExecuteResult result) {
super.notifyUserStatusChange(state);
}

protected void updateJobMetrics(ExecutableState state, ExecutableContext context, ExecuteResult result) {
JobMetrics jobMetrics = new JobMetrics();
jobMetrics.setJobId(getId());
jobMetrics.setJobType(getJobType().toString());
jobMetrics.setJobState(state.toStringState());
jobMetrics.setProjectName(getProject());
jobMetrics.setModel(getTargetSubjectAlias());
jobMetrics.setSubmitter(StringHelper.noBlank(getSubmitter(), "missing submitter"));
jobMetrics.setJobEngine(MailNotificationUtil.getLocalHostName());

jobMetrics.setBuildTime(getJobEndTime());
long dayStart = TimeUtil.getDayStart(getJobEndTime());
Date date = new Date(dayStart);
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd", Locale.getDefault(Locale.Category.FORMAT));
String buildDate = sdf.format(date);
try {
jobMetrics.setBuildDate(sdf.parse(buildDate));
} catch (ParseException e) {
logger.error("time format is error");
}
jobMetrics.setBuildDay(dayStart);
jobMetrics.setBuildFirstDayOfWeek(TimeUtil.getWeekStart(getJobEndTime()));
jobMetrics.setBuildFirstDayOfMonth(TimeUtil.getMonthStart(getJobEndTime()));

if (state == ExecutableState.SUCCEED) {
jobMetrics.setDuration(getDuration());
jobMetrics.setWaitTime(getWaitTime());
jobMetrics.setModelSize(getByteSize());
jobMetrics.setPerBytesTimeCost(getPerBytesTimeCost(getByteSize(), getDuration()));
} else if (state == ExecutableState.ERROR) {
AbstractExecutable errorTask = null;
Output errorOutput;
List<AbstractExecutable> tasks = getTasks();
for (AbstractExecutable task : tasks) {
errorOutput = getManager().getOutput(task.getId());
if (errorOutput.getState() == ExecutableState.ERROR) {
errorTask = task;
break;
}
}
if (errorTask != null) {
jobMetrics.setErrorType(errorTask.getName());
jobMetrics.setErrorInfo(result.getShortErrMsg());
}
}
updateJobMetrics(jobMetrics);
}

private void updateJobMetrics(JobMetrics jobMetrics) {
RDBMJobMetricsDAO jobMetricsDAO = RDBMJobMetricsDAO.getInstance();
jobMetricsDAO.insert(jobMetrics);

}

private static double getPerBytesTimeCost(long byteSize, long time) {
if (byteSize <= 0) {
return 0;
}
return time * 1.0 / byteSize;
}
}
Loading