kris/data-dump
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【feat】数据更新失败记录及数据更新字段判断条件

Browse Source
This commit is contained in:
Kris 2025-08-29 14:30:42 +08:00
parent fef14ca0b7
commit 468aba2515
8 changed files with 325 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
.run/data-dump（亚数）.run.xml Normal file
View File

@ -0,0 +1,14 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="data-dump（亚数）" type="docker-deploy" factoryName="dockerfile" server-name="亚数Docker">
<deployment type="dockerfile">
<settings>
<option name="imageTag" value="data-dump" />
<option name="buildCliOptions" value="--platform=linux/amd64" />
<option name="containerName" value="data-dump" />
<option name="commandLineOptions" value="-p 90:80 -v /usr/local/data-dump/file:/data-dump/file -v /usr/local/data-dump/logs:/data-dump/logs --restart=always --network=host" />
<option name="sourceFilePath" value="Dockerfile" />
</settings>
</deployment>
<method v="2" />
</configuration>
</component>

14
.run/data-dump（启德）.run.xml Normal file
View File

@ -0,0 +1,14 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="data-dump（启德）" type="docker-deploy" factoryName="dockerfile" server-name="启德docker">
<deployment type="dockerfile">
<settings>
<option name="imageTag" value="data-dump" />
<option name="buildCliOptions" value="--platform=linux/amd64" />
<option name="containerName" value="data-dump" />
<option name="commandLineOptions" value="-p 90:80 -v /usr/local/data-dump/file:/data-dump/file -v /usr/local/data-dump/logs:/data-dump/logs --restart=always --network=host" />
<option name="sourceFilePath" value="Dockerfile" />
</settings>
</deployment>
<method v="2" />
</configuration>
</component>

4
src/main/java/com/celnet/datadump/entity/DataField.java
View File

