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

【feat】 增加数据拉取，获取new_id、更新数据、一次性插入埋点

Browse Source
This commit is contained in:
Kris 2025-08-20 11:18:31 +08:00
parent e3e1f58482
commit f951e3627e
5 changed files with 82 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/main/java/com/celnet/datadump/entity/DataLog.java
View File

@ -6,6 +6,7 @@ import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
@ -101,5 +102,13 @@ public class DataLog implements Serializable {
@ApiModelProperty(value = "状态")
private String status;
public DataLog(String requestUrl, String requestData, Date startTime, Date endTime, String requestMethod, String requestType) {
this.requestUrl = requestUrl;
this.requestData = requestData;
this.startTime = startTime;
this.endTime = endTime;
this.requestMethod = requestMethod;
this.requestType = requestType;
}
}

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

@ -83,6 +83,8 @@ public class CommonServiceImpl implements CommonService {
private MetaclassConfigService metaclassConfigService;
@Autowired
private LinkConfigService linkConfigService;
@Autowired
private DataLogService dataLogService;
@Override
public ReturnT<String> increment(SalesforceParam param) throws Exception {
@ -686,7 +688,7 @@ public class CommonServiceImpl implements CommonService {
}
// 获取数据库字段进行比对
List<String> fields = customMapper.getFields(api).stream().map(String::toUpperCase).collect(Collectors.toList());
int failCount = 0;
int failCount = 0,batch = 0;
while (true) {
try {
// 获取创建时间
@ -704,10 +706,14 @@ public class CommonServiceImpl implements CommonService {
}
log.info("query sql: {}", sql);
XxlJobLogger.log("query sql: {}", sql);
DataLog dataLog = new DataLog(api, null, new Date(), null, "数据拉取，拉取第" + batch + "批数据", "QuerySF");
QueryResult queryResult = connect.queryAll(sql);
if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
break;
}
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
SObject[] records = queryResult.getRecords();
objects = DataUtil.toJsonArray(records, dsrFields);
// 获取最大修改时间和等于该修改时间的数据id
@ -720,8 +726,12 @@ public class CommonServiceImpl implements CommonService {
.max(String::compareTo).get();
lastCreatedDate = maxDate;
}
DataLog dataLog1 = new DataLog(api, null, new Date(), null, "数据拉取，Upsert第" + batch + "批数据", "UpsertDB");
// 存储更新
saveOrUpdate(api, fields, records, objects, true);
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
count += records.length;
TimeUnit.MILLISECONDS.sleep(1);
String format = DateFormatUtils.format(lastCreatedDate, "yyyy-MM-dd HH:mm:ss");

10
src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java
View File

@ -64,6 +64,8 @@ public class DataDumpSpecialServiceImpl implements DataDumpSpecialService {
private SalesforceConnect salesforceConnect;
@Autowired
private SalesforceExecutor salesforceExecutor;
@Autowired
private DataLogService dataLogService;
@Override
public ReturnT<String> dataDumpSpecial(DataDumpSpecialParam param) throws Throwable {
@ -180,7 +182,7 @@ public class DataDumpSpecialServiceImpl implements DataDumpSpecialService {
// 判断是否存在要排除的id
salesforceParam.setMaxId(maxId);
map.put("param", salesforceParam);
DataLog dataLog = new DataLog(api, "拉取" + api + "对象数据", new Date(), null, "拉取第" + batch + "批数据", "Pull");
DataLog dataLog = new DataLog(api, null, new Date(), null, "数据拉取，拉取第" + batch + "批数据", "QuerySF");
String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderById", map);
log.info("query sql: {}", sql);
XxlJobLogger.log("query sql: {}", sql);
@ -188,11 +190,17 @@ public class DataDumpSpecialServiceImpl implements DataDumpSpecialService {
if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
break;
}
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
SObject[] records = queryResult.getRecords();
objects = DataUtil.toJsonArray(records, dsrFields);
maxId = ((JSONObject) objects.get(objects.size() - 1)).getString(Const.ID);
DataLog dataLog1 = new DataLog(api, null, new Date(), null, "数据拉取，Upsert第" + batch + "批数据", "UpsertDB");
// 存储更新
commonService.saveOrUpdate(api, fields, records, objects, true);
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
count += records.length;
TimeUnit.MILLISECONDS.sleep(1);
log.info("dump success count: {}", count);

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

@ -88,6 +88,9 @@ public class DataImportNewServiceImpl implements DataImportNewService {
@Autowired
private LinkConfigService linkConfigService;
@Autowired
private DataLogService dataLogService;
private static final String TEMP_FILE_PATH = "verify/";
static {
@ -790,6 +793,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
for (int i = 0; i < page; i++) {
DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新，查询并组装第" + z*50 + i + "批数据", "QueryDB");
int startIndex = 200 * i;
int endIndex = Math.min(startIndex + 200, size);
if (startIndex >= endIndex) {
@ -853,11 +857,13 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
accounts[j++] = account;
}
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
if (infoFlag != null && "1".equals(infoFlag.get("value"))){
printlnAccountsDetails(accounts,list);
}
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新，更新SF第" + z*50 + i + "批数据", "UpdateSF");
SaveResult[] saveResults = partnerConnection.update(accounts);
for (SaveResult saveResult : saveResults) {
if (!saveResult.getSuccess()) {
@ -869,6 +875,8 @@ public class DataImportNewServiceImpl implements DataImportNewService {
targetCount ++;
}
}
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
} catch (Throwable e) {
log.info(JSON.toJSONString(e));
@ -2888,6 +2896,8 @@ public class DataImportNewServiceImpl implements DataImportNewService {
int page = size%200 == 0 ? size/200 : (size/200) + 1;
for (int i = 0; i < page; i++) {
DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入，查询并组装" + (z*50+i) + "批数据", "QueryBD");
int startIndex = 200 * i;
int endIndex = Math.min(startIndex + 200, size);
if (startIndex >= endIndex) {
@ -2955,11 +2965,20 @@ public class DataImportNewServiceImpl implements DataImportNewService {
break;
}
}
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
if (infoFlag != null && "1".equals(infoFlag.get("value"))){
printlnAccountsDetails(accounts, list);
}
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入，插入SF第" + (z*50+i) + "批数据", "InsertSF");
SaveResult[] saveResults = partnerConnection.create(accounts);
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入，更新DB第" + (z*50+i) + "批数据", "UpdateDB");
int index = 0;
for (SaveResult saveResult : saveResults){
if (saveResult.getSuccess()) {
@ -2975,6 +2994,12 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
index++;
}
dataLog2.setEndTime(new Date());
dataLogService.save(dataLog2);
TimeUnit.MILLISECONDS.sleep(1);
}catch (InterruptedException e){
return;
} catch (Exception e) {
log.error("insertSingle error api:{},错误信息：{}", api, JSON.toJSONString(e));
}
@ -3085,6 +3110,8 @@ public class DataImportNewServiceImpl implements DataImportNewService {
for (int i = 0; i < page; i++) {
try {
DataLog dataLog = new DataLog(api, null, new Date(), null, "一次性插入，查询并组装" + (z*50+i) + "批数据", "QueryBD");
int startIndex = 200 * i;
int endIndex = Math.min(startIndex + 200, size);
if (startIndex >= endIndex) {
@ -3150,11 +3177,19 @@ public class DataImportNewServiceImpl implements DataImportNewService {
break;
}
}
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
if (infoFlag != null && "1".equals(infoFlag.get("value"))){
printlnAccountsDetails(accounts, list);
}
DataLog dataLog1 = new DataLog(api, null, new Date(), null, "一次性插入，插入SF第" + (z*50+i) + "批数据", "InsertSF");
SaveResult[] saveResults = partnerConnection.create(accounts);
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
DataLog dataLog2 = new DataLog(api, null, new Date(), null, "一次性插入，更新DB第" + (z*50+i) + "批数据", "UpdateDB");
int index = 0;
for (SaveResult saveResult : saveResults){
if (saveResult.getSuccess()) {
@ -3170,6 +3205,9 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
index++;
}
dataLog2.setEndTime(new Date());
dataLogService.save(dataLog2);
TimeUnit.MILLISECONDS.sleep(1);
}catch (InterruptedException e){
return;

14
src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java
View File

@ -315,6 +315,8 @@ public class DataImportServiceImpl implements DataImportService {
if (startIndex >= endIndex) {
break;
}
DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "获取new_id，查询并组装第" + z*50 + i + "批数据", "QueryDB");
List<Map<String, Object>> mapList = mapsList.subList(startIndex, endIndex);
int sized = mapList.size();
SObject[] accounts = new SObject[sized];
@ -398,11 +400,20 @@ public class DataImportServiceImpl implements DataImportService {
}
}
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "获取new_id，插入SF第" + z*50 + i + "批数据", "InserSF");
if (infoFlag != null && "1".equals(infoFlag.get("value"))){
printAccountsDetails(accounts, list);
}
SaveResult[] saveResults = partnerConnection.create(accounts);
log.info("sf return saveResults------"+ JSONArray.toJSONString(saveResults));
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "获取new_id，更新DB第" + z*50 + i + "批数据", "UpdateDB");
int index = 0;
for (SaveResult saveResult : saveResults){
if (saveResult.getSuccess()) {
@ -417,6 +428,9 @@ public class DataImportServiceImpl implements DataImportService {
}
index++;
}
dataLog2.setEndTime(new Date());
dataLogService.save(dataLog2);
TimeUnit.MILLISECONDS.sleep(1);
} catch (Exception e) {
log.info(JSON.toJSONString(e));