Skip to content

D
dataease
提交 afd2a7ae authored 作者: taojinlong's avatar taojinlong
浏览文件
操作

fix: 定时任务创建全量任务，执行完成后修改为增量，发现增量不生效

上级 ec9139bd
隐藏空白字符变更
内嵌 并排
正在显示 包含 100 行增加 和 73 行删除
backend/src/main/java/io/dataease/service/chart/ChartGroupService.java
浏览文件 @ afd2a7ae
......@@ -31,10 +31,6 @@ public class ChartGroupService {
@Resource
private ExtChartGroupMapper extChartGroupMapper;
@Resource
private ExtDataSetGroupMapper extDataSetGroupMapper;
@Resource
private SysAuthService sysAuthService;
@Resource
private ExtChartViewMapper extChartViewMapper;
public ChartGroupDTO save(ChartGroup chartGroup) {
......
backend/src/main/java/io/dataease/service/dataset/DataSetTableService.java
浏览文件 @ afd2a7ae
......@@ -2245,8 +2245,7 @@ public class DataSetTableService {
List<DatasetTable> jobStoppeddDatasetTables = new ArrayList<>();
datasetTableMapper.selectByExample(example).forEach(datasetTable -> {
if (StringUtils.isEmpty(datasetTable.getQrtzInstance()) || !activeQrtzInstances.contains(
datasetTable.getQrtzInstance().substring(0, datasetTable.getQrtzInstance().length() - 13))) {
if (StringUtils.isNotEmpty(datasetTable.getQrtzInstance()) && !activeQrtzInstances.contains(datasetTable.getQrtzInstance().substring(0, datasetTable.getQrtzInstance().length() - 13))) {
jobStoppeddDatasetTables.add(datasetTable);
}
});
......@@ -2314,4 +2313,8 @@ public class DataSetTableService {
saveTableField(datasetTable);
return datasetTable;
}
public int updateByExampleSelective(DatasetTable record, DatasetTableExample example ){
return datasetTableMapper.updateByExampleSelective(record, example);
}
}
backend/src/main/java/io/dataease/service/dataset/DataSetTableTaskLogService.java
浏览文件 @ afd2a7ae
......@@ -10,6 +10,7 @@ import io.dataease.dto.dataset.DataSetTaskLogDTO;
import io.dataease.plugins.common.base.domain.DatasetTableTaskLog;
import io.dataease.plugins.common.base.domain.DatasetTableTaskLogExample;
import io.dataease.plugins.common.base.mapper.DatasetTableTaskLogMapper;
import io.dataease.plugins.common.base.mapper.DatasetTableTaskMapper;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
......@@ -29,8 +30,13 @@ public class DataSetTableTaskLogService {
private DatasetTableTaskLogMapper datasetTableTaskLogMapper;
@Resource
private ExtDataSetTaskMapper extDataSetTaskMapper;
@Resource
private DatasetTableTaskMapper datasetTableTaskMapper;
public DatasetTableTaskLog save(DatasetTableTaskLog datasetTableTaskLog) {
if(datasetTableTaskMapper.selectByPrimaryKey(datasetTableTaskLog.getTaskId()) == null){
return datasetTableTaskLog;
}
if (StringUtils.isEmpty(datasetTableTaskLog.getId())) {
datasetTableTaskLog.setId(UUID.randomUUID().toString());
datasetTableTaskLog.setCreateTime(System.currentTimeMillis());
......
backend/src/main/java/io/dataease/service/dataset/DataSetTableTaskService.java
浏览文件 @ afd2a7ae
......@@ -108,26 +108,32 @@ public class DataSetTableTaskService {
DataEaseException.throwException(Translator.get("i18n_not_exec_add_sync"));
}
}
if (existSyncTask(dataSetTableService.get(datasetTableTask.getTableId()), datasetTableTask)) {
if (existSyncTask(datasetTableTask.getTableId(), datasetTableTask.getId())) {
DataEaseException.throwException(Translator.get("i18n_sync_job_exists"));
}
}
private synchronized boolean existSyncTask(DatasetTable datasetTable, DatasetTableTask datasetTableTask) {
datasetTable.setSyncStatus(JobStatus.Underway.name());
private synchronized boolean existSyncTask(String datasetTableId, String datasetTableTaskId) {
DatasetTable record = new DatasetTable();
record.setSyncStatus(JobStatus.Underway.name());
DatasetTableExample example = new DatasetTableExample();
example.createCriteria().andIdEqualTo(datasetTable.getId()).andSyncStatusNotEqualTo(JobStatus.Underway.name());
example.or(example.createCriteria().andIdEqualTo(datasetTable.getId()).andSyncStatusIsNull());
Boolean existSyncTask = datasetTableMapper.updateByExampleSelective(datasetTable, example) == 0;
example.createCriteria().andIdEqualTo(datasetTableId).andSyncStatusNotEqualTo(JobStatus.Underway.name());
example.or(example.createCriteria().andIdEqualTo(datasetTableId).andSyncStatusIsNull());
Boolean existSyncTask = datasetTableMapper.updateByExampleSelective(record, example) == 0;
if (!existSyncTask) {
Long startTime = System.currentTimeMillis();
datasetTableTask.setLastExecTime(startTime);
datasetTableTask.setLastExecStatus(JobStatus.Underway.name());
datasetTableTask.setStatus(TaskStatus.Exec.name());
update(datasetTableTask);
DatasetTableTask datasetTableTaskRecord = new DatasetTableTask();
datasetTableTaskRecord.setLastExecTime(startTime);
datasetTableTaskRecord.setLastExecStatus(JobStatus.Underway.name());
datasetTableTaskRecord.setStatus(TaskStatus.Exec.name());
DatasetTableTaskExample datasetTableTaskExample = new DatasetTableTaskExample();
datasetTableTaskExample.createCriteria().andIdEqualTo(datasetTableTaskId);
updateByExampleSelective(datasetTableTaskRecord, datasetTableTaskExample);
DatasetTableTaskLog datasetTableTaskLog = new DatasetTableTaskLog();
datasetTableTaskLog.setTableId(datasetTableTask.getTableId());
datasetTableTaskLog.setTaskId(datasetTableTask.getId());
datasetTableTaskLog.setTableId(datasetTableId);
datasetTableTaskLog.setTaskId(datasetTableTaskId);
datasetTableTaskLog.setStatus(JobStatus.Underway.name());
datasetTableTaskLog.setStartTime(startTime);
datasetTableTaskLog.setTriggerType(TriggerType.Custom.name());
......@@ -173,7 +179,7 @@ public class DataSetTableTaskService {
}
}
public void checkTaskIsStopped(DatasetTableTask datasetTableTask) {
public void checkTaskIsStopped(final DatasetTableTask datasetTableTask) {
if (StringUtils.isNotEmpty(datasetTableTask.getEnd()) && datasetTableTask.getEnd().equalsIgnoreCase("1")) {
BaseGridRequest request = new BaseGridRequest();
ConditionEntity conditionEntity = new ConditionEntity();
......@@ -186,24 +192,30 @@ public class DataSetTableTaskService {
return;
}
if (dataSetTaskDTOS.get(0).getNextExecTime() == null || dataSetTaskDTOS.get(0).getNextExecTime() <= 0) {
datasetTableTask.setStatus(TaskStatus.Stopped.name());
update(datasetTableTask);
DatasetTableTask record = new DatasetTableTask();
record.setStatus(TaskStatus.Stopped.name());
DatasetTableTaskExample datasetTableTaskExample = new DatasetTableTaskExample();
datasetTableTaskExample.createCriteria().andIdEqualTo(datasetTableTask.getId());
updateByExampleSelective(record, datasetTableTaskExample);
return;
}
if (dataSetTaskDTOS.get(0).getNextExecTime() > datasetTableTask.getEndTime()) {
datasetTableTask.setStatus(TaskStatus.Stopped.name());
update(datasetTableTask);
DatasetTableTask record = new DatasetTableTask();
record.setStatus(TaskStatus.Stopped.name());
DatasetTableTaskExample datasetTableTaskExample = new DatasetTableTaskExample();
datasetTableTaskExample.createCriteria().andIdEqualTo(datasetTableTask.getId());
updateByExampleSelective(record, datasetTableTaskExample);
}
}
}
public void updateTaskStatus(DatasetTableTask datasetTableTask, JobStatus lastExecStatus) {
datasetTableTask.setLastExecStatus(lastExecStatus.name());
DatasetTableTask recore = new DatasetTableTask();
recore.setLastExecStatus(lastExecStatus.name());
if (datasetTableTask.getRate().equalsIgnoreCase(ScheduleType.SIMPLE.name())) {
datasetTableTask.setStatus(TaskStatus.Stopped.name());
recore.setStatus(TaskStatus.Stopped.name());
} else {
// datasetTableTask = datasetTableTaskMapper.selectByPrimaryKey(datasetTableTask.getId());
datasetTableTask.setLastExecStatus(lastExecStatus.name());
recore.setLastExecStatus(lastExecStatus.name());
if (StringUtils.isNotEmpty(datasetTableTask.getEnd()) && datasetTableTask.getEnd().equalsIgnoreCase("1")) {
BaseGridRequest request = new BaseGridRequest();
ConditionEntity conditionEntity = new ConditionEntity();
......@@ -216,19 +228,25 @@ public class DataSetTableTaskService {
return;
}
if (dataSetTaskDTOS.get(0).getNextExecTime() == null || dataSetTaskDTOS.get(0).getNextExecTime() <= 0) {
datasetTableTask.setStatus(TaskStatus.Stopped.name());
recore.setStatus(TaskStatus.Stopped.name());
} else {
datasetTableTask.setStatus(TaskStatus.Underway.name());
recore.setStatus(TaskStatus.Underway.name());
}
} else {
datasetTableTask.setStatus(TaskStatus.Underway.name());
recore.setStatus(TaskStatus.Underway.name());
}
}
update(datasetTableTask);
DatasetTableTaskExample datasetTableTaskExample = new DatasetTableTaskExample();
datasetTableTaskExample.createCriteria().andIdEqualTo(datasetTableTask.getId());
updateByExampleSelective(recore, datasetTableTaskExample);
}
public DatasetTableTask selectByPrimaryKey(String id) {
return datasetTableTaskMapper.selectByPrimaryKey(id);
}
public void update(DatasetTableTask datasetTableTask) {
datasetTableTaskMapper.updateByPrimaryKeySelective(datasetTableTask);
public void updateByExampleSelective(DatasetTableTask datasetTableTask, DatasetTableTaskExample datasetTableTaskExample) {
datasetTableTaskMapper.updateByExampleSelective(datasetTableTask, datasetTableTaskExample);
}
public List<DatasetTableTask> list(DatasetTableTask datasetTableTask) {
......
backend/src/main/java/io/dataease/service/dataset/ExtractDataService.java
浏览文件 @ afd2a7ae
......@@ -65,6 +65,7 @@ import org.quartz.JobExecutionContext;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.io.*;
import java.util.*;
......@@ -85,12 +86,6 @@ public class ExtractDataService {
@Lazy
private DataSetTableTaskService dataSetTableTaskService;
@Resource
private DatasourceMapper datasourceMapper;
@Resource
private DatasetTableMapper datasetTableMapper;
@Resource
private DatasetTableTaskMapper datasetTableTaskMapper;
@Resource
private DatasourceService datasourceService;
@Resource
private ExtChartViewMapper extChartViewMapper;
......@@ -133,24 +128,28 @@ public class ExtractDataService {
" exit 1\n" +
"fi\n";
public synchronized boolean existSyncTask(DatasetTable datasetTable, DatasetTableTask datasetTableTask, Long startTime) {
datasetTable.setSyncStatus(JobStatus.Underway.name());
public synchronized boolean existSyncTask(String datasetTableId, String datasetTableTaskId, Long startTime) {
DatasetTable datasetTableRecord = new DatasetTable();
datasetTableRecord.setSyncStatus(JobStatus.Underway.name());
DatasetTableExample example = new DatasetTableExample();
example.createCriteria().andIdEqualTo(datasetTable.getId()).andSyncStatusNotEqualTo(JobStatus.Underway.name());
example.or(example.createCriteria().andIdEqualTo(datasetTable.getId()).andSyncStatusIsNull());
boolean existSyncTask = datasetTableMapper.updateByExampleSelective(datasetTable, example) == 0;
example.createCriteria().andIdEqualTo(datasetTableId).andSyncStatusNotEqualTo(JobStatus.Underway.name());
example.or(example.createCriteria().andIdEqualTo(datasetTableId).andSyncStatusIsNull());
boolean existSyncTask = dataSetTableService.updateByExampleSelective(datasetTableRecord, example) == 0;
if (existSyncTask) {
DatasetTableTaskLog datasetTableTaskLog = new DatasetTableTaskLog();
datasetTableTaskLog.setTaskId(datasetTableTask.getId());
datasetTableTaskLog.setTableId(datasetTable.getId());
datasetTableTaskLog.setTaskId(datasetTableTaskId);
datasetTableTaskLog.setTableId(datasetTableId);
datasetTableTaskLog.setStatus(JobStatus.Underway.name());
List<DatasetTableTaskLog> datasetTableTaskLogs = dataSetTableTaskLogService.select(datasetTableTaskLog);
return CollectionUtils.isEmpty(datasetTableTaskLogs) || !datasetTableTaskLogs.get(0).getTriggerType().equalsIgnoreCase(TriggerType.Custom.name());
} else {
datasetTableTask.setLastExecTime(startTime);
datasetTableTask.setLastExecStatus(JobStatus.Underway.name());
datasetTableTask.setStatus(TaskStatus.Exec.name());
dataSetTableTaskService.update(datasetTableTask);
DatasetTableTask record = new DatasetTableTask();
record.setLastExecTime(startTime);
record.setLastExecStatus(JobStatus.Underway.name());
record.setStatus(TaskStatus.Exec.name());
DatasetTableTaskExample datasetTableTaskExample = new DatasetTableTaskExample();
datasetTableTaskExample.createCriteria().andIdEqualTo(datasetTableTaskId);
dataSetTableTaskService.updateByExampleSelective(record, datasetTableTaskExample);
return false;
}
}
......@@ -179,7 +178,7 @@ public class ExtractDataService {
return o1.getColumnIndex().compareTo(o2.getColumnIndex());
});
DatasetTableTaskLog datasetTableTaskLog = writeDatasetTableTaskLog(datasetTableId, ops);
DatasetTableTaskLog datasetTableTaskLog = writeDatasetTableTaskLog(datasetTableId, ops);
switch (updateType) {
case all_scope: // 全量更新
try {
......@@ -195,7 +194,7 @@ public class ExtractDataService {
}
replaceTable(TableUtils.tableName(datasetTableId));
saveSuccessLog(datasetTableTaskLog);
updateTableStatus(datasetTableId, datasetTable, JobStatus.Completed, execTime);
updateTableStatus(datasetTableId, JobStatus.Completed, execTime);
if (ops.equalsIgnoreCase("替换")) {
List<DatasetTableField> oldFileds = getDatasetTableFields(datasetTable.getId());
List<DatasetTableField> toAdd = new ArrayList<>();
......@@ -228,7 +227,7 @@ public class ExtractDataService {
}
} catch (Exception e) {
saveErrorLog(datasetTableTaskLog, e);
updateTableStatus(datasetTableId, datasetTable, JobStatus.Error, null);
updateTableStatus(datasetTableId, JobStatus.Error, null);
dropDorisTable(TableUtils.tmpName(TableUtils.tableName(datasetTableId)));
} finally {
deleteFile("all_scope", datasetTableId);
......@@ -247,10 +246,10 @@ public class ExtractDataService {
extractExcelDataForSimpleMode(datasetTable, "incremental_add");
}
saveSuccessLog(datasetTableTaskLog);
updateTableStatus(datasetTableId, datasetTable, JobStatus.Completed, execTime);
updateTableStatus(datasetTableId, JobStatus.Completed, execTime);
} catch (Exception e) {
saveErrorLog(datasetTableTaskLog, e);
updateTableStatus(datasetTableId, datasetTable, JobStatus.Error, null);
updateTableStatus(datasetTableId, JobStatus.Error, null);
} finally {
deleteFile("incremental_add", datasetTableId);
deleteFile("incremental_delete", datasetTableId);
......@@ -273,7 +272,7 @@ public class ExtractDataService {
LogUtil.error("Can not find DatasetTable: " + datasetTableId);
return;
}
DatasetTableTask datasetTableTask = datasetTableTaskMapper.selectByPrimaryKey(taskId);
DatasetTableTask datasetTableTask = dataSetTableTaskService.selectByPrimaryKey(taskId);
if (datasetTableTask == null) {
return;
}
......@@ -284,19 +283,22 @@ public class ExtractDataService {
}
Long startTime = System.currentTimeMillis();
if (existSyncTask(datasetTable, datasetTableTask, startTime)) {
if (existSyncTask(datasetTable.getId(), datasetTableTask.getId(), startTime)) {
LogUtil.info("Skip synchronization task for dataset due to exist others, dataset ID : " + datasetTableId);
return;
}
DatasetTableTaskLog datasetTableTaskLog = getDatasetTableTaskLog(datasetTableId, taskId, startTime);
UpdateType updateType = UpdateType.valueOf(type);
if (context != null) {
datasetTable.setQrtzInstance(context.getFireInstanceId());
datasetTableMapper.updateByPrimaryKeySelective(datasetTable);
DatasetTable datasetTableRecord = new DatasetTable();
datasetTableRecord.setQrtzInstance(context.getFireInstanceId());
DatasetTableExample example = new DatasetTableExample();
example.createCriteria().andIdEqualTo(datasetTableId);
dataSetTableService.updateByExampleSelective(datasetTableRecord, example);
}
Datasource datasource = new Datasource();
if (StringUtils.isNotEmpty(datasetTable.getDataSourceId())) {
datasource = datasourceMapper.selectByPrimaryKey(datasetTable.getDataSourceId());
datasource = datasourceService.get(datasetTable.getDataSourceId());
} else {
datasource.setType(datasetTable.getType());
}
......@@ -337,7 +339,7 @@ public class ExtractDataService {
System.out.println(ignore.getMessage());
}
try {
updateTableStatus(datasetTableId, datasetTable, lastExecStatus, execTime);
updateTableStatus(datasetTableId, lastExecStatus, execTime);
} catch (Exception ignore) {
System.out.println(ignore.getMessage());
}
......@@ -395,7 +397,7 @@ public class ExtractDataService {
} catch (Exception ignore) {
}
try {
updateTableStatus(datasetTableId, datasetTable, lastExecStatus, execTime);
updateTableStatus(datasetTableId, lastExecStatus, execTime);
} catch (Exception ignore) {
}
}
......@@ -459,9 +461,9 @@ public class ExtractDataService {
String dataFile = null;
String script = null;
String streamLoadScript = "";
if(kettleFilesKeep){
if (kettleFilesKeep) {
streamLoadScript = shellScript;
}else {
} else {
streamLoadScript = shellScriptForDeleteFile;
}
switch (extractType) {
......@@ -586,14 +588,16 @@ public class ExtractDataService {
});
}
private void updateTableStatus(String datasetTableId, DatasetTable datasetTable, JobStatus completed, Long execTime) {
datasetTable.setSyncStatus(completed.name());
private void updateTableStatus(String datasetTableId, JobStatus jobStatus, Long execTime) {
DatasetTable datasetTableRecord = new DatasetTable();
datasetTableRecord.setId(datasetTableId);
datasetTableRecord.setSyncStatus(jobStatus.name());
if (execTime != null) {
datasetTable.setLastUpdateTime(execTime);
datasetTableRecord.setLastUpdateTime(execTime);
}
DatasetTableExample example = new DatasetTableExample();
example.createCriteria().andIdEqualTo(datasetTableId);
datasetTableMapper.updateByExampleSelective(datasetTable, example);
dataSetTableService.updateByExampleSelective(datasetTableRecord, example);
}
private void saveSuccessLog(DatasetTableTaskLog datasetTableTaskLog) {
......@@ -706,7 +710,7 @@ public class ExtractDataService {
for (ExcelSheetData sheet : excelXlsxReader.totalSheets) {
if (sheet.getExcelLable().equalsIgnoreCase(excelSheetData.getExcelLable())) {
for (List<String> dataItem : sheet.getData()) {
if(dataItem.size()>0){
if (dataItem.size() > 0) {
data.add(dataItem.toArray(new String[dataItem.size()]));
}
}
......@@ -792,9 +796,9 @@ public class ExtractDataService {
DorisConfiguration dorisConfiguration = new Gson().fromJson(dorisDatasource.getConfiguration(), DorisConfiguration.class);
String columns = columnFields + ",dataease_uuid";
String streamLoadScript = "";
if(kettleFilesKeep){
if (kettleFilesKeep) {
streamLoadScript = shellScript;
}else {
} else {
streamLoadScript = shellScriptForDeleteFile;
}
switch (extractType) {
......@@ -897,9 +901,9 @@ public class ExtractDataService {
case StarRocks:
MysqlConfiguration mysqlConfiguration = new Gson().fromJson(datasource.getConfiguration(), MysqlConfiguration.class);
dataMeta = new DatabaseMeta("db", "MYSQL", "Native", mysqlConfiguration.getHost().trim(), mysqlConfiguration.getDataBase().trim(), mysqlConfiguration.getPort().toString(), mysqlConfiguration.getUsername(), mysqlConfiguration.getPassword());
if(StringUtils.isNotEmpty(mysqlConfiguration.getExtraParams()) && mysqlConfiguration.getExtraParams().split("&").length > 0){
if (StringUtils.isNotEmpty(mysqlConfiguration.getExtraParams()) && mysqlConfiguration.getExtraParams().split("&").length > 0) {
String[] params = mysqlConfiguration.getExtraParams().split("&");
for(int i=0;i<params.length;i++){
for (int i = 0; i < params.length; i++) {
dataMeta.addExtraOption("MYSQL", params[i].split("=")[0], params[i].split("=")[1]);
}
}
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论