提交 7675f9c1 authored 作者: taojinlong's avatar taojinlong

fix: API 数据集增量同步

上级 264efa33
...@@ -342,22 +342,14 @@ public class ExtractDataService { ...@@ -342,22 +342,14 @@ public class ExtractDataService {
try { try {
if(datasource.getType().equalsIgnoreCase(DatasourceTypes.api.name())){ if(datasource.getType().equalsIgnoreCase(DatasourceTypes.api.name())){
extractData(datasetTable, datasource, datasetTableFields, "incremental_add", null); extractData(datasetTable, datasource, datasetTableFields, "incremental_add", null);
return; }else{
}
DatasetTableIncrementalConfig datasetTableIncrementalConfig = dataSetTableService.incrementalConfig(datasetTableId); DatasetTableIncrementalConfig datasetTableIncrementalConfig = dataSetTableService.incrementalConfig(datasetTableId);
if (datasetTableIncrementalConfig == null || StringUtils.isEmpty(datasetTableIncrementalConfig.getTableId())) {
updateTableStatus(datasetTableId, datasetTable, JobStatus.Completed, null);
return;
}
if (datasetTable.getLastUpdateTime() == null || datasetTable.getLastUpdateTime() == 0) { if (datasetTable.getLastUpdateTime() == null || datasetTable.getLastUpdateTime() == 0) {
updateTableStatus(datasetTableId, datasetTable, JobStatus.Completed, null); throw new Exception("未进行全量同步");
saveErrorLog(datasetTableId, taskId, new Exception("未进行全量同步"));
lastExecStatus = JobStatus.Error;
return;
} }
execTime = System.currentTimeMillis(); execTime = System.currentTimeMillis();
if (StringUtils.isNotEmpty(datasetTableIncrementalConfig.getIncrementalAdd()) && StringUtils.isNotEmpty(datasetTableIncrementalConfig.getIncrementalAdd().replace(" ", ""))) {// 增量添加 if (datasetTableIncrementalConfig != null && StringUtils.isNotEmpty(datasetTableIncrementalConfig.getIncrementalAdd()) && StringUtils.isNotEmpty(datasetTableIncrementalConfig.getIncrementalAdd().replace(" ", ""))) {// 增量添加
String sql = datasetTableIncrementalConfig.getIncrementalAdd().replace(lastUpdateTime, datasetTable.getLastUpdateTime().toString()) String sql = datasetTableIncrementalConfig.getIncrementalAdd().replace(lastUpdateTime, datasetTable.getLastUpdateTime().toString())
.replace(currentUpdateTime, Long.valueOf(System.currentTimeMillis()).toString()); .replace(currentUpdateTime, Long.valueOf(System.currentTimeMillis()).toString());
extractData(datasetTable, datasource, datasetTableFields, "incremental_add", sql); extractData(datasetTable, datasource, datasetTableFields, "incremental_add", sql);
...@@ -368,8 +360,8 @@ public class ExtractDataService { ...@@ -368,8 +360,8 @@ public class ExtractDataService {
.replace(currentUpdateTime, Long.valueOf(System.currentTimeMillis()).toString()); .replace(currentUpdateTime, Long.valueOf(System.currentTimeMillis()).toString());
extractData(datasetTable, datasource, datasetTableFields, "incremental_delete", sql); extractData(datasetTable, datasource, datasetTableFields, "incremental_delete", sql);
} }
}
saveSuccessLog(datasetTableTaskLog); saveSuccessLog(datasetTableTaskLog);
msg = true; msg = true;
lastExecStatus = JobStatus.Completed; lastExecStatus = JobStatus.Completed;
} catch (Exception e) { } catch (Exception e) {
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论