@ -100,9 +100,9 @@ public class DataField implements Serializable {
/**
* 是否更新（目标库）
*/
@TableField("is_update")
@TableField("is_updateable")
@ApiModelProperty(value = "是否更新")
private Boolean isUpdate;
private Boolean isUpdateable;
/**
* 是否为空（目标库）

34
src/main/java/com/celnet/datadump/job/DataDumpNewJob.java
View File

@ -4,10 +4,7 @@ import com.alibaba.fastjson.JSON;
import com.celnet.datadump.config.SalesforceConnect;
import com.celnet.datadump.entity.DataObject;
import com.celnet.datadump.param.SalesforceParam;
import com.celnet.datadump.service.CommonService;
import com.celnet.datadump.service.DataImportBatchService;
import com.celnet.datadump.service.DataImportNewService;
import com.celnet.datadump.service.DataObjectService;
import com.celnet.datadump.service.*;
import com.celnet.datadump.util.DataUtil;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
@ -40,6 +37,9 @@ public class DataDumpNewJob {
@Autowired
private DataObjectService dataObjectService;
@Autowired
private CommonBatchService commonBatchService;
/**
* bulk批量大数据生成newSFID
@ -75,7 +75,7 @@ public class DataDumpNewJob {
*/
@XxlJob("dataUpdateBatchJob")
public ReturnT<String> dataUpdateBatchJob(String paramStr) throws Exception {
log.info("dataImportBatchJob execute start ..................");
log.info("dataUpdateBatchJob execute start ..................");
SalesforceParam param = new SalesforceParam();
try {
if (StringUtils.isNotBlank(paramStr)) {
@ -90,6 +90,30 @@ public class DataDumpNewJob {
return dataImportBatchService.immigrationUpdateBatch(param);
}
/**
* 存量任务（大批量）
*
* @param paramStr 参数json
* @return result
*/
@XxlJob("dataDumpManualBatchJob")
public ReturnT<String> dataDumpManualBatchJob(String paramStr) throws Exception {
log.info("dataDumpManualBatchJob execute start ..................");
SalesforceParam param = new SalesforceParam();
try {
if (StringUtils.isNotBlank(paramStr)) {
param = JSON.parseObject(paramStr, SalesforceParam.class);
}
} catch (Throwable throwable) {
return new ReturnT<>(500, "参数解析失败！");
}
// 参数转换
param.setBeginCreateDate(param.getBeginDate());
param.setEndCreateDate(param.getEndDate());
return commonBatchService.dumpBatch(param);
}
/**
* 返写NewId
* @param paramStr 参数json

11
src/main/java/com/celnet/datadump/service/CommonBatchService.java Normal file
View File

@ -0,0 +1,11 @@
package com.celnet.datadump.service;
import com.celnet.datadump.param.SalesforceParam;
import com.xxl.job.core.biz.model.ReturnT;
public interface CommonBatchService {
ReturnT<String> dumpBatch(SalesforceParam param) throws Exception;
}

237
src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java Normal file
View File

@ -0,0 +1,237 @@
package com.celnet.datadump.service.impl;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.celnet.datadump.config.SalesforceConnect;
import com.celnet.datadump.config.SalesforceExecutor;
import com.celnet.datadump.entity.DataBatch;
import com.celnet.datadump.entity.DataBatchHistory;
import com.celnet.datadump.entity.DataObject;
import com.celnet.datadump.entity.DataReport;
import com.celnet.datadump.global.Const;
import com.celnet.datadump.param.DataDumpParam;
import com.celnet.datadump.param.DataDumpSpecialParam;
import com.celnet.datadump.param.SalesforceParam;
import com.celnet.datadump.service.*;
import com.celnet.datadump.util.DataUtil;
import com.celnet.datadump.util.EmailUtil;
import com.celnet.datadump.util.SqlUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sforce.async.BulkConnection;
import com.sforce.soap.partner.PartnerConnection;
import com.sforce.soap.partner.QueryResult;
import com.sforce.soap.partner.sobject.SObject;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@Service
@Slf4j
public class CommonBatchServiceImpl implements CommonBatchService {
@Autowired
private SalesforceExecutor salesforceExecutor;
@Autowired
private DataObjectService dataObjectService;
@Autowired
private DataBatchService dataBatchService;
@Autowired
private CommonService commonService;
@Autowired
private SalesforceConnect salesforceConnect;
@Autowired
private DataFieldService dataFieldService;
@Override
public ReturnT<String> dumpBatch(SalesforceParam param) throws Exception {
// List<Future<?>> futures = Lists.newArrayList();
// try {
// if (StringUtils.isNotBlank(param.getApi())) {
// // 手动任务
// ReturnT<String> result = manualBatchDump(param, futures);
// if (result != null) {
// return result;
// }
// } else {
// // 自动任务
// autoDump(param, futures);
// }
// return ReturnT.SUCCESS;
// } catch (Throwable throwable) {
// salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
// log.error("dump error", throwable);
// throw throwable;
// }
return ReturnT.SUCCESS;
}
// private ReturnT<String> manualBatchDump(SalesforceParam param, List<Future<?>> futures) throws InterruptedException {
// List<String> apis = DataUtil.toIdList(param.getApi());
// String join = StringUtils.join(apis, ",");
// log.info("dump apis: {}", join);
//
// // 根据参数获取sql
// for (String api : apis) {
// DataObject update = new DataObject();
// TimeUnit.MILLISECONDS.sleep(1);
// try {
// // 判断当前表是否存在
// commonService.checkApi(api, true);
//
//// if (!hasCreatedDate(api)) {
//// DataDumpSpecialParam dataDumpSpecialParam = new DataDumpSpecialParam();
//// dataDumpSpecialParam.setApi(api);
//// Future<?> future = dataDumpSpecialService.getData(dataDumpSpecialParam, salesforceConnect.createConnect());
//// // 等待当前所有线程执行完成
//// salesforceExecutor.waitForFutures(future);
//// continue;
//// }
//
// List<SalesforceParam> salesforceParams = null;
// update.setName(api);
// update.setDataLock(1);
// dataObjectService.updateById(update);
// QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
// dbQw.eq("name", api)
// .isNull("first_sync_date");
//
// List<DataBatch> list = dataBatchService.list(dbQw);
// AtomicInteger batch = new AtomicInteger(1);
// if (CollectionUtils.isNotEmpty(list)) {
// salesforceParams = list.stream().map(t -> {
// SalesforceParam salesforceParam = param.clone();
// salesforceParam.setApi(t.getName());
// salesforceParam.setBeginCreateDate(t.getSyncStartDate());
// salesforceParam.setEndCreateDate(t.getSyncEndDate());
// salesforceParam.setBatch(batch.getAndIncrement());
// return salesforceParam;
// }).collect(Collectors.toList());
// } else {
// salesforceParams = DataUtil.splitTask(param);
// }
// // 手动任务优先执行
// for (SalesforceParam salesforceParam : salesforceParams) {
// Future<?> future = salesforceExecutor.execute(() -> {
// try {
// dumpBatchData(salesforceParam);
// } catch (Throwable throwable) {
// log.error("salesforceExecutor error", throwable);
// throw new RuntimeException(throwable);
// }
// }, salesforceParam.getBatch(), 1);
// futures.add(future);
// }
//
// // 等待当前所有线程执行完成
// salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
//
// update.setDataWork(0);
// } catch (Throwable e) {
// log.error("manualDump error", e);
// throw new RuntimeException(e);
// } finally {
// update.setName(api);
// update.setDataLock(0);
// dataObjectService.updateById(update);
// }
// }
// return null;
// }
//
//
// private void dumpBatchData(SalesforceParam param) throws Throwable {
// String api = param.getApi();
// BulkConnection bulkConnect = null;
// try {
// DataBatchHistory dataBatchHistory = new DataBatchHistory();
// dataBatchHistory.setName(api);
// dataBatchHistory.setStartDate(new Date());
// dataBatchHistory.setSyncStartDate(param.getBeginCreateDate());
// if (param.getEndCreateDate() != null) {
// dataBatchHistory.setSyncEndDate(DateUtils.addSeconds(param.getEndCreateDate(), -1));
// }
// dataBatchHistory.setBatch(param.getBatch());
// if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null) {
// log.info("NO.{} dump {}, date: {} ~ {} start", param.getBatch(),
// param.getApi(),
// DateFormatUtils.format(param.getBeginCreateDate(), "yyyy-MM-dd HH:mm:ss"),
// DateFormatUtils.format(param.getEndCreateDate(), "yyyy-MM-dd HH:mm:ss"));
// }
// bulkConnect = salesforceConnect.createBulkConnect();
// // 存在isDeleted 只查询IsDeleted为false的
// if (dataFieldService.hasDeleted(param.getApi())) {
// param.setIsDeleted(false);
// } else {
// // 不存在 过滤
// param.setIsDeleted(null);
// }
// // 若count数量过多 可能导致超时出不来结果 对该任务做进一步拆分
// int failCount = 0;
// boolean isSuccess = false;
// while (failCount <= Const.MAX_FAIL_COUNT) {
// try {
// // 不能 count Id
// if (!"FeedItem".equals(api) && !"ContentFolderMember ".equals(api)){
// dataBatchHistory.setSfNum(commonService.countSfNum(connect, param));
// }
// isSuccess = true;
// break;
// } catch (Throwable throwable) {
// failCount++;
// }
// }
// // 不成功 做任务拆分
// if (!isSuccess) {
// if (splitTask(param)) {
// return connect;
// }
// }
// getAllSfData(param, connect, dataReport);
// updateDataBatch(param, dataBatchHistory);
// } catch (Throwable throwable) {
// log.error("dataDumpJob error api:{}", api, throwable);
// String type = param.getType() == 1 ? "存量" : "增量";
// String format = String.format("%s数据迁移 error, api name: %s, \nparam: %s, \ncause:\n%s", type, api, JSON.toJSONString(param, DataDumpParam.getFilter()), throwable);
// EmailUtil.send("DataDump ERROR", format);
// throw throwable;
// }
// return connect;
// }
//
// /**
// * 统计salesforce数据量
// *
// * @param connect connect
// * @param param 参数
// * @return sf统计数量
// */
// public Integer countSfNum(PartnerConnection connect, SalesforceParam param) throws Exception {
// Map<String, Object> map = Maps.newHashMap();
// map.put("param", param);
// String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.count", map);
// log.info("count sql: {}", sql);
// XxlJobLogger.log("count sql: {}", sql);
// QueryResult queryResult = connect.queryAll(sql);
// SObject record = queryResult.getRecords()[0];
// Integer num = (Integer) record.getField("num");
// log.info("count sf num: {}", num);
// XxlJobLogger.log("count sf num: {}", num);
// return num;
// }
}

11
src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java
View File

@ -579,7 +579,9 @@ public class CommonServiceImpl implements CommonService {
param.setIsDeleted(null);
}
dataBatchHistory.setSfNum(countSfNum(connect, param));
if (!"FeedItem".equals(api) && !"ContentFolderMember ".equals(api)) {
dataBatchHistory.setSfNum(countSfNum(connect, param));
}
getAllSfData(param, connect, dataReport);
@ -917,7 +919,10 @@ public class CommonServiceImpl implements CommonService {
maps.add(paramMap2);
}
Map<String, Object> paramMap3 = Maps.newHashMap();
paramMap3.put("key", "is_update");
paramMap3.put("value", 0);
maps.add(paramMap3);
if (existsIds.contains(id)) {
customMapper.updateById(api, maps, id);
@ -999,7 +1004,7 @@ public class CommonServiceImpl implements CommonService {
dataField.setField(field.getName());
dataField.setName(field.getLabel());
dataField.setIsCreateable(field.getCreateable());
dataField.setIsUpdate(field.getUpdateable());
dataField.setIsUpdateable(field.getUpdateable());
dataField.setIsNillable(field.getNillable());
dataField.setIsDefaultedOnCreate(field.getDefaultedOnCreate());
String join = null;

20
src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java
View File

@ -738,20 +738,20 @@ public class DataImportNewServiceImpl implements DataImportNewService {
if (1 == param.getType()) {
if (api.endsWith("Share")){
sql = "where RowCause = 'Manual' and is_update != 1 and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
sql2 = "RowCause = 'Manual' and is_update != 1 and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit ";
sql = "where RowCause = 'Manual' and is_update = 0 and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
sql2 = "RowCause = 'Manual' and is_update = 0 and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit ";
}else {
sql = "where new_id is not null and is_update != 1 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
sql2 = "new_id is not null and is_update != 1 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit ";
sql = "where new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
sql2 = "new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit ";
}
}else {
String updateDateField = dataFieldService.returnUpdateDateField(api);
if (api.endsWith("Share")){
sql = "where RowCause = 'Manual' and is_update != 1 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' ";
sql2 = "RowCause = 'Manual' and is_update != 1 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit ";
sql = "where RowCause = 'Manual' and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' ";
sql2 = "RowCause = 'Manual' and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit ";
}else {
sql = "where new_id is not null and and is_update != 1 "+updateDateField+" >= '" + beginDateStr + "' ";
sql2 = "new_id is not null and is_update != 1 and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit ";
sql = "where new_id is not null and and is_update = 0 "+updateDateField+" >= '" + beginDateStr + "' ";
sql2 = "new_id is not null and is_update = 0 and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit ";
}
}
//表内数据总量
@ -818,7 +818,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
//根据旧sfid查找引用对象新sfid
if (field.equals("Id")) {
account.setId(String.valueOf(map.get("new_id")));
} else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdate() != null && !dataField.getIsUpdate())) {
} else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdateable() != null && !dataField.getIsUpdateable())) {
continue;
} else if ("reference".equals(dataField.getSfType()) && map.get(dataField.getField()) != null) {
@ -1036,7 +1036,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
//根据旧sfid查找引用对象新sfid
if (field.equals("Id")) {
account.setId(String.valueOf(map.get("new_id")));
} else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdate() != null && !dataField.getIsUpdate())) {
} else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdateable() != null && !dataField.getIsUpdateable())) {
continue;
} else if ("reference".equals(dataField.getSfType()) && map.get(dataField.getField()) != null) {